kafka实践教程

第一章 Kafka 核心概念

写业务代码前，先理清 Kafka 的通信模型与基础组件。不理解这些设计动机，使用 KafkaTemplate、@KafkaListener 与 DefaultErrorHandler 时就容易只会照搬配置。

1.1 同步调用的瓶颈

假设一个电商系统在处理创建订单请求 POST /orders 时，采用传统的单体同步调用：

public void createOrder() {
    orderService.create();    // 100ms
    stockService.decrease();  // 100ms
    smsService.send();        // 300ms
    pointService.add();       // 100ms
    logService.record();      // 50ms
}

整个请求的链路耗时为各项之和（约 650ms）。但在多数业务场景下，调用方只需确认“订单是否创建成功”，而短信通知、积分发放、行为日志与推荐更新等操作，延迟数百毫秒甚至数秒并不会破坏业务一致性。

引入事件驱动后，订单服务仅需在本地完成核心入库逻辑，随后对外发布一条“订单已创建”事件即可返回响应，耗时直接压缩到 100ms 左右：

                 ┌── 短信服务
                 │
OrderCreated ────┼── 积分服务
                 │
                 ├── 推荐服务
                 │
                 └── 日志服务

后续旁路系统各自监听事件、异步消费，彼此独立执行。

1.2 引入消息队列的收益

消息队列充当了调用方与处理方之间的缓冲媒介：

1. 异步加速：将非强一致性的从属逻辑剥离出主响应链路，直接降低接口延迟。
2. 系统解耦：上游只负责投递结构化事件，下游增减消费者（如新增风控系统）无需改动订单核心服务的代码。
3. 流量削峰：当下游数据库最大承载能力为 1000 QPS，而瞬时请求激增到 10000 QPS 时，消息先在队列中持久化排队，消费者按自身处理能力匀速拉取，避免底层存储被瞬间压垮。

1.3 Kafka 消息模型与关键拓扑

Apache Kafka 是一套高吞吐、持久化、支持水平扩展的分布式事件流平台。在本项目中，各组件承载如下分工：

OrderController -> OrderProducer -> Kafka -> OrderConsumer -> OrderProcessService

一条写入 Kafka 的物理记录通常包含以下属性：

topic:      所属逻辑主题
partition:  所在物理分区编号
offset:     分区内的位点
key:        路由键，决定分区归属
value:      实际业务载荷（如序列化后的 JSON 字符串）
timestamp:  时间戳
headers:    元数据键值对

项目消费者在参数中拿到的 ConsumerRecord<String, String>，就是该记录在客户端中的映射对象。

Broker 与集群

Broker 是 Kafka 的服务节点。生产集群通常由多个 Broker 协同工作以实现高可用；本 Demo 为简化部署，配置单节点服务监听于 localhost:9092。

Topic 与 Partition

Topic 是划分消息类型的逻辑主题，例如 order-events（正常事件）与 order-events-dlt（死信事件）。每个 Topic 可进一步切分为多个 Partition。

Partition 是 Kafka 保证并发能力与横向扩展的物理单元。单 Partition 同一时刻只能由同一 Consumer Group 内的一个消费者实例独占拉取。增加 Partition 数量，能够提升该消费组的消费并发上限。

单分区有序性

Partition 是仅支持顺序追加写（Append-only）的提交日志。消息按入队次序依次分配单调递增的偏移量：

Partition 0:  [offset 0: A] -> [offset 1: B] -> [offset 2: C] -> [offset 3: D]

Kafka 仅保证单个 Partition 内部的消息有序，不同 Partition 之间不存在全局顺序保证。

Offset

Offset 是消息在其所属 Partition 内的序列编号。它是相对概念，并非全局唯一 ID。定位一条物理记录的完整坐标为：Topic + Partition + Offset。消费者通过提交已消费的 Offset，向集群记录当前组在该分区的读取进度。

Consumer Group

同属一个 Consumer Group 的消费者协同消费一个 Topic。若 Topic 包含 3 个分区：

• 组内有 1 个 Consumer 时，该实例同时处理 3 个分区；
• 组内有 3 个 Consumer 时，每个实例分别独占消费 1 个分区；
• 组内若启动 4 个以上的 Consumer，多出的实例将处于空闲待命状态。

若有两个不同的 Consumer Group（例如订单处理服务与数据分析服务）订阅同一 Topic，两组会完全独立地推进各自的 Offset，互不影响。

消息路由键：Key 的选择

Kafka 默认使用 Murmur2 算法对 Key 计算哈希，并对 Partition 总数取模，以此决定消息路由的目标分区。只要 Topic 的分区数量不发生变更，相同 Key 的消息将始终落入同一个 Partition。

如果使用每个事件唯一的 UUID 作为 Key，同一订单的“创建”、“支付”、“发货”可能会散落到不同分区，被不同线程并发处理，从而导致“支付先于创建被消费”的乱序问题。因此本项目强制将业务主键 orderId 设为 Kafka Key，利用单分区有序性保证同一订单的生命周期顺序；每次操作唯一的 eventId 则放在消息体中，用于消费端的幂等去重。

ACK 与重复消费成因

ACK 是消费者向框架确认消息已处理完成的信号。在手动 ACK 模式下，代码必须在本地事务成功提交之后调用 acknowledge()。如果过早确认，后续业务执行抛出异常时，该消息在 Kafka 看来已被消费，将直接引发漏处理风险。

相反，如果业务处理已在 MySQL 中成功入库，但在执行 ACK 或框架提交 Offset 的瞬间进程宕机，Kafka 由于未收到确认，在消费者重启后仍会自上次已提交的 Offset 处重新投递该消息。因为存在这种“处理成功但确认丢失”的情况，消费端必须实现业务幂等。

临时重试与死信队列（DLT）

网络抖动或短时死锁属于瞬时异常，通常在重试一到两次后即可自愈；而 JSON 结构损坏属于永久性异常，原地无限重试只会白白占用线程与数据库连接，甚至造成单分区阻塞（即毒消息，Poison Message）。因此系统需要实施有限退避重试，耗尽重试次数后将数据投递至死信队列（order-events-dlt），隔离异常并交由告警系统介入。

第二章 项目架构总览

2.1 技术栈

组件	版本 / 来源	用途
Spring Boot	4.1.1	基础设施框架
Spring Kafka	与 Boot 版本配套	Kafka 客户端封装
Jackson	与 Boot 版本配套	JSON 序列化工具
MySQL	8.0+	业务存储与防重记录
Java	17	运行环境

2.2 目录组织

com.zzh.kafkademo
├── config
│   ├── JacksonConfig.java        # 序列化配置
│   ├── KafkaErrorConfig.java     # 重试与死信队列拓扑配置
│   └── KafkaTopicConfig.java     # 主题声明与分区参数
├── consumer
│   ├── OrderConsumer.java        # 订单业务监听器
│   └── OrderDltConsumer.java     # 死信消息监听器
├── controller
│   └── OrderController.java      # 发送与重放接口
├── exception
│   └── BadMessageException.java  # 不可重试异常标识
├── model
│   ├── CreateOrderRequest.java   # HTTP 请求体
│   ├── OrderEvent.java           # 核心领域事件
│   └── ProcessResult.java        # 处理状态枚举（SUCCESS/DUPLICATE）
├── producer
│   └── OrderProducer.java        # 消息发送与回调组件
└── service
    ├── FailureSimulator.java     # 故障注入工具
    └── OrderProcessService.java  # 事务落库与去重核心逻辑

2.3 核心数据流

整个流程按“先提交业务事务、再确认位点”的顺序执行：

HTTP POST /orders
    │
    ▼
OrderController (生成事件 UUID，设定 orderId 为路由键)
    │
    ▼
OrderProducer (通过 Jackson 序列化为 JSON 文本)
    │
    ▼
Kafka Broker (落入 order-events 的特定 Partition)
    │
    ▼
OrderConsumer (拉取记录，解析 JSON 为 OrderEvent)
    │
    ▼
OrderProcessService.process() [@Transactional 事务保护]
    ├── 1. INSERT IGNORE INTO processed_event (占位排重)
    │      若主键冲突直接返回 DUPLICATE
    ├── 2. FailureSimulator.maybeFail() (故障模拟拦截)
    └── 3. INSERT INTO orders (业务落库)
    │
    ├─ 正常提交 ──> 执行 acknowledgment.acknowledge()
    │
    └─ 抛出异常 ──> 触发数据库 ROLLBACK (去重占位同步撤销)
                   │
                   ▼ 异常传递给 DefaultErrorHandler
                   ├── 可重试异常：按 FixedBackOff 重新拉取
                   └── 达到上限或不可重试：转入 order-events-dlt

第三章 环境搭建

3.1 容器化运行 Kafka（KRaft 模式）

Kafka 现代版本原生支持 KRaft 协议管理集群元数据，去除了对 ZooKeeper 的外部依赖。通过 Docker Compose 启动单节点服务：

services:
  kafka:
    image: confluentinc/cp-kafka:7.6.0
    container_name: kafka
    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_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

执行命令在后台拉起实例：

docker compose up -d

3.2 数据库 Schema 设计

在 MySQL 中创建去重表与订单业务表：

CREATE DATABASE IF NOT EXISTS kafka_demo
DEFAULT CHARACTER SET utf8mb4
DEFAULT COLLATE utf8mb4_general_ci;

USE kafka_demo;

-- 幂等去重底表
CREATE TABLE IF NOT EXISTS processed_event (
    event_id     VARCHAR(36) PRIMARY KEY COMMENT '事件唯一标识',
    order_id     BIGINT      NOT NULL COMMENT '订单ID',
    processed_at DATETIME    NOT NULL COMMENT '处理时间'
) ENGINE = InnoDB;

-- 业务订单表
CREATE TABLE IF NOT EXISTS orders (
    order_id   BIGINT        PRIMARY KEY COMMENT '订单ID',
    user_id    BIGINT        NOT NULL COMMENT '用户ID',
    product_id BIGINT        NOT NULL COMMENT '商品ID',
    amount     DECIMAL(10,2) NOT NULL COMMENT '订单金额',
    status     VARCHAR(20)   NOT NULL COMMENT '订单状态',
    created_at DATETIME      NOT NULL COMMENT '创建时间'
) ENGINE = InnoDB;

去重依靠 event_id 主键约束：并发或重复投递的同一事件再次写入时，会被 InnoDB 的主键唯一性约束拦下。

3.3 验证环境可用性

# 验证 Kafka Broker 协议响应
docker exec kafka kafka-broker-api-versions --bootstrap-server localhost:9092

# 验证 MySQL 库创建成功
mysql -u root -p -e "SHOW DATABASES LIKE 'kafka_demo'"

第四章 Kafka 配置详解

4.1 配置文件一览 (application.yml)

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      properties:
        enable.idempotence: true
    consumer:
      group-id: order-service-v2
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      enable-auto-commit: false
      auto-offset-reset: earliest
    listener:
      ack-mode: manual_immediate

4.2 生产端核心参数

acks = all

该参数指示 Leader 副本在收到消息后，必须等待同步副本集合（In-Sync Replicas, ISR）全量完成同步确认，才向生产者返回成功响应。从单节点扩展到集群后，它用来防止 Leader 掉线时丢失消息。

enable.idempotence = true

开启生产端原生幂等后，Kafka 客户端会为每个 Producer 实例分配唯一的 PID，并为发送到每个分区的消息维护单调自增的 Sequence Number。一旦因网络丢包导致客户端未收到 Broker 确认并自动重试，Broker 端可根据 PID 与序列号过滤掉重复追加的记录。需要注意，这只防范了 Producer 到 Broker 链路的写入重复，不能解决消费端的业务重复问题。

为什么选择手写 JSON 序列化？

Spring Kafka 提供了开箱即用的 JsonSerializer，但其默认会在 Record Header 中塞入全限定 Java 类名，导致不同微服务间存在强类路径耦合。这里先用 Jackson 把领域对象序列化为普通 JSON 字符串，再交给 StringSerializer 转成字节流，其他系统解析报文时不依赖 Java 类路径。

4.3 消费端与进度控制

enable-auto-commit = false
ack-mode = manual_immediate

开启自动提交后，消费者会在后续的 poll() 调用中按 auto.commit.interval.ms（默认 5 秒）提交已拉取到的 Offset。如果消息在 poll 循环中同步处理，宕机时还没提交的进度会让消息被重新消费，更常见的后果是重复；如果处理被交给异步线程，Offset 可能在处理完成前就已提交，此时宕机，这部分消息就会丢失。关闭自动提交并设置为 manual_immediate 后，Offset 由业务代码显式提交，只有消息处理成功，位点才会提交到 Broker。

auto-offset-reset = earliest

当一个全新的 Consumer Group 首次接入，或者其原有的位点已在集群中过期淘汰时，该参数决定读取策略。earliest 表示从该分区当前尚存的最早位点开始回放；latest 则只接收启动之后新追加的消息。若该消费组在集群中已存在有效的提交记录，Kafka 会沿用现有进度，不会从头读取。

4.4 Topic 编排

@Configuration
public class KafkaTopicConfig {

    public static final String ORDER_TOPIC = "order-events";
    public static final String ORDER_DLT = "order-events-dlt";

    @Bean
    public NewTopic orderTopic() {
        return new NewTopic(ORDER_TOPIC, 3, (short) 1);
    }

    @Bean
    public NewTopic orderDltTopic() {
        return new NewTopic(ORDER_DLT, 3, (short) 1);
    }
}

通过 Spring 容器管理 NewTopic，应用在启动时会自动补全缺失的主题定义。分区数设为 3，当前消费组最多可由 3 个消费者并发消费。副本因子受限于单节点 Broker 环境，在此设置为 1。

第五章 消息生产者

5.1 生产端逻辑实现

@Service
public class OrderProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;

    public OrderProducer(KafkaTemplate<String, String> kafkaTemplate, ObjectMapper objectMapper) {
        this.kafkaTemplate = kafkaTemplate;
        this.objectMapper = objectMapper;
    }

    public void send(OrderEvent event) {
        try {
            // 1. 将事件统一转化为标准化 JSON 载荷
            String json = objectMapper.writeValueAsString(event);
            // 2. 路由键采用 orderId，确保同一个订单的数据落入固定分区
            String key = String.valueOf(event.orderId());

            // 3. 异步发送并挂载 CompletableFuture 回调
            kafkaTemplate.send(KafkaTopicConfig.ORDER_TOPIC, key, json)
                    .whenComplete((result, ex) -> {
                        if (ex != null) {
                            System.err.println("Kafka发送失败: " + ex.getMessage());
                            return;
                        }
                        var metadata = result.getRecordMetadata();
                        System.out.printf("Kafka发送成功 topic=%s partition=%d offset=%d%n",
                                metadata.topic(), metadata.partition(), metadata.offset());
                    });
        } catch (JsonProcessingException e) {
            throw new RuntimeException("订单事件序列化失败", e);
        }
    }
}

5.2 发送行为与模式对比

发送消息通常有三种交互方式：

1. 发后即忘（Fire-and-forget）：调用 send() 后不处理返回值。响应快，但网络故障时无法感知投递丢失，可靠性最弱。
2. 同步阻塞（Synchronous）：调用 send().get() 显式等待服务端应答。线程阻塞在网络 I/O 上，吞吐量低，也违背了事件驱动的初衷。
3. 异步回调（Asynchronous Callback）：使用 whenComplete() 注册监听逻辑。业务主线程向底层发送缓冲区追加消息后即可继续向下执行，底层的 I/O 线程会在收到 ACK 时异步触发相应回调，兼顾吞吐与可观测性。

5.3 控制器入口与测试设计

@RestController
@RequestMapping("/orders")
public class OrderController {

    private final OrderProducer orderProducer;

    public OrderController(OrderProducer orderProducer) {
        this.orderProducer = orderProducer;
    }

    @PostMapping
    public OrderEvent createOrder(@RequestBody CreateOrderRequest request) {
        OrderEvent event = new OrderEvent(
                UUID.randomUUID().toString(),
                request.orderId(),
                request.userId(),
                request.productId(),
                request.amount(),
                LocalDateTime.now()
        );
        orderProducer.send(event);
        return event;
    }

    @PostMapping("/replay")
    public OrderEvent replay(@RequestBody OrderEvent event) {
        // 原封不动重投指定事件，用于测试幂等判定与死信逻辑
        orderProducer.send(event);
        return event;
    }
}

第六章 消息消费者

6.1 消费端监听逻辑

@Component
public class OrderConsumer {

    private final ObjectMapper objectMapper;
    private final OrderProcessService orderProcessService;

    public OrderConsumer(ObjectMapper objectMapper, OrderProcessService orderProcessService) {
        this.objectMapper = objectMapper;
        this.orderProcessService = orderProcessService;
    }

    @KafkaListener(topics = KafkaTopicConfig.ORDER_TOPIC, groupId = "order-service-v2")
    public void consume(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        System.out.printf("收到消息 topic=%s partition=%d offset=%d key=%s%n",
                record.topic(), record.partition(), record.offset(), record.key());

        OrderEvent event;
        try {
            event = objectMapper.readValue(record.value(), OrderEvent.class);
        } catch (JsonProcessingException e) {
            // 静态畸形报文，反序列化必然永久失败，必须抛出非可重试异常直接推进死信流程
            throw new BadMessageException("订单消息 JSON 无法解析", e);
        }

        // 业务持久化处理
        ProcessResult result = orderProcessService.process(event);
        System.out.println("ProcessResult: " + result);

        // 业务处理无异常后，向外确认位点
        acknowledgment.acknowledge();
    }
}

6.2 异常传递与位点提交控制

编写 Consumer 时最常见的误区是随意包裹大粒度的 try-catch (Exception e) 并在 catch 块中仅打印一条日志后继续向下执行。这会导致异常被内部消化，Spring Kafka 容器认为本轮消费正常结束，进而触发 ACK 并推进 Offset，失败的消息就这样被悄悄跳过。

正确的做法是让业务异常向外抛出至容器层。Spring Kafka 拦截到未处理异常后，会中止本次位点确认，并将其转交给配置好的 DefaultErrorHandler 执行后续重试或降级分流。

第七章 重试与死信队列

7.1 异常的区分与分流策略

生产环境中的消费异常大致可归为两类：

• 瞬态故障（Transient Failure）：如数据库连接池打满、慢查超时、微服务网络毛刺。这些错误通常在数秒内具备自愈可能，应当使用退避策略原地重试。
• 确定性故障（Deterministic Failure）：如业务载荷字段缺失、JSON 字符串破损、数字反序列化溢出。这类输入无论尝试多少次结果均不会发生改变，盲目重试只会消耗系统资源，必须直接拦截并路由至死信队列。

7.2 错误处理器配置

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<String, String> kafkaTemplate) {
        // 1. 指定死信转储目标：追加 -dlt 后缀，并尽可能保留源分区编号
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
                kafkaTemplate,
                (record, exception) -> new TopicPartition(record.topic() + "-dlt", record.partition())
        );

        // 2. 固定退避：间隔 2000ms，最多追加重试 2 次（加上首次执行共计 3 次）
        FixedBackOff fixedBackOff = new FixedBackOff(2000L, 2L);

        DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, fixedBackOff);

        // 3. 注册不可重试异常黑名单
        errorHandler.addNotRetryableExceptions(BadMessageException.class);

        // 4. 当消息成功转入 DLT 后，必须提交原消息的消费进度，防止其重启后无限阻塞分区
        errorHandler.setCommitRecovered(true);

        // 5. 打印异常根因链
        errorHandler.setRetryListeners((record, exception, deliveryAttempt) -> {
            Throwable rootCause = exception;
            while (rootCause.getCause() != null) {
                rootCause = rootCause.getCause();
            }
            System.err.printf("消费失败 [Topic: %s, Partition: %d, Offset: %d, 尝试序号: %d] 根因: %s: %s%n",
                    record.topic(), record.partition(), record.offset(), deliveryAttempt,
                    rootCause.getClass().getSimpleName(), rootCause.getMessage());
        });

        return errorHandler;
    }
}

setCommitRecovered 的必要性

当一条消息耗尽所有重试机会并成功转储至 DLT 后，setCommitRecovered(true) 会通知容器自动提交该原消息在业务 Partition 的 Offset。若未开启此项，消费者实例一旦在后续某个时刻发生重平衡或重启，仍会从这条未提交的“毒消息”处重新拉取，再次陷入重试循环。

7.3 死信队列监听器

@Component
public class OrderDltConsumer {

    @KafkaListener(topics = KafkaTopicConfig.ORDER_DLT, groupId = "order-dlt-inspector")
    public void consume(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        System.err.printf("[死信告警] 捕获不可恢复事件 Key: %s, 内容: %s%n", record.key(), record.value());
        // 生产实践中，应在此处接入钉钉/企业微信报警通知，并将记录持久化至异常巡检表供人工介入
        acknowledgment.acknowledge();
    }
}

第八章 幂等消费与事件去重

8.1 为什么必须在业务层做幂等

在分布式环境下，Kafka 遵循 At-Least-Once（至少投递一次）语义。只要存在网络抖动，就无法完全避免重复消费。

配置了 enable.idempotence = true 的 Producer 只能确保同一个消息批次写入 Kafka 日志时不重复；一旦 Consumer 处理完毕，但位点因断网没能提交到 Kafka，这条记录在重新平衡或重启后必会被二次投递。

8.2 基于本地事务的去重实现

@Service
public class OrderProcessService {

    private final JdbcTemplate jdbcTemplate;
    private final FailureSimulator failureSimulator;

    public OrderProcessService(JdbcTemplate jdbcTemplate, FailureSimulator failureSimulator) {
        this.jdbcTemplate = jdbcTemplate;
        this.failureSimulator = failureSimulator;
    }

    @Transactional
    public ProcessResult process(OrderEvent event) {
        // 第一步：利用 INSERT IGNORE 抢占事件主键
        int inserted = jdbcTemplate.update(
                "INSERT IGNORE INTO processed_event (event_id, order_id, processed_at) VALUES (?, ?, ?)",
                event.eventId(),
                event.orderId(),
                Timestamp.valueOf(LocalDateTime.now())
        );

        // 主键已存在说明属于重复投递，立即平稳返回
        if (inserted == 0) {
            return ProcessResult.DUPLICATE;
        }

        // 第二步：故障模拟判定
        failureSimulator.maybeFail(event);

        // 第三步：核心业务落库
        jdbcTemplate.update(
                "INSERT INTO orders (order_id, user_id, product_id, amount, status, created_at) VALUES (?, ?, ?, ?, ?, ?)",
                event.orderId(),
                event.userId(),
                event.productId(),
                event.amount(),
                "CREATED",
                Timestamp.valueOf(event.createdAt())
        );

        return ProcessResult.SUCCESS;
    }
}

为什么去重记录与业务操作必须绑定在同一个 @Transactional 中？

假设将去重标志写入内存 Set，或在非事务环境下提前独立写入去重表：

1. 消费端拉取消息，在去重表中插入了记录；
2. 随后在执行业务落库或外部逻辑时抛出异常崩溃；
3. 本次消费流程失败中断，Kafka 触发稍后重试；
4. 重试拉取同一条消息再次执行，此时由于去重表中已存在记录，系统直接返回 DUPLICATE 并确认位点。

结果订单根本没有创建，这条业务数据就丢了。把 processed_event 和 orders 放进同一个本地事务后，业务异常回滚时去重占位记录会一起撤销；只有某次重试完整执行成功，两条记录才会一起提交。

第九章 故障模拟与实战验证

9.1 模拟器实现

@Component
public class FailureSimulator {

    private final ConcurrentHashMap<String, AtomicInteger> attempts = new ConcurrentHashMap<>();

    public void maybeFail(OrderEvent event) {
        int attempt = attempts.computeIfAbsent(event.eventId(), key -> new AtomicInteger(0))
                .incrementAndGet();

        // 订单号为 8888 时，前两次执行主动抛出运行时异常，模拟短时偶发故障
        if (event.orderId().equals(8888L) && attempt <= 2) {
            throw new RuntimeException("模拟临时故障，第 " + attempt + " 次失败");
        }

        // 订单号为 9999 时，无限抛出不可恢复异常
        if (event.orderId().equals(9999L)) {
            throw new RuntimeException("模拟永久业务异常");
        }
    }
}

9.2 场景验证

场景一：正常流程写入

curl -X POST http://localhost:8080/orders \
-H "Content-Type: application/json" \
-d '{"orderId":1001,"userId":1,"productId":100,"amount":99.9}'

控制台正常打印写入日志，MySQL 中 orders 与 processed_event 均生成相应记录，位点正常提交。

场景二：瞬时异常与退避自愈

curl -X POST http://localhost:8080/orders \
-H "Content-Type: application/json" \
-d '{"orderId":8888,"userId":1,"productId":100,"amount":50.0}'

• T = 0s：首次执行，触发异常，本地事务回滚，打印第 1 次失败；
• T = 2s：等待 2 秒退避后自动重试，再次触发异常回滚，打印第 2 次失败；
• T = 4s：等待 2 秒后发起第 2 次重试（第 3 次执行），模拟器放行，事务提交，向 Kafka 发起 ACK。数据最终一致。

场景三：永久异常耗尽重试入死信

curl -X POST http://localhost:8080/orders \
-H "Content-Type: application/json" \
-d '{"orderId":9999,"userId":1,"productId":100,"amount":50.0}'

经历初次执行与 2 次退避重试后，系统确认重试次数耗尽，DeadLetterPublishingRecoverer 捕获该事件并转发至 order-events-dlt。死信监听器输出告警，原队列位点向前推进。

场景四：重复投递与幂等拦截

先发送一笔正常订单并记录其返回报文中的 eventId：

curl -X POST http://localhost:8080/orders \
-H "Content-Type: application/json" \
-d '{"orderId":2001,"userId":2,"productId":200,"amount":199.9}'

将刚才的完整报文原样通过重放接口二次投递：

curl -X POST http://localhost:8080/orders/replay \
-H "Content-Type: application/json" \
-d '{"eventId":"<替换为上述实际ID>","orderId":2001,"userId":2,"productId":200,"amount":199.9,"createdAt":"2026-09-19T10:00:00"}'

控制台输出 ProcessResult: DUPLICATE，业务跳过实际 SQL 插入直接确认位点，MySQL 中并未产生重复订单行。

场景五：格式损坏直接分流

用 Kafka 自带的命令行生产者向 order-events 写入一条非 JSON 文本：

docker exec -it kafka kafka-console-producer \
--bootstrap-server localhost:9092 \
--topic order-events \
--property "parse.key=true" \
--property "key.separator=:"

输入非法载荷：

1002:this is not json

Consumer 捕获到 JsonProcessingException 并重新包裹为 BadMessageException 向上抛出。由于该异常被显式标记为不可重试，系统跳过退避等待，直接将其投递进 order-events-dlt。

第十章 生产环境最佳实践

10.1 轮询生命周期参数管理

max.poll.records

定义了消费者底层单次调用 KafkaConsumer.poll() 允许返回的最大数据条数（默认 500）。如果单次拉取数据过多，而每条业务处理耗时较长，整批任务的总耗时就可能超过 max.poll.interval.ms。因此耗时较长的任务应当调低该值。

max.poll.interval.ms

两次 poll 之间允许的最大时间跨度（默认 5 分钟）。如果某批记录处理过慢导致超过该阈值，Kafka 集群会判定该消费者工作线程已挂死，直接将其踢出 Consumer Group 并触发 Rebalance，导致该分区的任务被频繁重新指派并二次执行。

10.2 重平衡（Rebalance）与规避

Rebalance 是 Consumer Group 重新分配分区的过程。触发条件包括：

• 组内实例扩容、缩容或滚动发布重启；
• 某个实例因 Full GC 暂停或业务长时阻塞，导致心跳超时（session.timeout.ms）或轮询超时（max.poll.interval.ms）；
• 订阅的主题发生了分区扩容。

发生频繁重平衡期间，消费者需要挂起当前工作重新认领分区，易造成消费抖动与 Lag 攀升。规避手段包括设置合理的 GC 策略避免长暂停、调大心跳超时窗口、并将耗时过长的非核心业务（如第三方支付验证）异步化。

10.3 消费积压（Consumer Lag）的排查路径

当监控平台发出 Lag 持续增长告警时，排查次序通常如下：

1. 检查消费状态：确认工作节点是否存在大量错误重试日志，或者发生频繁 Rebalance；
2. 分析耗时瓶颈：排查数据库是否有慢查引发了线程池等待，或者外部 RPC 调用出现高延迟；
3. 核算并行度上限：如果所有消费者节点 CPU 与 I/O 正常，仅为纯粹的业务流量突增，应评估当前的分区数。若 Topic 仅有 3 个 Partition，把消费者实例加到 10 个也没有效果，需要先扩充 Partition 数量，再水平扩容消费者。

10.4 消息契约演进原则

生产系统的报文格式迟早要变更，演进时遵循以下约定：

• 只增不减：只允许向报文中新增可为空的字段，严禁删除已有字段或变更现有字段的物理类型；
• 宽容解析：消费端 ObjectMapper 应当启用反序列化忽略未知属性配置，避免生产端发出新属性时下游全线崩溃；
• 复杂体系契约治理：对于跨部门的大规模微服务架构，可逐步引入 Apache Avro 或 Protobuf，依托 Schema Registry 服务在集群服务端进行 Schema 兼容性强校验。

附录 A 常见问题排查

A.1 为什么消费端始终无法拉取到消息？

依次排查：

1. 执行 kafka-topics --list，确认投递的主题名称拼写是否一致；
2. 执行 GetOffsetShell 检查该主题分区的 LOG-END-OFFSET 是否在持续递增，确认消息是否真正写进了集群；
3. 执行 kafka-consumer-groups --describe --group <group-id>，核对当前消费组的分配拓扑，排查是否由于组内节点数量多于分区数导致当前节点处于空闲状态。

A.2 为什么修改了 group-id 后旧消息全部重新被消费了一遍？

当更换了全新的 group-id 时，Kafka 会认为这是一个完全未曾见过的消费群体。由于该群体在服务端不存在任何已提交的 Offset 记录，客户端将退回并应用 auto-offset-reset: earliest 的策略，从当前依然保存在日志中的最老位点从头回放全量数据。

A.3 为什么增加消费者节点后消费速率毫无变化？

Kafka 的分配约束是：同一个 Consumer Group 内，单个 Partition 在同一时刻只能指派给一个 Consumer。如果主题只有 3 个分区，那么组内最多只有 3 个消费者在并行工作。继续追加节点只会使其陷入无分区可分配的闲置状态，必须对 Topic 执行分区扩容才能解锁更高的并行度。

附录 B 关键配置速查表

B.1 核心概念边界判定

概念	定位	作用
orderId	业务主键 / Kafka Key	决定消息散列的分区，守住单订单维度的有序处理
eventId	操作唯一标识 (UUID)	存入去重表 processed_event，防止重复消费产生业务副作用
Partition	物理并发与存储单元	决定 Consumer Group 消费吞吐量的并发上限
Offset	单分区日志递增位点	供消费者标记当前分区的进度，作为故障恢复的基点

B.2 三种幂等与事务的边界区分

• Producer 幂等（enable.idempotence = true）：利用内部 PID 与 Sequence Number，消除生产客户端向 Broker 投递时因网络重试引发的日志重复写入。
• Consumer 幂等（processed_event 约束）：解决消费端在接收到多次重复投递时，防止业务执行（如扣款、插表）产生重复副作用。
• 本地事务（@Transactional）：确保“去重标记插入”与“订单业务插入”在本地数据库中一起提交或一起回滚。业务异常回滚时，去重记录同步消失，从而保证后续的重试得以正常重入。