Spring Boot教程之五十六：用 Apache Kafka 消费 JSON 消息

Spring Boot | 如何使用 Apache Kafka 消费 JSON 消息

Apache Kafka 是一个流处理系统，可让您在进程、应用程序和服务器之间发送消息。在本文中，我们将了解如何使用 Apache Kafka 在 Spring Boot 应用程序的控制台上发布 JSON 消息。

为了了解如何创建 Spring Boot 项目，请参阅本文。

工作步骤 

步骤 1：

转到Spring 初始化程序并创建具有以下依赖项的启动项目： 
Spring for Apache Kafka

步骤 2：

在 IDE 中打开项目并同步依赖项。在本文中，我们将创建一个学生模型，我们将在其中发布学生详细信息。因此，创建一个模型类Student。添加数据成员并创建构造函数并重写toString方法以查看 JSON 格式的消息。以下是学生类的实现：

• 学生模型

// Java program to implement a // student class   // Creating a student class public class Student {       // Data members of the class     int id;     String firstName;     String lastName;       // Constructor of the student     // Class     public Student()     {     }       // Parameterized constructor of     // the student class     public Student(int id, String firstName,                    String lastName)     {         this.id = id;         this.firstName = firstName;         this.lastName = lastName;     }       @Override     public String toString()     {         return "Student{"             + "id = " + id             + ", firstName = '" + firstName + "'"             + ", lastName = '" + lastName + "'"             + "}";     } }

步骤 3：

创建一个新的类Config并添加注释@Configuration和@EnableKafka。现在使用 Student 类对象创建 Bean ConsumerFactory和ConcurrentKafkaListenerContainerFactory 。

• 配置类

@EnableKafka @Configuration public class Config {       // Function to establish a connection     // between Spring application     // and Kafka server     @Bean     public ConsumerFactory<String, Student>     studentConsumer()     {           // HashMap to store the configurations         Map<String, Object> map             = new HashMap<>();           // put the host IP in the map         map.put(ConsumerConfig                     .BOOTSTRAP_SERVERS_CONFIG,                 "127.0.0.1:9092");           // put the group ID of consumer in the map         map.put(ConsumerConfig                     .GROUP_ID_CONFIG,                 "id");         map.put(ConsumerConfig                     .KEY_DESERIALIZER_CLASS_CONFIG,                 StringDeserializer.class);         map.put(ConsumerConfig                     .VALUE_DESERIALIZER_CLASS_CONFIG,                 JsonDeserializer.class);           // return message in JSON formate         return new DefaultKafkaConsumerFactory<>(             map, new StringDeserializer(),             new JsonDeserializer<>(Student.class));     }       @Bean     public ConcurrentKafkaListenerContainerFactory<String,                                                    Student>     studentListner()     {         ConcurrentKafkaListenerContainerFactory<String,                                                 Student>             factory             = new ConcurrentKafkaListenerContainerFactory<>();         factory.setConsumerFactory(studentConsumer());         return factory;     } }

步骤 4：

创建一个带有@Service注释的KafkaService类。此类将包含用于在控制台上发布消息的侦听器方法。 

• KafkaService 类

@Service public class KafkaService {       // Annotation required to listen     // the message from Kafka server     @KafkaListener(topics = "JsonTopic",                    groupId = "id", containerFactory                                    = "studentListner")     public void     publish(Student student)     {         System.out.println("New Entry: "                            + student);     } }

步骤 5：

启动 zookeeper 和 Kafka 服务器。现在我们需要创建一个名为JsonTopic的新主题。为此，打开一个新的命令提示符窗口并将目录更改为 Kafka 目录。

步骤6：

现在使用下面给出的命令创建一个新主题： 

bin/Kafka-topics.sh –create –zookeeper localhost:2181 –replication-factor 1 –partitions 1 –topic topic_name // 用于 mac 和 linux
.\bin\windows\Kafka-topics.bat –create –zookeeper localhost:2181 –replication-factor 1 –partitions 1 –topic topic_name // 用于 windows 

步骤 7：

现在运行 Kafka 生产者控制台，使用以下命令： 

bin/Kafka-console-producer.sh –broker-list localhost:9092 –topic Kafka_Example // 适用于 mac 和 linux
.\bin\windows\Kafka-console-producer.bat –broker-list localhost:9092 –topic Kafka_Example // 适用于 windows 

步骤 8：

运行应用程序并在 Kafka 生产器上输入消息并按回车键。