Spring Kafka

这一课你将能做到

1. 在 Spring Boot 中引入 spring-kafka，用 yml 配好 bootstrap-servers、group-id、acks、手动 ack 模式
2. 用 KafkaTemplate 发送带 key 的消息、用 @KafkaListener 消费，并说出「事务提交后再发」的正确写法
3. 配置 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 写法	对应底层
发送	KafkaTemplate	KafkaProducer
消费	@KafkaListener	KafkaConsumer + 监听容器
消费组	注解 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));
        }
    });
}

更彻底的方案（Outbox）见「投递语义」课。

步骤 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 能看到消息、监听器有日志。

讲清楚要点

1. 因为Boot 自动配置了 Producer/Consumer Factory，所以业务代码统一走 Template/Listener，不要到处 new KafkaProducer 散落配置。
2. 因为@KafkaListener 底层仍是 Consumer Group，所以 groupId 规则一模一样：一职责一 group，不同监听器用不同组实现多订阅。
3. 因为concurrency=N 是「每个实例内起 N 个监听线程」，所以实际并行 = min(实例数 × N，分区数)，超出的线程空转——并发调优第一步永远是看分区数。
4. 因为默认错误策略有限，所以上生产前显式配 DefaultErrorHandler：可重试异常退避重试，重试耗尽投死信； Poison 消息（反序列化失败）不重试直接死信。
5. 因为发送与 DB 事务是两个系统，所以用 afterCommit 或 Outbox 划清边界；KafkaTemplate.send 本身不是「分布式事务」。
6. 因为注解不替你幂等，所以监听器里改库仍按「eventId 唯一 + 存在即跳过」处理（投递语义课）。

你可能会懵的地方

1. 「yml 配了 group-id，注解又写 groupId，听谁的？」 注解上的 groupId 优先（更具体）。多监听器各配不同组即可实现多订阅；全局 yml 的值是兜底。

2. 「发送成功了监听没动静？」 排查链：profile 是否激活（监听 bean 有没有建）→ Topic 名是否一致 → bootstrap 是否连上（启动日志有元数据）→ 组是否停在 latest（换新组或 earliest 验证）。先用 console consumer 确认消息在不在 Topic 里，再查应用侧。

3. 「ack-mode=manual 和裸客户端 commitSync 什么关系？」 ack.acknowledge() 底层就是提交位点（按 ack-mode 决定粒度）。manual = 你处理完手动调；record = 容器逐条代提。关键纪律不变：处理成功才 ack。

4. 「Listener 里能直接改订单主状态吗？」 技术上可以，但主状态变更属于命令路径，通常留在 API 事务里；监听器更常做投影/通知/审计类副产物，且必须幂等。

错法对照

错法	问题	正解
@Transactional 方法体内直接 kafkaTemplate.send	事务回滚后消息已出去 → 幽灵事件	afterCommit 发送或 Outbox
所有监听器共用一个 groupId	互抢分区，各自收不全	一职责一 groupId
监听器无幂等直接插业务表	rebalance/重启后重复执行	eventId 唯一约束 + 跳过
解析失败的消息无限重试	组卡死在这一条（毒丸）	DefaultErrorHandler 重试上限 + DLT
生产主站强依赖 kafka 配置	无 Broker 时应用起不来	独立 profile，主路径可关

卡点急救

1. 应用启动报连不上 localhost:9092：先 docker compose ps 确认 Broker 在跑；再确认激活了正确 profile 而不是主配置里漏配。
2. @KafkaListener 方法没被调用：查类是否被组件扫描到（在启动类包路径下）、topic 拼写、auto-offset-reset 与组位点；用 kafka-consumer-groups --describe 看该组是否有成员。
3. 消息重复消费告警：先确认幂等是否生效（去重表插没插上），再查是否 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 更重要。

自测三问

1. @PostMapping 创建订单后如何调用 KafkaTemplate.send？key 传什么？发送失败在哪感知？
2. @KafkaListener 的 groupId 与 yml 的 group-id 关系？为什么通知与审计不能共用？
3. 为什么不能在 Listener 里改 MySQL 主表却不做幂等？给一个最小幂等方案。

小结一句话

Spring Kafka = Boot 配置 + KafkaTemplate 发送 + 注解消费 + DefaultErrorHandler 兜底；独立 profile 做实验，幂等与「事务后发送」纪律不因注解而消失。

延伸

• Spring for Apache Kafka 官方参考文档
• 下一课「消息格式」：消息体里该放什么字段、怎么演进才不炸掉老消费者。