SpringBoot集成kafka

在Spring Boot项目中集成Apache Kafka主要涉及以下几个步骤：

1. **添加依赖**:
首先，在你的`pom.xml`或`build.gradle`文件中添加Kafka和Spring Cloud Stream的依赖。对于Maven：

   <dependency>
       <groupId>org.springframework.cloud</groupId>
       <artifactId>spring-cloud-stream-binder-kafka</artifactId>
   </dependency>
   <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka-clients</artifactId>
   </dependency>

或者Gradle：

   implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka'
   implementation 'org.apache.kafka:kafka-clients'

2. **配置应用**:
在`application.yml`或`application.properties`中配置Kafka的相关信息，如bootstrap servers（Kafka集群地址）和其他连接选项。

   spring:
     cloud:
       stream:
         bindings:
           inputChannel:
             destination: myTopic
             group: myConsumerGroup
           output:
             binder: kafka
             destination: myOutputTopic
   kafka:
     consumer:
       bootstrap-servers: localhost:9092
       auto-offset-reset: earliest
     producer:
       bootstrap-servers: localhost:9092

3. **创建绑定和处理器**:
创建一个`@StreamListener`注解的方法来处理从Kafka接收的消息。例如：

   @Bean
   public KStreamProcessor processor(KafkaStreams streams) {
      return streams.process("myProcessorName", Consumed.with(Serdes.String(), Serdes.String()),
          Produced.with(Serdes.String(), Serdes.String()));
   }

   @StreamListener(Processor.INPUT)
   public void handle(String message, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
      System.out.println("Received message from partition " + partition + ": " + message);
   }

4. **启动消息生产者/消费者**:
如果你需要自己创建消息生产者，可以使用`KafkaTemplate`。如果只需要消费，可以在启动时自动注册消费者组。

5. **运行应用**:
启动Spring Boot应用，它将会连接到Kafka并开始监听指定的主题。