kafka使用springboot

Kafka的Spring Boot集成是通过Spring Kafka项目实现的。要在Spring Boot中使用Kafka，首先需要在项目的pom.xml文件中引入spring-kafka依赖。你可以在pom.xml文件中添加以下代码来引入该依赖：

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

接下来，你需要在应用程序的配置文件中配置Kafka的相关属性。你可以参考官方文档或其他教程来学习如何配置Kafka的属性。

当你完成了依赖和配置的设置后，就可以在你的Spring Boot应用程序中使用Kafka了。你可以使用注解`@KafkaListener`来监听指定的topic，并在消费消息时执行相应的操作。在`@KafkaListener`注解中，你可以通过设置`topics`属性来指定要监听的topic的名称。在监听方法中，你可以使用`ConsumerRecord`对象来获取消息内容，通过注解`@Header(KafkaHeaders.RECEIVED_TOPIC)`来获取topic的名称，使用`Acknowledgment`对象来手动确认消息已被消费。

下面是一个简单的使用Spring Boot和Kafka的示例代码：

@KafkaListener(topics = "my_topic_name")
public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
    log.info("消费消息: " + record.topic() + " ---- " + record.partition() + " ---- " + record.value());
    System.out.println(JSON.parseObject(record.value().toString(), KafkaEvent.class));
    ack.acknowledge();
}

以上代码中，`@KafkaListener`注解指定了要监听的topic为"my_topic_name"。在`onMessage`方法中，通过`ConsumerRecord`对象获取消息内容，并通过`log`和`System.out.println`打印消息内容。最后，使用`ack.acknowledge()`方法手动确认消息已被消费。