Springboot集成kafka消费指定的topic

要在Spring Boot应用程序中集成Kafka消费者，您需要完成以下步骤：

1. 引入Kafka客户端依赖项

在Maven或Gradle构建文件中添加Kafka客户端依赖项。例如，在Maven中添加以下依赖项：

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

2. 配置Kafka消费者

在Spring Boot应用程序中，您可以使用application.properties或application.yml文件配置Kafka消费者。以下是一个示例application.yml文件：

spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

这些属性指定了Kafka消费者连接到的Kafka服务器，消费者所属的消费组以及如何处理消费者从Kafka服务器读取的消息。

3. 创建Kafka消费者

在Spring Boot应用程序中，您可以使用Spring Kafka提供的KafkaTemplate类来创建Kafka消费者。以下是一个示例代码片段：

@Service
public class KafkaConsumerService {
 
    @KafkaListener(topics = "my-topic")
    public void listen(String message) {
        // 处理接收到的消息
    }
}

使用@KafkaListener注释将这个方法标记为Kafka消费者。在这个示例中，消费者订阅名为“my-topic”的Kafka主题，并在接收到消息时调用listen()方法。

4. 运行应用程序

现在，您可以运行Spring Boot应用程序并开始使用Kafka消费者。当有新的消息发布到“my-topic”主题时，Kafka消费者将自动从Kafka服务器读取消息并将其发送到KafkaConsumerService类的listen()方法中进行处理。

注意：在运行应用程序之前，请确保Kafka服务器正在运行，并且“my-topic”主题已经创建。