Spring Boot - Apache Kafka - 几微学院 - 几微助手 | 微信商城 | 小程序商城

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

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

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

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

<dependency>   <groupId>org.springframework.kafka</groupId>   <artifactId>spring-kafka</artifactId>   <version>2.1.0.RELEASE</version></dependency>

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<String, String> producerFactory() {      Map<String, Object> 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<String, String> kafkaTemplate() {      return new KafkaTemplate<>(producerFactory());   }}

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

@Autowiredprivate KafkaTemplate<String, String> 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<String, String> consumerFactory() {      Map<String, Object> 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<String, String> kafkaListenerContainerFactory() {      ConcurrentKafkaListenerContainerFactory<String, String>       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<String, String> 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

<?xml version = "1.0" encoding = "UTF-8"?><project xmlns = "http://maven.apache.org/POM/4.0.0"    xmlns:xsi = "http://www.w3.org/2001/XMLSchema-instance"   xsi:schemaLocation = "http://maven.apache.org/POM/4.0.0    http://maven.apache.org/xsd/maven-4.0.0.xsd">      <modelVersion>4.0.0</modelVersion>   <groupId>com.IT屋</groupId>   <artifactId>kafka-demo</artifactId>   <version>0.0.1-SNAPSHOT</version>   <packaging>jar</packaging>   <name>kafka-demo</name>   <description>Demo project for Spring Boot</description>   <parent>      <groupId>org.springframework.boot</groupId>      <artifactId>spring-boot-starter-parent</artifactId>      <version>1.5.9.RELEASE</version>      <relativePath /> <!-- lookup parent from repository -->   </parent>   <properties>      <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>      <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>      <java.version>1.8</java.version>   </properties>   <dependencies>      <dependency>         <groupId>org.springframework.boot</groupId>         <artifactId>spring-boot-starter</artifactId>      </dependency>      <dependency>         <groupId>org.springframework.kafka</groupId>         <artifactId>spring-kafka</artifactId>         <version>2.1.0.RELEASE</version>      </dependency>      <dependency>         <groupId>org.springframework.boot</groupId>         <artifactId>spring-boot-starter-test</artifactId>         <scope>test</scope>      </dependency>   </dependencies>   <build>      <plugins>         <plugin>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-maven-plugin</artifactId>         </plugin>      </plugins>   </build>   </project>

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，使用如下所示的命令 :