Spring Boot集成Apache Kafka的实战指南
Apache Kafka 是一个分布式流处理平台，广泛用于构建实时数据管道、日志聚合系统和事件溯源架构。Spring Boot 提供了对 Kafka 的良好集成支持，使得开发者可以非常便捷地在项目中使用 Kafka。
本文将手把手教你如何在 Spring Boot 项目中集成 Kafka，包括生产者（Producer）和消费者（Consumer）的实现，并提供完整的代码示例。
开发环境准备
Java 17+
Maven 或 Gradle
Spring Boot 3.x
Apache Kafka 3.0+（本地或远程）
IDE（如 IntelliJ IDEA、VS Code）
创建 Spring Boot 项目
你可以通过 Spring Initializr 创建一个新的 Spring Boot 项目，选择以下依赖：
- Spring Web
- Spring for Apache Kafka
或者手动添加 pom.xml 中的依赖：
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
Spring Boot 会自动管理版本兼容性，无需手动指定版本号。
配置 Kafka 连接信息
在 application.yml 或 application.properties 文件中配置 Kafka 相关参数：
application.yml 示例：
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
group-id: my-group
auto-offset-reset: earliest
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
编写 Kafka 生产者（Producer）
创建一个服务类用于发送消息到 Kafka 主题。
KafkaProducer.java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class KafkaProducer {
private final KafkaTemplate<String, String> kafkaTemplate;
public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
public void sendMessage(String topic, String message) {
kafkaTemplate.send(topic, message);
System.out.println("Sent message: " + message);
}
}
编写 Kafka 消费者（Consumer）
使用 @KafkaListener 注解监听特定主题的消息。
KafkaConsumer.java
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@Service
public class KafkaConsumer {
@KafkaListener(topics = "test-topic", groupId = "my-group")
public void listen(ConsumerRecord<String, String> record) {
System.out.printf("Received message: topic - %s, partition - %d, offset - %d, key - %s, value - %s%n",
record.topic(), record.partition(), record.offset(), record.key(), record.value());
}
}
添加 REST 接口用于测试发送消息
为了方便测试，我们可以创建一个简单的 REST 控制器来触发消息发送。
KafkaController.java
import org.springframework.web.bind.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
@RestController
@RequestMapping("/kafka")
public class KafkaController {
@Autowired
private KafkaProducer kafkaProducer;
@PostMapping("/send")
public String sendMessage(@RequestParam String msg) {
kafkaProducer.sendMessage("test-topic", msg);
return "Message sent: " + msg;
}
}
启动 Kafka 环境（可选）
如果你还没有运行 Kafka，可以按照以下步骤快速启动：
启动 Zookeeper（Kafka 依赖）
bin/zookeeper-server-start.sh config/zookeeper.properties
启动 Kafka 服务
bin/kafka-server-start.sh config/server.properties
创建测试 Topic
bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
测试接口
启动 Spring Boot 应用后，访问如下接口发送消息：
POST http://localhost:8080/kafka/send?msg=HelloKafka
观察控制台输出，确认是否收到类似以下内容：
Received message: topic - test-topic, partition - 0, offset - 5, key - null, value - HelloKafka
扩展功能建议
使用 JSON 格式传输对象（自定义序列化/反序列化）
多消费者组配置与负载均衡
异常处理与重试机制（@DltHandler, SeekToCurrentErrorHandler）
Kafka Streams 实现实时流处理逻辑
配置 SSL、SASL 安全认证
结合 Spring Cloud Stream 构建云原生事件驱动架构
到此这篇关于Spring Boot集成Apache Kafka的实战指南的文章就介绍到这了,更多相关SpringBoot集成Apache Kafka内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
这篇文章主要介绍了如何定位java程序中占用cpu最高的线程堆栈信息方法的相关资料,需要的朋友可以参考下2022-11-11
这篇文章主要介绍了从java源码分析线程池(池化技术)的实现原理，池化技术是一种编程技巧，当程序出现高并发时，能够明显的优化程序，降低系统频繁创建销毁连接等额外开销，下文更多的相关介绍需要的小伙伴可以参考一下2022-04-04
今天小编就为大家分享一篇关于java字符串中${}或者{}等的占位符替换工具类，小编觉得内容挺不错的，现在分享给大家，具有很好的参考价值，需要的朋友一起跟随小编来看看吧2018-12-12
文章讲述了在Java中进行批量操作的最佳实践,包括使用MyBatis的ExecutorType.BATCH模式,作者建议根据实际需求选择是否进行批量操作,因为批量操作并不总是能提升性能,且有副作用,如无法获取自增ID等2024-12-12
Java实现WAV音频拼接彻底摆脱FFmpeg的轻量本地方案
在音频处理领域,FFmpeg 是几乎无所不能的存在,从音频解码、格式转换、拼接到混音,几乎所有任务都能用一句命令完成,然而,正因为它全能,也意味着笨重,本文给大家介绍了利用Java实现WAV音频拼接彻底摆脱FFmpeg的轻量本地方案,需要的朋友可以参考下2025-11-11
本篇文章是对在Java中一些基础概念的使用进行了详细的分析介绍，需要的朋友参考下2013-05-05
Jackson 是 Java 中最流行的 JSON 处理库之一，它提供了许多注解来简化 JSON 的序列化和反序列化过程。这篇文章将介绍一些 Jackson 常用的注解，以帮助您更轻松地处理 JSON 数据2023-05-05
本文详细解析了Token在防止表单重复提交、身份验证及过期处理中的应用,涵盖Token的生成流程、过期机制及续期方案,特别介绍了单token与双token两种策略,需要的朋友可以参考下2026-05-05
这篇文章主要介绍了导入SpringCloud依赖踩的坑及解决方案，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2023-04-04
这篇文章主要介绍了redis发布订阅Java代码实现过程解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2019-09-09
最新评论