📌  相关文章
📜  Spring Boot Kafka 消费者示例

📅  最后修改于: 2022-05-13 01:55:10.840000             🧑  作者: Mango

Spring Boot Kafka 消费者示例

Spring Boot 是Java编程语言中最流行和最常用的框架之一。它是一个基于微服务的框架，使用 Spring Boot 制作可用于生产的应用程序需要的时间非常少。 Spring Boot 可以轻松创建独立的、生产级的基于 Spring 的应用程序，您可以“直接运行”。所以下面列出了 Spring boot 的一些主要特性。

  • 创建独立的 Spring 应用程序
  • 直接嵌入 Tomcat、Jetty 或 Undertow。
  • 提供“入门”依赖项以简化构建配置。
  • 尽可能自动配置 Spring 和 3rd 方库。
  • 提供生产就绪功能，例如运行状况检查、指标和外部化配置。
  • 几乎不需要代码生成，也不需要 XML 配置。

Apache Kafka是一个发布-订阅消息系统。消息传递系统允许您在进程、应用程序和服务器之间发送消息。从广义上讲，Apache Kafka 是可以定义和进一步处理主题（一个主题可能是一个类别）的软件。应用程序可以连接到该系统并将消息传输到主题上。消息可以包含任何类型的信息，来自您个人博客上的任何事件，也可以是非常简单的文本消息，会触发任何其他事件。在这里，我们将讨论如何使用来自 Kafka 主题的消息，并使用 Spring Boot 在控制台中显示它们，其中 Kafka 是先决条件。

例子：

第 1 步：转到此链接并创建一个 Spring Boot 项目。将“ Spring for Apache Kafka ”依赖项添加到您的 Spring Boot 项目中。

第 2 步：创建一个名为KafkaConfig的配置文件。下面是KafkaConfig 的代码。 Java文件。

Java
// Java Program to Illustrate Kafka Configuration
  
package com.amiya.kafka.apachekafkaconsumer.config;
  
// Importing required classes
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;
  
// Annotations
@EnableKafka
@Configuration
  
// Class
public class KafkaConfig {
  
    @Bean
    public ConsumerFactory consumerFactory()
    {
  
        // Creating a Map of string-object pairs
        Map config = new HashMap<>();
  
        // Adding the Configuration
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                   "127.0.0.1:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG,
                   "group_id");
        config.put(
            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class);
        config.put(
            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class);
  
        return new DefaultKafkaConsumerFactory<>(config);
    }
  
    // Creating a Listener
    public ConcurrentKafkaListenerContainerFactory
    concurrentKafkaListenerContainerFactory()
    {
        ConcurrentKafkaListenerContainerFactory<
            String, String> factory
            = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

Java
// Java Program to Illustrate Kafka Consumer
  
package com.amiya.kafka.apachekafkaconsumer.consumer;
  
// Importing required classes
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
  
@Component
  
// Class
public class KafkaConsumer {
  
    @KafkaListener(topics = "NewTopic",
                   groupId = "group_id")
  
    // Method
    public void
    consume(String message)
    {
        // Print statement
        System.out.println("message = " + message);
    }
}

第 3 步：创建一个名为KafkaConsumer的消费者文件

Java

// Java Program to Illustrate Kafka Consumer
  
package com.amiya.kafka.apachekafkaconsumer.consumer;
  
// Importing required classes
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
  
@Component
  
// Class
public class KafkaConsumer {
  
    @KafkaListener(topics = "NewTopic",
                   groupId = "group_id")
  
    // Method
    public void
    consume(String message)
    {
        // Print statement
        System.out.println("message = " + message);
    }
}

第 4 步：现在我们必须执行以下操作才能使用 Spring Boot 使用来自 Kafka 主题的消息

  • 运行 Apache Zookeeper 服务器
  • 运行 Apache Kafka 服务器
  • 从 Kafka 主题发送消息

使用此命令运行 Apache Zookeeper 服务器

C:\kafka>.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

同样，使用此命令运行 Apache Kafka 服务器

C:\kafka>.\bin\windows\kafka-server-start.bat .\config\server.properties

运行以下命令从 Kafka 主题发送消息

C:\kafka>.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic NewTopic

第 5 步：现在运行您的 Spring Boot 应用程序。确保您已更改application.properties文件中的端口号

server.port=8081

让我们在 ApacheKafkaConsumerApplication 文件中运行 Spring Boot 应用程序

输出：在输出中，您可以看到当您从 Kafka 主题发送消息时，它会实时显示在控制台上。

输出