1. 这个Demo到底解决什么问题从一次订单通知改造说起我最早接触Spring Boot集成Kafka和大多数人一样第一个需求就是找一个能直接跑起来的springboot整合kafka案例demo。但很快发现市面上的demo大多停留在“生产者发一条字符串消费者打印出来”的程度真正落地到业务时序列化怎么配、offset怎么提交、消息重复怎么处理这些反而没人讲清楚。我实际搞过一个订单通知服务。一开始没有Kafka下单接口里直接同步调用通知逻辑用户下单高峰期数据库连接、短信通道、外部推送接口全被打满。后来改成“订单创建成功先写库再发一条Kafka消息通知服务异步消费”整个接口的耗时立刻降下来下游再抖动也不至于拖垮主链路。Kafka在这个场景里解决的核心问题就是解耦、削峰、异步重试。这篇文章就按这份实践拆开写环境怎么搭、生产者怎么写、消费者怎么配、序列化和幂等怎么处理、测试和排错怎么走。适合两类人一类是刚接触Spring Boot整合Kafka的开发者想通过demo把整个机制跑通另一类是已经能跑通demo但不确定生产配置怎么选、后续会遇到哪些坑的人。1.1 没有Kafka时我们是怎么做的旧的调用链路是同步的。用户在前端点“提交订单”后端先写订单表再调用会员服务更新积分再调用消息渠道发通知任何一个下游慢用户就得等一个下游挂了整个下单事务可能要回滚或者重试到超时。这个模式在小流量下没问题流量一上来就特别脆弱。尤其是通知这类服务它本身不要求跟主交易强一致你晚几秒甚至晚几分钟发给用户影响不大。所以它天然适合异步化主流程只做必须做的事通知这类动作丢到消息队列里慢慢消化。1.2 为什么选Kafka而不是其他方案技术选型上团队当时也犹豫过。如果只是“给服务之间加个缓冲”轻量级队列也能做但我们需要几个能力消息要能留存一段时间消费端出问题后可以重置offset重新消费多个业务模块要能独立订阅同一份事件流互不影响消费端要能水平扩展实例多了之后自动分担分区后续可能还要对订单事件做统计、分析、归档。这几条综合下来Kafka是比较合适的选择。它不像消息队列那样强调“推”模式而是消费端主动拉取、记录自己的消费位置天然适合“每条消息被多个消费组各消费一次”的场景。2. 环境准备与依赖选型最小代价把Kafka在本地跑起来很多demo跑不起来不是代码问题是Kafka本身没起来。所以先把环境这关过掉后面一切才顺。2.1 Maven依赖与版本匹配Spring Boot整合Kafka官方已经替我们做了自动化配置依赖上加一个spring-kafka即可。这里务必注意版本匹配Spring Boot对Kafka客户端的版本有固定管理手动乱填版本号很容易出现序列化类找不到、消费组协调协议对不上等奇怪问题。我自己用到的依赖长这样dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdorg.springframework.kafka/groupId artifactIdspring-kafka/artifactId /dependency dependency groupIdorg.springframework.kafka/groupId artifactIdspring-kafka-test/artifactId scopetest/scope /dependency如果你是Spring Boot 2.7.x那对应的spring-kafka是2.8.x如果是Spring Boot 3.xspring-kafka也会对应到3.x。不要坚信“高版本客户端一定能连低版本broker”虽然有兼容性设计但本地项目里没必要在这个点上冒险。2.2 用容器启动单机KafkaKafka本地启动最省事的方式是直接用容器镜像跑单机实例。Kafka从3.x开始支持不再依赖ZooKeeper的KRaft模式单机一个节点就能起一个完整的broker我把配置贴在下面可以直接抄services: kafka: image: apache/kafka:3.6.0 ports: - 9092:9092 environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_CONTROLLER_QUORUM_VOTERS: 1localhost:9093 KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1保存为docker-compose.yml执行docker compose up -d。容器日志中出现类似“started”的字样基本就说明broker已经正常启动。这里有个新手经常踩的坑KAFKA_ADVERTISED_LISTENERS必须写成宿主机地址不能写成容器内地址否则你的Spring Boot应用在宿主机上访问localhost:9092会失败。我第一次自己打包容器时就是没注意这个应用一直报连接超时。2.3 校验是否正常不用急着写代码先确认端口和topic操作是否正常。命令行里执行# 创建测试主题 docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \ --create --topic order-notify-topic \ --partitions 3 --replication-factor 1 \ --bootstrap-server localhost:9092返回“Created topic”说明broker和topic都正常。我建议topic直接创建成3个分区不要用默认的1个分区。原因后面讲消费并发时会提到这里先留个伏笔。3. 生产端配置与消息发送决定消息能不能出去的五个细节环境起来之后先写生产者。很多人以为生产者的核心代码只有一行kafkaTemplate.send(...)其实Send只是最后一步前面那些配置参数直接影响消息能不能可靠发出去、会不会乱序、会不会重复。3.1 一份可以直接抄走的application.yml以“订单通知”这个demo为例消息体是一个对象而不是简单字符串所以值序列化用JsonSerializerspring: kafka: bootstrap-servers: localhost:9092 producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer acks: all properties: enable.idempotence: true linger.ms: 5 batch.size: 16384 max.in.flight.requests.per.connection: 5这里每个参数都不是随便写的acksall要求分区副本都确认收到后才算发送成功。单机环境其实无所谓但生产环境多副本场景下这个配置能避免leader切换时消息丢失。enable.idempotencetrue开启幂等生产者配合acksall和max.in.flight.requests.per.connection5可以在发送重试的情况下依然保证消息不重复、顺序不乱。注意幂等一旦开启同一个生产者在同一分区内不会因为重试产生重复消息。linger.ms5让生产者攒一批消息再发而不是来一条发一条。5毫秒对大多数业务来说感知不到延迟但吞吐量会有明显提升。batch.size1638416KB的批次大小和linger配合适合高频小消息场景。3.2 KafkaTemplate发送与异步回调依赖注入KafkaTemplate直接发送一个对象。Spring Kafka会自动把对象序列化成JSON并附带类型信息头供消费者反序列化使用。Service public class OrderNotifyPublisher { private static final Logger log LoggerFactory.getLogger(OrderNotifyPublisher.class); private final KafkaTemplateString, OrderNotifyMessage kafkaTemplate; public OrderNotifyPublisher(KafkaTemplateString, OrderNotifyMessage kafkaTemplate) { this.kafkaTemplate kafkaTemplate; } public void publish(OrderNotifyMessage message) { CompletableFutureSendResultString, OrderNotifyMessage future kafkaTemplate.send(order-notify-topic, message.orderId(), message); future.whenComplete((result, ex) - { if (ex null) { log.info(消息已确认 topic{}, partition{}, offset{}, result.getRecordMetadata().topic(), result.getRecordMetadata().partition(), result.getRecordMetadata().offset()); } else { log.error(消息发送失败orderId{}, message.orderId(), ex); // 这里要落本地表后续补偿任务扫描重发 } }); } }注意send本身是异步的调用后立刻返回。如果你只调send不处理回调失败消息就悄无声息丢了。这是审查生产环境代码时必看的一个点。key传message.orderId()有两个作用一是相同订单的消息落到同一个分区保证同一个订单的消息消费顺序二是当消息体自身不带唯一标识时key是幂等判断的重要依据。3.3 生产者重试到底能不能解决所有问题不能这是必须说清楚的一点。retries虽然默认值已经不低但重试只能处理broker临时不可用、网络抖动这类问题。下面的情况仍然需要业务层兜底消息体太大超过broker限制怎么重试都会失败topic不存在且自动创建被关闭自定义序列化器抛异常。所以生产者的真实实践是发消息时不要只调send必须在回调里记录失败日志并落到一张本地消息表由定时任务扫描后重新发送。demo可以简化生产代码不行。4. 消费端配置与KafkaListener线程模型手写监听器之前的必要认知生产者还相对简单消费端才是真正容易出问题的地方。刚接触时我以为KafkaListener就是一个“收到消息就回调”的注解结果被消费线程模型、offset提交机制反复教育。4.1 消费端配置清单还是先把配置贴全逐项解释spring: kafka: consumer: group-id: order-notify-group auto-offset-reset: earliest enable-auto-commit: false key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: spring.json.trusted.packages: * max.poll.interval.ms: 300000 max.poll.records: 200 listener: type: batch ack-mode: manual_immediategroup-id是消费组标识同一个组的多个实例会分摊topic分区不同组各自消费全量消息。一个常见错误是不同业务模块共用一个group-id结果消息被互相“抢走”。auto-offset-resetearliest要结合有没有提交过offset理解第一次消费、没有历史offset时earliest表示从最早消息开始消费如果这个消费组已经提交过offset这个配置就不再生效。新手经常在这里犯迷糊以为改了earliest就能重新消费实际要先重置offset。enable-auto-commitfalse很重要。默认的自动提交是每5秒提交一次当前offset存在两个风险一是处理失败但offset已提交消息就丢了二是提交延迟导致重启后重复消费。生产环境我建议一律手动提交。4.2 线程模型与会话超时KafkaListener背后是一个监听器容器里面会根据并发配置启动若干个消费线程。一条核心规则是一个消费线程同一时刻只能处理一个分区线程数超过分区数时多出来的线程是闲置的。这也是前面建议topic建3个分区的原因。如果只有一个分区那么哪怕你把并发放到10也只有一条消费线程在拉消息。我见过有人把并发从1调到10发现消费速度没有任何变化最后查下来topic只有1个分区。另一个高频故障是“消费线程假死”。Kafka通过max.poll.interval.ms控制消费者两次拉消息的间隔。如果你的监听方法里做了耗时的数据库操作一次处理超过5分钟还没返回broker会认为该消费者挂了触发rebalance把分区分配给其他消费者导致的消息重复消费几乎必然发生。提示KafkaListener的处理方法里尽量只做轻量操作。耗时的IO操作要么拆出去要么把max.poll.interval.ms调大但调大只是缓解不是根治。4.3 手动提交Offset的正确姿势手动提交有两种常用模式manual和manual_immediate。区别在于manual是一次poll批次处理完成后统一提交如果中途抛出异常这一批之前已经处理完的消息offset也会被提交回退容易造成部分重复manual_immediate则是每次方法返回后立刻提交配合批量监听更加可控。批量监听方法大致这样写Component public class OrderNotifyConsumer { KafkaListener(topics order-notify-topic, groupId order-notify-group) public void onMessage(ListConsumerRecordString, OrderNotifyMessage records, Acknowledgment acknowledgment) { for (ConsumerRecordString, OrderNotifyMessage record : records) { log.info(消费消息 key{}, value{}, record.key(), record.value()); // 业务处理 } // 全部处理成功后再确认这一批offset acknowledgment.acknowledge(); } }这里listener.typebatch对应方法参数里的ListConsumerRecord...。如果不用批量模式参数直接写ConsumerRecord即可。手动提交的核心原则只有一条**业务成功了才提交没成功就不要提交。**但实际工程里更稳妥的是下面要讲的幂等消费思路。5. 序列化格式、批量消费与幂等消费Demo跑通后必须面对的三道坎如果只是send字符串、接收字符串demo能跑通但业务系统里的消息几乎都是对象。这里有三道坎是跑通demo之后立刻会遇到的。5.1 Json序列化与泛型丢失用JsonSerializer发送对象时Spring Kafka会在消息头写入类型信息比如包名和类名。消费者端对应的JsonDeserializer依赖这个Header完成反序列化。这就带来第一个坑如果你的消费者类路径跟生产者不一致比如生产者在服务A、消费在服务B但实体类所在包的路径不同反序列化时就会报ClassNotFoundException。解决方式是配置spring.json.trusted.packages指定允许反序列化的包前缀同时让生产者和消费者共用同一个实体类包。千万不要图方便直接配*内部demo可以生产环境等于把反序列化限制全部放开风险太大。如果消息类型不确定另一个办法是value不用对象统一用JSON字符串。代价是消费端需要自己解析好处是彻底避开信任包配置和泛型丢失问题。我见过不少团队从“对象直传”改成“字符串传递消费端二次解析”反而省了很多跨服务联调问题。5.2 批量消费如何落库批量消费通常是为了吞吐。一次拉200条业务上往往需要批量落库否则一条条insert就失去了批量优势。批量落库的第一步是数据校验把不合格的记录直接过滤掉第二步是主键冲突处理用数据库upsert或者insert ignore第三步才是在全部记录处理完后调用acknowledge()。这里要特别小心如果200条里有一条处理异常你选择“整个批次不提交、从头再来”那这一批前面的199条都会重新消费。比较好的做法是给每条消息一个唯一业务键落库用唯一索引去重这样重放多少遍都不会产生脏数据。5.3 幂等消费与消息重复Kafka的“至少一次”语义加上消费者重平衡决定了重复消费是常态而不是例外。我不建议在业务代码里赌“肯定不会重复”而是默认每条消息都可能消费两遍。最实用的幂等方案是两个条件叠加第一消息体里要有一个全局唯一业务ID第二消费端把它作为数据库唯一索引或缓存锁。以订单通知为例可以建一张消息处理记录表business_id设为唯一索引CREATE TABLE message_consume_log ( id BIGINT AUTO_INCREMENT PRIMARY KEY, business_id VARCHAR(64) NOT NULL, topic VARCHAR(64) NOT NULL, status VARCHAR(16) NOT NULL, created_time DATETIME NOT NULL, UNIQUE KEY uk_business (business_id) );处理前先尝试插入这条记录插入冲突说明之前处理过直接跳过try { messageConsumeLogMapper.insertIgnore(businessId); } catch (DuplicateKeyException e) { log.info(重复消息跳过处理 businessId{}, businessId); return; }这种方式比单纯用Redis分布式锁更可靠因为数据库唯一索引是强一致性。业务处理完后再把状态更新为成功配合定时扫描状态不是成功的记录还能实现补偿。这个思路才是生产级消费端的核心骨架。6. 测试与排错从本地日志到那把最常见的悬案环境和代码都写完最后一定要把测试和排错讲透。因为demo项目里一切正常不代表换到生产环境一切正常。6.1 本地集成测试怎么写本地集成测试最简单的做法不是用Mock而是真实启动一个嵌入式KafkaSpring Kafka的EmbeddedKafka注解可以直接干这件事SpringBootTest EmbeddedKafka(partitions 3, topics {order-notify-topic}) class OrderNotifyIntegrationTest { Autowired private OrderNotifyPublisher publisher; Autowired private KafkaTemplateString, String testTemplate; Test void shouldSendAndConsumeMessage() { // 发送一条真实消息等待消费 publisher.publish(new OrderNotifyMessage(DEMO001, ORDER_CREATED)); // 这里通过引入真正的消费者容器断言业务表或日志里已经出现对应记录 await().atMost(Duration.ofSeconds(10)) .until(() - consumeLogExists(DEMO001)); } }这种测试会真实走一遍生产者和消费者的完整链路包括序列化、offset提交、反序列化比单测有价值得多。注意测试环境的bootstrap-servers要用嵌入式Kafka的端口通常通过测试配置覆盖。6.2 一个典型的重复消费案例有一次在测试环境我模拟“消费者处理到一半抛异常”的场景发现消息被反复消费而且不是重试机制造成的是rebalance引起的。排查链路是这样的生产者发送100条消息到3个分区消费者实例A处理到第50条时有一条消息处理数据库超时超过了max.poll.interval.msbroker判定消费者死亡触发rebalance把A负责的分区分配给BB从上次提交的offset继续消费A已经处理但未提交的那批消息又被B重新拉取。这类重复很难完全避免唯一有效的应对措施就是前面提到的幂等消费。从那以后我的每一个Kafka消费端都默认带上消息处理记录表先写记录再处理业务处理完成再更新状态。6.3 反序列化失败与死信队列当消息内容不合法、类结构变更后反序列化失败会让消费者一直卡在这条消息上不提交offset就会无限重试。处理方式是用DefaultErrorHandler配合死信topic把反复失败的消息转储Bean public DefaultErrorHandler kafkaErrorHandler(KafkaTemplateObject, Object kafkaTemplate) { DeadLetterPublishingRecoverer recoverer new DeadLetterPublishingRecoverer(kafkaTemplate, (record, ex) - new TopicPartition(record.topic() .DLT, record.partition())); return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L)); }配置好之后重试3次仍然失败的消息会进入以.DLT结尾的死信topic消费者主题里不再无限阻塞。这个机制一定要在demo里实现一遍否则你上线后迟早要手动写脚本去跳过卡住的offset。排查这类问题时首选看这几项消费者日志里有没有CommitFailedException有就是offset提交失败导致重复消费有没有NoSuchMethodError或ClassNotFoundException有就是客户端版本和broker不兼容有没有SerializationException有就是消息格式和实体类对不上。我排查过好几次“消费者突然不消费”的假象最后原因都很简单topic被删掉重建消费组还在继续拉但offset从最早开始追看起来就像卡住。这时候用命令行重置一下消费组offset消息马上就能继续消费。bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \ --group order-notify-group --reset-offsets --to-earliest --execute这个操作风险很大只建议在验证topic确实需要重放时使用不要在生产环境乱执行。如果要从这份实践里带走一句话我的建议是Kafka整合Spring Boot这件事配置比代码重要消费比生产复杂幂等一定要前置。先把这三块想清楚你的“案例demo”就已经超过大多数生产代码的水平了。