springboot2.6整合kafka

在Spring Boot 2.6中，可以使用Spring Kafka来简化与Kafka的交互。以下是整合Kafka的步骤：

1. 添加依赖

在`pom.xml`文件中添加以下依赖：

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

2. 配置Kafka

在`application.properties`或`application.yml`文件中添加以下Kafka配置：

spring.kafka.bootstrap-servers=localhost:9092 # Kafka服务器地址
spring.kafka.consumer.group-id=my-group # 消费者组ID
spring.kafka.consumer.auto-offset-reset=earliest # 自动重置偏移量
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer # 键序列化器
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer # 值序列化器

3. 创建生产者

在Spring Boot中创建Kafka生产者非常简单，只需使用`KafkaTemplate`即可。以下是示例代码：

@Service
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

4. 创建消费者

在Spring Boot中创建Kafka消费者也非常简单，只需使用`@KafkaListener`注解即可。以下是示例代码：

@Service
public class KafkaConsumer {

    @KafkaListener(topics = "my-topic", groupId = "my-group")
    public void receiveMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

在上面的示例中，`@KafkaListener`注解将会监听`my-topic`主题，并且使用`my-group`作为消费者组ID。

5. 测试

现在，可以通过向Kafka生产者发送消息来测试整合是否成功。以下是示例代码：

@RestController
public class TestController {

    @Autowired
    private KafkaProducer kafkaProducer;

    @GetMapping("/send")
    public String sendMessage() {
        kafkaProducer.sendMessage("my-topic", "Hello, Kafka!");
        return "Message sent!";
    }
}

启动Spring Boot应用程序，并访问`/send`端点，应该会在控制台上看到类似以下的输出：

Received message: Hello, Kafka!

这意味着消息已经成功地从Kafka生产者发送到Kafka消费者。

以上就是在Spring Boot 2.6中整合Kafka的步骤。