springboot使用kafka（spring boot kafka）

SpringBoot 集成 Kafka

简介

Apache Kafka 是一个分布式流处理平台，主要用于构建实时数据管道、流分析和事件驱动的应用程序。SpringBoot 是一个用于简化 Spring 应用开发的框架。通过整合 Kafka，你可以轻松地将 Kafka 的功能整合到你的 SpringBoot 应用程序中。

集成 Kafka

要将 Kafka 集成到 SpringBoot 应用中，你需要执行以下步骤：

1. 依赖引入

在你的项目中添加以下依赖：```xml org.springframework.kafkaspring-kafka2.7.2 ```

2. 配置 Kafka 属性

在 `application.properties` 中配置 Kafka 属性：```properties # Kafka 服务器地址 spring.kafka.bootstrap-servers=localhost:9092 # 消费组 ID spring.kafka.consumer.group-id=my-group # 偏移提交频率 spring.kafka.consumer.auto-offset-reset=earliest # 生产者重试次数 spring.kafka.producer.retries=3 ```

3. 创建 Kafka 接收器

使用 `@KafkaListener` 注解来创建 Kafka 接收器，以处理接收到的消息：```java @KafkaListener(topics = "my-topic") public void listen(String message) {System.out.println("Received message: " + message); } ```

4. 创建 Kafka 发送器

使用 `KafkaTemplate` 来发送消息到 Kafka 主题：```java @Autowired private KafkaTemplate kafkaTemplate;public void sendMessage(String message) {kafkaTemplate.send("my-topic", message); } ```

高级功能

SpringBoot 集成 Kafka 还提供了高级功能，包括：

流处理：

使用 Spring Cloud Stream 来构建流式数据处理管道。

事务：

管理事务性消息生产和消费。

记录过滤器：

过滤不符合特定标准的消息。

错误处理：

自定义错误处理策略。

示例

以下是一个完整的 SpringBoot 集成 Kafka 的示例：```java @SpringBootApplication public class KafkaExampleApplication {public static void main(String[] args) {SpringApplication.run(KafkaExampleApplication.class, args);} }@KafkaListener(topics = "my-topic") public class KafkaConsumer {@Autowiredprivate KafkaTemplate kafkaTemplate;public void listen(String message) {System.out.println("Received message: " + message);} }@RestController public class KafkaProducerController {@PostMapping("/send")public void sendMessage(@RequestBody String message) {kafkaTemplate.send("my-topic", message);} } ```使用以下命令启动应用程序：```bash mvn spring-boot:run ```