CodeRoad Map 程序员的学习成长路线图。内容免费开放，登录后可同步路线图与课时进度。
由 程序员柏风 独立维护 · 关于作者
学习 更多
Spring Kafka
这一课你将能做到
在 Spring Boot 中引入 spring-kafka，用 yml 配好 bootstrap-servers、group-id、acks、手动 ack 模式
用 KafkaTemplate 发送带 key 的消息、用 @KafkaListener 消费，并说出「事务提交后再发」的正确写法
配置 DefaultErrorHandler + 死信 Topic，说清「消费失败重试」在 Spring 里的落地方式
为什么要学
裸 KafkaProducer/KafkaConsumer 能看清底层；进 Spring Boot 项目后，Spring Kafka 用自动配置、KafkaTemplate、@KafkaListener 把连接管理、监听线程、序列化样板全收掉，风格和你用 RedisTemplate/StringRedisTemplate 一致。
生产里最常出事的两处，本课直接给答案：一是失败怎么办 （DefaultErrorHandler 重试 + 死信，替代裸客户端的手写重试）；二是事务边界 （@Transactional 方法里不能随手 send，要 afterCommit）。
人话：Spring Kafka = 把「连接、序列化、监听线程」接到 Boot 惯例上的胶水；注解很香，但幂等与位点纪律不会因为用了注解而消失。
先建立图像
application.yml spring.kafka.*
│
▼
ProducerFactory ──► KafkaTemplate.send(topic, key, value) （包着 KafkaProducer）
ConsumerFactory ──► @KafkaListener 方法（容器线程里跑 poll） （包着 KafkaConsumer）
Profile 隔离：
默认 profile → 不启 Kafka（主站不受影响）
dev,kafka → localhost:9092 lab 实验
能力 Spring 写法 对应底层 发送 KafkaTemplateKafkaProducer消费 @KafkaListenerKafkaConsumer + 监听容器消费组 注解 groupId 或 spring.kafka.consumer.group-id group.id失败处理 DefaultErrorHandler + DLT重试 + 死信 Topic 测试 Testcontainers / @EmbeddedKafka 真协议假集群
跟着做
步骤 1：加依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
版本由 Boot BOM 管理，不要手写版本号。
步骤 2：写配置（独立 profile）
# application-kafka.yml
spring:
kafka:
bootstrap-servers: localhost:9092
producer:
acks: all
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
group-id: coderoadmap-lab
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
enable-auto-commit: false
listener:
# RECORD：每条处理完就 ack；手动模式配合 Acknowledgment
ack-mode: manual
启动加 --spring.profiles.active=dev,kafka。主 API（MySQL + Redis 路径）完全不受影响。
步骤 3：发送服务（事务后发送） @Service
@RequiredArgsConstructor
public class OrderEventPublisher {
private final KafkaTemplate<String, String> kafkaTemplate;
/** 只允许在事务提交成功后调用，避免下游处理"幽灵订单" */
public void publishCreated(String orderId, String json) {
kafkaTemplate.send("coderoadmap.order-events", orderId, json)
.whenComplete((result, ex) -> {
if (ex != null) {
// 失败记日志/告警，进补偿任务
} else {
System.out.printf("sent p=%d off=%d%n",
result.getRecordMetadata().partition(),
result.getRecordMetadata().offset());
}
});
}
}
在 @Transactional 方法里保证「提交后再发」，最直观的写法：
@Transactional
public void createOrder(CreateOrderCmd cmd) {
orderRepo.save(Order.from(cmd));
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override public void afterCommit() {
// 此刻事务已落库，发送才安全
orderEventPublisher.publishCreated(order.getId(), toJson(order));
}
});
}
步骤 4：监听器（手动 ack） @Component
@RequiredArgsConstructor
public class OrderEventListener {
private final OrderCreatedHandler handler; // 内含幂等
@KafkaListener(topics = "coderoadmap.order-events",
groupId = "coderoadmap-notify",
concurrency = "3")
public void onMessage(String payload,
@Header(KafkaHeaders.RECEIVED_KEY) String key,
@Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
Acknowledgment ack) {
handler.handleOrderCreated(parse(payload));
ack.acknowledge(); // 处理成功才提交位点
}
}
应看到：发一条，监听日志出现；把 groupId 换成 coderoadmap-audit 再写一个方法 = 第二个独立订阅者。
步骤 5：失败重试 + 死信 Topic @Configuration
public class KafkaErrorConfig {
@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
// 失败消息最终投递到 <原Topic>-dlt
var recoverer = new DeadLetterPublishingRecoverer(template);
var backOff = new ExponentialBackOff(1000L, 2.0); // 1s、2s、4s...
backOff.setMaxAttempts(4);
return new DefaultErrorHandler(recoverer, backOff);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
ConsumerFactory<String, String> cf, DefaultErrorHandler eh) {
var factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
factory.setConsumerFactory(cf);
factory.setCommonErrorHandler(eh);
return factory;
}
}
验证：往监听器塞一条必然解析失败的 JSON，观察日志出现指数退避重试 4 次，最终消息出现在 coderoadmap.order-events-dlt。
步骤 6：接到 REST 端点验证全链路 @PostMapping("/lab/orders")
public Order create(@RequestBody CreateOrderCmd cmd) {
Order order = orderService.create(cmd); // 写 MySQL
publisher.publishCreated(order.getId(), toJson(order));
return order;
}
curl 打一笔，确认三件事：DB 有行、console consumer 能看到消息、监听器有日志。
讲清楚要点
因为 Boot 自动配置了 Producer/Consumer Factory，所以 业务代码统一走 Template/Listener，不要到处 new KafkaProducer 散落配置。因为 @KafkaListener 底层仍是 Consumer Group，所以 groupId 规则一模一样：一职责一 group，不同监听器用不同组实现多订阅。因为 concurrency=N 是「每个实例内起 N 个监听线程」，所以 实际并行 = min(实例数 × N，分区数)，超出的线程空转——并发调优第一步永远是看分区数。因为 默认错误策略有限，所以 上生产前显式配 DefaultErrorHandler：可重试异常退避重试，重试耗尽投死信； Poison 消息（反序列化失败）不重试直接死信。因为 发送与 DB 事务是两个系统，所以 用 afterCommit 或 Outbox 划清边界；KafkaTemplate.send 本身不是「分布式事务」。因为 注解不替你幂等，所以 监听器里改库仍按「eventId 唯一 + 存在即跳过」处理（投递语义课）。
你可能会懵的地方
「yml 配了 group-id，注解又写 groupId，听谁的？」
注解上的 groupId 优先（更具体）。多监听器各配不同组即可实现多订阅；全局 yml 的值是兜底。
「发送成功了监听没动静？」
排查链：profile 是否激活（监听 bean 有没有建）→ Topic 名是否一致 → bootstrap 是否连上（启动日志有元数据）→ 组是否停在 latest（换新组或 earliest 验证）。先用 console consumer 确认消息在不在 Topic 里，再查应用侧。
「ack-mode=manual 和裸客户端 commitSync 什么关系？」
ack.acknowledge() 底层就是提交位点（按 ack-mode 决定粒度）。manual = 你处理完手动调；record = 容器逐条代提。关键纪律不变：处理成功才 ack 。
「Listener 里能直接改订单主状态吗？」
技术上可以，但主状态变更属于命令路径，通常留在 API 事务里；监听器更常做投影/通知/审计类副产物，且必须幂等。
错法对照 错法 问题 正解 @Transactional 方法体内直接 kafkaTemplate.send事务回滚后消息已出去 → 幽灵事件 afterCommit 发送或 Outbox 所有监听器共用一个 groupId 互抢分区，各自收不全 一职责一 groupId 监听器无幂等直接插业务表 rebalance/重启后重复执行 eventId 唯一约束 + 跳过 解析失败的消息无限重试 组卡死在这一条（毒丸） DefaultErrorHandler 重试上限 + DLT 生产主站强依赖 kafka 配置 无 Broker 时应用起不来 独立 profile，主路径可关
卡点急救
应用启动报连不上 localhost:9092 ：先 docker compose ps 确认 Broker 在跑；再确认激活了正确 profile 而不是主配置里漏配。@KafkaListener 方法没被调用auto-offset-reset 与组位点；用 kafka-consumer-groups --describe 看该组是否有成员。消息重复消费告警 ：先确认幂等是否生效（去重表插没插上），再查是否 rebalance 频繁（日志关键字 rebalance）；重复是模型常态，幂等是答案，不是异常。
面试常问
Spring Kafka 消费失败怎么处理？
分层答：可重试异常（网络、锁超时）→ DefaultErrorHandler 配指数退避重试；重试耗尽 → DeadLetterPublishingRecoverer 投 <topic>-dlt 死信并告警；不可重试异常（反序列化失败）→ 直接死信不浪费时间。同时监听器内部保持幂等，因为重试本质就是再次投递。
@KafkaListener 的 concurrency 怎么理解？
它是实例内 监听容器线程数，不是跨实例并行。总并行 = min(实例数 × concurrency，分区数)。Topic 3 分区、单实例 concurrency=6，只有 3 个线程真正在消费。所以并发瓶颈先看分区规划。
怎么在 Spring 里做到「事务提交后再发消息」？
两种：①TransactionSynchronizationManager.registerSynchronization 的 afterCommit 回调里发送（轻量，但发送失败需补偿）；②Outbox 模式：同事务写业务表 + outbox 表，由定时任务/CDC 异步发 Kafka（强一致场景）。面试能说出「为什么要 afterCommit」比背 API 更重要。
自测三问
@PostMapping 创建订单后如何调用 KafkaTemplate.send？key 传什么？发送失败在哪感知？@KafkaListener 的 groupId 与 yml 的 group-id 关系？为什么通知与审计不能共用？为什么不能在 Listener 里改 MySQL 主表却不做幂等？给一个最小幂等方案。
小结一句话 Spring Kafka = Boot 配置 + KafkaTemplate 发送 + 注解消费 + DefaultErrorHandler 兜底；独立 profile 做实验，幂等与「事务后发送」纪律不因注解而消失。
延伸