Spring Boot - Apache Kafka

Spring Boot Apache Kafka - 从简单和简单的步骤学习Spring Boot，从基本到高级概念，包括简介，快速入门，Bootstrapping，Tomcat部署，构建系统，代码结构，Spring Bean和依赖注入，Runners，应用程序属性，日志记录等示例，构建RESTful Web服务，异常处理，拦截器，Servlet过滤器，Tomcat端口号，Rest模板，文件处理，服务组件，Thymeleaf，使用RESTful Web服务，CORS支持，国际化，调度，启用HTTPS，Eureka服务器，服务注册Eureka，Zuul代理服务器和路由，Spring云配置服务器，Spring云配置客户端，执行器，管理服务器，管理客户端，启用Swagger2，创建Docker镜像，跟踪微服务日志，Flyway数据库，发送电子邮件，Hystrix，Web套接字，批处理服务，Apache Kafka的Spring，Twilio，单元测试用例，静止控制器单元测试，数据库处理，保护Web A.应用程序，带有JWT的OAuth2，Google云端平台，Google OAuth2登录。

Apache Kafka是一个开源项目，用于根据容错消息传递系统发布和订阅消息.它设计快速，可扩展和分布.如果您是Kafka的初学者，或想要更好地了解它，请参阅此链接 : www.it1352.com/apache_kafka/

在本章中，我们将看看如何在Spring Boot应用程序中实现Apache Kafka.

首先，我们需要在构建配置文件中添加Spring Kafka依赖项.

Maven用户可以在pom.xml文件中添加以下依赖项.

   org.springframework.kafka   spring-kafka   2.1.0.RELEASE

Gradle用户可以在build.gradle文件中添加以下依赖项.

compile group: 'org.springframework.kafka', name: 'spring-kafka', version: '2.1.0.RELEASE'

制作消息

要向Apache Kafka生成消息，我们需要为Producer配置定义Configuration类，如下所示 :

package com.it1352.kafkademo; import java.util.HashMap;import java.util.Map;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.serialization.StringSerializer;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.kafka.core.DefaultKafkaProducerFactory;import org.springframework.kafka.core.KafkaTemplate;import org.springframework.kafka.core.ProducerFactory;@Configurationpublic class KafkaProducerConfig {   @Bean   public ProducerFactory producerFactory() {      Map configProps = new HashMap<>();      configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");      configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);      configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);      return new DefaultKafkaProducerFactory<>(configProps);   }   @Bean   public KafkaTemplate kafkaTemplate() {      return new KafkaTemplate<>(producerFactory());   }}

要发布消息，请自动连接Kafka Template对象并生成消息，如图所示.

@Autowiredprivate KafkaTemplate kafkaTemplate; public void sendMessage(String msg) {   kafkaTemplate.send(topicName, msg);}

使用消息

要使用消息，我们需要编写Consumer配置类文件如下所示.

package com.it1352.kafkademo; import java.util.HashMap;import java.util.Map;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.common.serialization.StringDeserializer;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.kafka.annotation.EnableKafka;import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;import org.springframework.kafka.core.ConsumerFactory;import org.springframework.kafka.core.DefaultKafkaConsumerFactory;@EnableKafka@Configurationpublic class KafkaConsumerConfig {   @Bean   public ConsumerFactory consumerFactory() {      Map props = new HashMap<>();      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:2181");      props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);      return new DefaultKafkaConsumerFactory<>(props);   }   @Bean   public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {      ConcurrentKafkaListenerContainerFactory       factory = new ConcurrentKafkaListenerContainerFactory<>();      factory.setConsumerFactory(consumerFactory());      return factory;   }}

接下来，编写一个监听器来收听消息.

@KafkaListener(topics ="it1352"，groupId ="group-id")public void listen(String message) {   System.out.println("Received Messasge in group - group-id: " + message);

让我们从主Spring Boot应用程序类文件中调用ApplicationRunner类run方法的sendMessage()方法，并使用来自相同的类文件.

您的主要Spring Boot应用程序类文件代码在下面和下面给出;

package com.it1352.kafkademo; import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.ApplicationArguments;import org.springframework.boot.ApplicationRunner;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.kafka.annotation.KafkaListener;import org.springframework.kafka.core.KafkaTemplate;@SpringBootApplicationpublic class KafkaDemoApplication implements ApplicationRunner {   @Autowired   private KafkaTemplate kafkaTemplate;   public void sendMessage(String msg) {      kafkaTemplate.send("it1352", msg);   }   public static void main(String[] args) {      SpringApplication.run(KafkaDemoApplication.class, args);   }   @KafkaListener(topics = "it1352", groupId = "group-id")   public void listen(String message) {      System.out.println("Received Messasge in group - group-id: " + message);   }   @Override   public void run(ApplicationArguments args) throws Exception {      sendMessage("Hi Welcome to Spring For Apache Kafka");   }}

完整构建配置文件的代码如下所示.

Maven - pom.xml

      4.0.0   com.IT屋   kafka-demo   0.0.1-SNAPSHOT   jar   kafka-demo   Demo project for Spring Boot         org.springframework.boot      spring-boot-starter-parent      1.5.9.RELEASE                   UTF-8      UTF-8      1.8                     org.springframework.boot         spring-boot-starter                     org.springframework.kafka         spring-kafka         2.1.0.RELEASE                     org.springframework.boot         spring-boot-starter-test         test                                       org.springframework.boot            spring-boot-maven-plugin

Gradle - build.gradle

buildscript {   ext {      springBootVersion = '1.5.9.RELEASE'   }   repositories {      mavenCentral()   }   dependencies {      classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")   }}apply plugin: 'java'apply plugin: 'eclipse'apply plugin: 'org.springframework.boot'group = 'com.it1352'version = '0.0.1-SNAPSHOT'sourceCompatibility = 1.8repositories {   mavenCentral()}dependencies {   compile('org.springframework.boot:spring-boot-starter')   compile group: 'org.springframework.kafka', name: 'spring-kafka', version: '2.1.0.RELEASE'   testCompile('org.springframework.boot:spring-boot-starter-test')}

现在，创建一个可执行的JA R文件，并使用下面的Maven或Gradle命令运行Spring Boot应用程序，如下所示 :

对于Maven，使用如下所示的命令 :

mvn clean install

在"BUILD SUCCESS"之后，您可以在目标目录下找到JAR文件.

对于Gradle，使用如下所示的命令 :

gradle clean build

在"BUILD SUCCESSFUL"之后，您可以在build/libs目录下找到JAR文件.

使用运行JAR文件这里给出的命令 :

java -jar

您可以在控制台窗口中看到输出.