企业官网建设流程全解析

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> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka-test</artifactId> <scope>test</scope> </dependency>

如果你是Spring Boot 2.7.x，那对应的spring-kafka是2.8.x；如果是Spring Boot 3.x，spring-kafka也会对应到3.x。不要坚信“高版本客户端一定能连低版本broker”，虽然有兼容性设计，但本地项目里没必要在这个点上冒险。

2.2 用容器启动单机Kafka

Kafka本地启动最省事的方式是直接用容器镜像跑单机实例。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: 1@localhost: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为例，消息体是一个对象，而不是简单字符串，所以值序列化用JsonSerializer：

spring: 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

这里每个参数都不是随便写的：

• acks=all：要求分区副本都确认收到后才算发送成功。单机环境其实无所谓，但生产环境多副本场景下，这个配置能避免leader切换时消息丢失。
• enable.idempotence=true：开启幂等生产者，配合acks=all和max.in.flight.requests.per.connection=5，可以在发送重试的情况下依然保证消息不重复、顺序不乱。注意，幂等一旦开启，同一个生产者在同一分区内不会因为重试产生重复消息。
• linger.ms=5：让生产者攒一批消息再发，而不是来一条发一条。5毫秒对大多数业务来说感知不到延迟，但吞吐量会有明显提升。
• batch.size=16384：16KB的批次大小，和linger配合，适合高频小消息场景。

3.2 KafkaTemplate发送与异步回调

依赖注入KafkaTemplate，直接发送一个对象。Spring Kafka会自动把对象序列化成JSON，并附带类型信息头，供消费者反序列化使用。

@Service public class OrderNotifyPublisher { private static final Logger log = LoggerFactory.getLogger(OrderNotifyPublisher.class); private final KafkaTemplate<String, OrderNotifyMessage> kafkaTemplate; public OrderNotifyPublisher(KafkaTemplate<String, OrderNotifyMessage> kafkaTemplate) { this.kafkaTemplate = kafkaTemplate; } public void publish(OrderNotifyMessage message) { CompletableFuture<SendResult<String, 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_immediate

group-id是消费组标识，同一个组的多个实例会分摊topic分区，不同组各自消费全量消息。一个常见错误是不同业务模块共用一个group-id，结果消息被互相“抢走”。

auto-offset-reset=earliest要结合有没有提交过offset理解：第一次消费、没有历史offset时，earliest表示从最早消息开始消费；如果这个消费组已经提交过offset，这个配置就不再生效。新手经常在这里犯迷糊，以为改了earliest就能重新消费，实际要先重置offset。

enable-auto-commit=false很重要。默认的自动提交是每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(List<ConsumerRecord<String, OrderNotifyMessage>> records, Acknowledgment acknowledgment) { for (ConsumerRecord<String, OrderNotifyMessage> record : records) { log.info("消费消息 key={}, value={}", record.key(), record.value()); // 业务处理 } // 全部处理成功后，再确认这一批offset acknowledgment.acknowledge(); } }

这里listener.type=batch对应方法参数里的List<ConsumerRecord<...>>。如果不用批量模式，参数直接写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，而是真实启动一个嵌入式Kafka，Spring Kafka的@EmbeddedKafka注解可以直接干这件事：

@SpringBootTest @EmbeddedKafka(partitions = 3, topics = {"order-notify-topic"}) class OrderNotifyIntegrationTest { @Autowired private OrderNotifyPublisher publisher; @Autowired private KafkaTemplate<String, 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.ms；
• broker判定消费者死亡，触发rebalance，把A负责的分区分配给B；
• B从上次提交的offset继续消费，A已经处理但未提交的那批消息又被B重新拉取。

这类重复很难完全避免，唯一有效的应对措施就是前面提到的幂等消费。从那以后，我的每一个Kafka消费端都默认带上消息处理记录表，先写记录再处理业务，处理完成再更新状态。

6.3 反序列化失败与死信队列

当消息内容不合法、类结构变更后，反序列化失败会让消费者一直卡在这条消息上，不提交offset就会无限重试。处理方式是用DefaultErrorHandler配合死信topic，把反复失败的消息转储：

@Bean public DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<Object, 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”就已经超过大多数生产代码的水平了。