SpringBoot3.x整合Kinfe4j

Spring Boot 3.x整合Apache Kafka通常涉及以下几个步骤：

1. 添加依赖：首先，在Spring Boot项目的`pom.xml`文件中添加Kafka和Spring Cloud Stream的相关依赖。例如：

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

2. 配置应用：在`application.yml`或`application.properties`中配置Kafka连接信息，如bootstrap servers地址、群组ID等。

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-consumer-group
      enable-auto-commit: true
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

3. 创建消息消费者：创建一个Spring Bean，它可以消费Kafka主题的消息。这可以是一个`KafkaTemplate`或一个实现了`MessageListener`接口的类。

@Service
public class MyConsumer {
    @KafkaListener(topics = "my-topic")
    public void consume(String message) {
        System.out.println("Received message: " + message);
    }
}

4. 生产者发送消息：同样地，可以创建一个Spring Bean来发布消息到Kafka主题。

@Service
public class MyProducer {
    private final KafkaTemplate<String, String> template;

    @Autowired
    public MyProducer(KafkaTemplate<String, String> template) {
        this.template = template;
    }

    public void sendMessage(String message) {
        template.send("my-topic", message);
    }
}

5. 使用Spring Cloud Stream：如果需要更高级的功能，比如将消息发送到多种目的地（如RabbitMQ、HTTP等），则可以利用Spring Cloud Stream的声明式绑定功能。