📌  相关文章
📜  春季启动-Apache Kafka

📅  最后修改于: 2020-11-11 05:53:09             🧑  作者: Mango

Apache Kafka是一个开源项目，用于基于容错消息传递系统发布和订阅消息。它是快速，可扩展的，并且可以按设计进行分发。如果您是Kafka的初学者，或者想对它有所了解，请参阅此链接-www.tutorialspoint.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.tutorialspoint.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;

@Configuration
public 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对象并产生消息，如图所示。

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

消费信息

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

package com.tutorialspoint.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
@Configuration
public 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 = "tutorialspoint", 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.tutorialspoint.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;

@SpringBootApplication
public class KafkaDemoApplication implements ApplicationRunner {
   @Autowired
   private KafkaTemplate kafkaTemplate;

   public void sendMessage(String msg) {
      kafkaTemplate.send("tutorialspoint", msg);
   }
   public static void main(String[] args) {
      SpringApplication.run(KafkaDemoApplication.class, args);
   }
   @KafkaListener(topics = "tutorialspoint", 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.tutorialspoint
   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.tutorialspoint'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = 1.8

repositories {
   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')
}

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

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

mvn clean install

在“ BUILD SUCCESS”之后，您可以在目标目录下找到JAR文件。

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

gradle clean build

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

使用此处给定的命令运行JAR文件-

java –jar

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