MQ 开发篇

高清原图

本文范围

本文讲的是开发侧怎么把 RabbitMQ、RocketMQ、Kafka 接进业务系统，重点是：消息什么时候该发、怎么保证不丢、不重复伤害业务、怎么处理顺序、事务消息、延迟消息、消费失败、死信、积压和补偿。

本文不展开部署、集群搭建、权限模型、Broker 参数、监控平台、备份恢复和服务器巡检。这些属于部署运维主线。开发同学要懂这些指标的含义，但不要在业务代码里替运维平台造一套集群管理系统。

前置条件

建议读者准备一个本地或测试环境的 MQ 实例，能创建 Topic、Queue、Consumer Group，并能看到队列堆积或消费者组位点。下面的命令和配置都按“开发验证”来写，生产环境要由平台侧给出地址、账号、网络和容量边界。

Java 项目侧建议准备 Spring Boot、Spring AMQP、RocketMQ 5.x Java SDK 或 Spring Kafka 中至少一种验证工程。所有示例都应该在测试环境先验证重复消费、消费失败、重试、死信和积压恢复，不要直接拿生产 Topic 做实验。

本文覆盖

• RabbitMQ、RocketMQ、Kafka 的开发使用边界和选型判断。
• 生产端可靠发送、消费者确认、手动提交、幂等表、本地消息表。
• 重复消费、顺序消息、事务消息、延迟消息、死信队列、积压恢复。
• 消费限速、下游保护、业务补偿和最终一致性验收。
• 开发排障：消息没发出、重复消费、消费失败、顺序错乱、积压暴涨、回放副作用。

本文不覆盖

• 不写 Broker 安装、集群部署、权限、证书、磁盘、监控平台搭建。
• 不展开 Kafka Streams、Flink、RocketMQ Streams 这类流计算。
• 不把 MQ 当分布式事务银弹；强一致库存、支付扣款、账户入账要先看业务事务边界。

关键结论

• MQ 默认按“至少一次”思维设计，消费者幂等是必选项。
• 顺序通常只做业务键局部顺序，不追全局顺序。
• 事务消息只解决“本地事务和消息发送”的一致，不替消费者做幂等和补偿。
• 重试不是限流，死信不是垃圾桶，积压恢复不能只靠盲目加消费者。
• 事件契约要像接口契约一样治理：版本、幂等键、回放策略、失败处理都要写清楚。

先把三套 MQ 的完整知识树摆出来

开发接 MQ 时不要只记“发消息、收消息”两个动作。真正出问题时，排障路径一定会落到路由、存储、确认、位点、重试和补偿上。先把三套产品的主链路记住，后面所有实践都围绕这张图展开。

产品	使用入口	核心对象	底层结构	失败时优先反推
RabbitMQ	RabbitTemplate、@RabbitListener、AMQP Channel	exchange、binding、queue、delivery tag、ack/nack、confirm、DLX	队列进程、未确认投递、队列 TTL、死信重新发布	路由不可达、confirm 未确认、unacked 堆积、requeue 热循环、DLX 目标不可用
RocketMQ	5.x Producer、PushConsumer、SimpleConsumer、消息类型 Topic	topic、queue、message group、message key、consumer group、retry、transaction checker	CommitLog、ConsumeQueue、消费进度、重试状态、事务半消息	Topic 类型不匹配、局部顺序阻塞、事务回查失败、重试进入 DLQ、消费进度不推进
Kafka	KafkaProducer、KafkaConsumer、Spring Kafka listener	topic、partition、record key、offset、consumer group、coordinator、ISR	分区追加日志、log segment、__consumer_offsets、副本同步	key 打散、自动提交过早、rebalance 重复、ISR 不足、lag 追赶压垮下游

上图解决“对象叫什么、消息会落到哪里”的问题；设计评审时还要继续追问每一跳有没有可观测证据。下面这张可靠链路对照图可以直接拿来做评审清单：生产端有没有闭环、Broker 有没有明确落点、消费端什么时候确认、失败后进入哪张表或哪个死信入口。

排障时也按这张图反推。比如“业务成功但下游没收到”，先看生产闭环：RabbitMQ 看 confirm 和 returns，RocketMQ 看发送结果或事务回查，Kafka 看 producer callback、acks 和 ISR；再看 Broker 落点和消费确认。如果证据链只到“代码调用了 send”，没有本地消息表、confirm/callback 或消费位点记录，可靠链路还没闭环。

源码和调试入口也按这个链路找。RabbitMQ 开发侧主要看 AMQP Channel#basicPublish、basicAck、basicNack、basicQos，以及 Spring AMQP 的 RabbitTemplate、监听容器和确认回调。RocketMQ 5.x 重点看消息类型、Producer#send、SimpleConsumer#receive/ack、事务检查器和消费进度；排底层时要能把消息定位到 Topic、queue、CommitLog 偏移和 ConsumeQueue 索引。Kafka 重点看 KafkaProducer#send、分区器、KafkaConsumer#poll、commitSync / commitAsync、ConsumerCoordinator、broker 侧分区日志和 group coordinator。

验证结果：你能从一条“订单支付成功”消息一路说到它进入哪个交换机、哪个队列或分区、哪个消费组、哪个幂等键、哪个补偿状态。如果只能说“发到 MQ 了”，这条链路还没有设计完。

选型先别吵，先看业务动作

场景说明：很多团队选 MQ 时喜欢从“谁吞吐高”开始争。开发落地时更该先问：消息是事件还是命令，是否允许延迟，是否需要局部顺序，是否需要延迟触发，是否需要灵活路由，是否需要历史回放。

直接做法：先按下面这个表选，不要一上来就把三个组件都接进系统。

场景	更常见选择	开发侧重点
传统业务队列、工作队列、灵活路由、DLX	RabbitMQ	exchange/queue/routing key、手动 ack、prefetch、DLX
订单、交易、事务消息、延迟、局部顺序	RocketMQ	message group、事务消息、消费重试、DLQ、消费进度
高吞吐日志、行为流、回放、分区并行	Kafka	key 分区、offset、consumer group、幂等生产者、手动提交

验证结果：你能用一句话说清楚“这条消息失败后谁负责补偿”。如果说不清，先别写生产者代码。

常见坑：把 MQ 当远程接口调用。消息是异步事件，调用方不能依赖消费者立刻完成。如果业务必须立即知道结果，用同步接口；如果只需要最终状态，用消息。

生产建议：核心链路要做事件评审。评审表至少包含事件名、生产时机、业务主键、幂等键、消费者、重试策略、死信处理、回放限制、保留时间和告警责任人。

先定义统一事件信封

场景说明：消息体随便塞字段，短期能跑，半年后最痛苦。消费者不知道字段能不能删，回放不知道会不会重复发券，排障不知道这条消息对应哪次业务操作。

直接做法：所有业务消息先套一层事件信封。

{
  "eventId": "evt_202607090001",
  "eventType": "order.paid",
  "eventVersion": 1,
  "bizKey": "order_10001",
  "idempotentKey": "order_10001:paid:v1",
  "occurredAt": "2026-07-09T10:15:30+08:00",
  "traceId": "trace-demo-001",
  "producer": "order-service",
  "payload": {
    "orderId": "order_10001",
    "paidAmount": 9900,
    "currency": "CNY"
  }
}

消费者落库一张幂等表。先插入幂等记录，再执行业务；如果唯一键冲突，说明这条消息已经处理过或正在处理。

CREATE TABLE mq_consume_record (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  consumer_group VARCHAR(128) NOT NULL,
  idempotent_key VARCHAR(191) NOT NULL,
  event_id VARCHAR(128) NOT NULL,
  status VARCHAR(32) NOT NULL,
  retry_count INT NOT NULL DEFAULT 0,
  last_error VARCHAR(1024),
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  UNIQUE KEY uk_consumer_idempotent (consumer_group, idempotent_key),
  KEY idx_event_id (event_id)
);

验证结果：重复投递同一条消息时，业务表只变化一次，幂等表能查到重复记录。

常见坑：只用 messageId 做幂等。很多业务真正要幂等的是“订单支付成功事件”，不是某个 MQ 产品生成的消息 ID。重发、转发、补偿、回放时，消息 ID 可能变化，业务幂等键必须由业务主键和事件语义组成。

生产建议：幂等表的状态不要只有 SUCCESS。至少保留 PROCESSING、SUCCESS、FAILED、TERMINATED，否则处理到一半宕机时无法判断是继续、重试还是人工介入。

RabbitMQ：手动 ack、confirm 和 DLX 是基本盘

场景说明：RabbitMQ 常见于传统业务队列、任务分发和灵活路由。开发上最容易翻车的是自动 ack、无限重回队列、没有 DLX、prefetch 太大导致单个消费者吞下太多未确认消息。

底层链路先记住：生产者在 AMQP channel 上把消息发布到 exchange，exchange 按 binding 和 routing key 把消息路由到 queue；如果路由不到队列，mandatory 和 returns callback 才有机会告诉生产者。消息进入队列后，消费者拿到的是一次 delivery，ack/nack 只是在确认这次 delivery 是否处理完成。publisher confirm 只覆盖“生产者到 Broker/队列 leader”的投递确认，consumer ack 只覆盖“Broker 到消费者”的处理确认，两者互相不知道对方。

直接做法：Spring Boot 项目里先把生产确认和消费确认打开。

spring:
  rabbitmq:
    host: mq.example.internal
    port: 5672
    username: app_user
    password: ${RABBITMQ_PASSWORD}
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 20
        default-requeue-rejected: false

声明业务队列和死信队列时，把死信交换机挂好。这里不写账号权限和集群策略，只写开发需要的队列关系。

@Configuration
class OrderPaidRabbitConfig {
    static final String EXCHANGE = "order.event.exchange";
    static final String QUEUE = "order.paid.benefit.queue";
    static final String DLX = "order.event.dlx";
    static final String DLQ = "order.paid.benefit.dlq";

    @Bean
    DirectExchange orderExchange() {
        return new DirectExchange(EXCHANGE, true, false);
    }

    @Bean
    DirectExchange deadLetterExchange() {
        return new DirectExchange(DLX, true, false);
    }

    @Bean
    Queue benefitQueue() {
        return QueueBuilder.durable(QUEUE)
            .withArgument("x-dead-letter-exchange", DLX)
            .withArgument("x-dead-letter-routing-key", "order.paid.benefit.dead")
            .build();
    }

    @Bean
    Queue benefitDeadQueue() {
        return QueueBuilder.durable(DLQ).build();
    }

    @Bean
    Binding benefitBinding() {
        return BindingBuilder.bind(benefitQueue())
            .to(orderExchange())
            .with("order.paid");
    }

    @Bean
    Binding benefitDeadBinding() {
        return BindingBuilder.bind(benefitDeadQueue())
            .to(deadLetterExchange())
            .with("order.paid.benefit.dead");
    }
}

生产者发送时，记录确认结果。confirm 失败不代表业务失败，但一定代表“消息投递到 Broker 的可靠性未闭环”，要进入本地消息表或补偿任务。

@Service
class OrderPaidRabbitPublisher {
    private final RabbitTemplate rabbitTemplate;

    OrderPaidRabbitPublisher(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        this.rabbitTemplate.setConfirmCallback((correlation, ack, cause) -> {
            if (!ack) {
                // 这里更新本地消息表为 SEND_FAILED，由定时补偿任务重投。
                System.err.println("rabbit confirm failed: " + cause);
            }
        });
        this.rabbitTemplate.setReturnsCallback(returned -> {
            // routing key 不可达时会进入这里。
            System.err.println("rabbit returned: " + returned.getRoutingKey());
        });
    }

    void publishOrderPaid(OrderPaidEvent event) {
        CorrelationData correlation = new CorrelationData(event.eventId());
        rabbitTemplate.convertAndSend(
            OrderPaidRabbitConfig.EXCHANGE,
            "order.paid",
            event,
            message -> {
                message.getMessageProperties().setMessageId(event.eventId());
                message.getMessageProperties().setHeader("idempotentKey", event.idempotentKey());
                return message;
            },
            correlation
        );
    }
}

消费者只在业务事务提交后 ack。失败时不要无脑 requeue，否则毒消息会反复冲击同一个消费者。

@RabbitListener(queues = OrderPaidRabbitConfig.QUEUE)
public void consume(OrderPaidEvent event, Message message, Channel channel) throws IOException {
    long tag = message.getMessageProperties().getDeliveryTag();
    try {
        consumeRecordService.runIdempotently(
            "benefit-service",
            event.idempotentKey(),
            () -> benefitService.grantBenefit(event.orderId())
        );
        channel.basicAck(tag, false);
    } catch (TransientException e) {
        // 不做立即 requeue，避免毒消息或下游故障形成热循环。
        // 生产上通常进入重试队列、DLX 或补偿任务，由外层控制间隔和次数。
        channel.basicNack(tag, false, false);
    } catch (Exception e) {
        // 业务不可恢复或达到重试上限，投递到 DLX。
        channel.basicNack(tag, false, false);
    }
}

验证结果：

rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers
rabbitmqctl list_bindings source_name destination_name routing_key

正常情况下，业务队列 messages_unacknowledged 不应长期增长；故意抛出不可恢复异常后，消息应进入 DLQ；重复投递时业务表只更新一次。

常见坑：

• 自动 ack 打开后，消费者刚拿到消息就宕机，消息可能已经被认为处理完成。
• basicNack(..., true) 无限重回队列，毒消息会一直阻塞正常消息。
• prefetch 过大，单个消费者堆住大量未确认消息，扩容消费者也看不到效果。
• TTL + DLX 做延迟队列时，过期消息要等到队头才会处理，不适合大量任意时间精确调度。

生产建议：RabbitMQ 的可靠性要同时看 publisher confirms 和 consumer ack。生产端确认只说明 Broker 已接收；消费端 ack 才说明业务处理完成。这两个动作彼此独立，不能互相替代。

RabbitMQ 底层现象怎么反推

如果 messages_ready 增长，说明消息在队列里没人处理或处理速度慢，先看消费者数量、消费者异常和下游耗时。如果 messages_unacknowledged 增长，说明消息已经投递给消费者但迟迟没有 ack，优先看业务处理卡住、线程池耗尽、prefetch 设置过大或 ack 写在了事务提交之前。

如果生产者 confirm 成功但消费者没收到，不要第一反应怀疑“消息丢了”。先查 exchange、binding、routing key 和队列名是否一致，再查是不是被死信、TTL 或长度限制转走。RabbitMQ 队列本身是有序集合，单 channel 发布到同一队列时能保持入队顺序；但多个连接、多个 channel 并发发布时，序列会交错，多个消费者并发处理也会让业务完成顺序变化。真正需要队列顺序时，要缩到单队列、单活跃消费者或业务状态机兜底。

DLX 也不是绝对安全网。死信本质上是 Broker 内部重新发布消息，目标交换机、目标队列或集群副本状态异常时仍然可能失败。核心链路如果把 DLQ 当作人工修复入口，除了声明 DLX 和 DLQ，还要让业务补偿表记录这条消息的 eventId、失败原因、重放次数和终止状态。这样 DLQ 不可用或被清理时，仍然能靠业务侧证据链恢复。

RocketMQ：把消息类型和业务语义绑紧

场景说明：RocketMQ 更常用于交易、订单、履约这类业务消息。它的顺序消息、事务消息、延迟消息和消费重试都很贴近业务，但也容易被滥用。

底层链路先记住：生产者先按 Topic 路由到 Broker，再写入 CommitLog；ConsumeQueue 更像按 Topic 和队列组织出来的消费索引，消费者按队列和 offset 拉取或接收消息。顺序、延迟、事务和重试不是魔法，它们都落在 Topic 类型、队列选择、消费进度和 Broker 对消息状态的管理上。

直接做法：RocketMQ 5.x 要先区分消息类型。一个 Topic 不要混发普通、顺序、事务、延迟消息，开发上按业务语义拆 Topic 更清晰。创建 5.x Topic 时要显式带上消息类型属性，普通 Topic 不能随便拿来发事务、FIFO 或延迟消息。

Message message = provider.newMessageBuilder()
    .setTopic("order_paid_fifo")
    .setKeys("order_10001")
    .setTag("paid")
    .setMessageGroup("order_10001") // 同一个订单进入同一个 FIFO 分组。
    .setBody(jsonBytes)
    .build();

SendReceipt receipt = producer.send(message);
System.out.println("messageId=" + receipt.getMessageId());

如果是延迟触发，例如支付后 30 分钟未履约检查，5.x 延迟消息按毫秒时间戳设置投递时间。

long deliveryTimestamp = Instant.now().plus(Duration.ofMinutes(30)).toEpochMilli();

Message delayMessage = provider.newMessageBuilder()
    .setTopic("order_fulfillment_delay")
    .setKeys("order_10001")
    .setDeliveryTimestamp(deliveryTimestamp)
    .setBody(jsonBytes)
    .build();
producer.send(delayMessage);

事务消息用于解决“本地事务成功但消息没发出去”的窗口。关键不是消费者怎么写，而是生产者的本地事务状态必须可回查。

// 伪代码：不同 SDK 封装会有差异，核心流程不变。
Transaction tx = producer.beginTransaction();
try {
    Message halfMessage = buildOrderPaidMessage(orderId);
    producer.send(halfMessage, tx);

    orderRepository.markPaid(orderId);
    localMessageRepository.markCommitted(eventId);

    tx.commit();
} catch (Exception e) {
    tx.rollback();
    throw e;
}

// 事务回查时，只查本地业务状态和本地消息状态，不靠内存变量。
TransactionResolution checkLocalTransaction(String eventId) {
    if (localMessageRepository.isCommitted(eventId)) {
        return TransactionResolution.COMMIT;
    }
    if (localMessageRepository.isRollback(eventId)) {
        return TransactionResolution.ROLLBACK;
    }
    return TransactionResolution.UNKNOWN;
}

验证结果：生产者进程在本地事务提交后、事务消息提交前异常退出时，Broker 回查后仍能根据本地表决定提交或回滚消息。

常见坑：

• 以为事务消息能保证消费者一定成功。它只保证生产者本地事务和消息发送最终一致，消费者仍然要幂等、重试和补偿。
• 顺序消息跨多个生产者并发发送，仍然可能破坏业务顺序。
• 用消费失败来做业务分流。RocketMQ 官方明确不建议把失败重试当业务流程控制，更不应该用失败重试限速。
• 延迟消息超过允许范围时可能立即投递，不能把超长业务预约完全压在 MQ 延迟消息上。

生产建议：顺序消息只按订单、账户、商品这类业务键局部有序。全局顺序基本等于把吞吐压成单通道，除非你真的能接受。

RocketMQ 底层现象怎么反推

RocketMQ 的“消息还在不在”不要只看消费者日志。消息写入后会进入 Broker 存储链路，消费成功通常只是推进消费进度或标记状态，不是马上从物理文件删除。排查时先确认 Topic 类型、queue、message key、consumer group 和消费进度，再看是否进入重试或死信。

顺序消息的核心是 messageGroup。同一个业务键进入同一 FIFO 分组后，Broker 会等消费者提交当前消息的消费结果；如果这条消息消费失败或超时，后面的同组消息会被挡住。线上看到“只有某些订单一直不推进”，不要盲目扩容消费者，先查是不是某个业务键的毒消息挡住了同组后续消息。

事务消息的核心是半消息和事务回查。Broker 先保存对消费者不可见的半消息，生产者本地事务完成后再提交 commit 或 rollback；如果第二阶段 ACK 丢了，Broker 会请求生产者集群回查本地事务状态。这里最容易翻车的是回查逻辑依赖内存变量、缓存或临时上下文。正确做法是只查本地业务表和本地消息表，能根据 eventId 或业务主键稳定判断提交还是回滚。

消费重试只能当保护措施。消费者抛异常、返回失败或处理超时，Broker 会按重试策略重新投递；达到上限后进入死信。它不适合做业务分流，也不适合拿来限速。限速应该在消费者线程池、令牌桶、下游调用并发和批量大小上控制，重试只负责“偶发失败后还能恢复”。

Kafka：key、offset 和手动提交是开发主线

场景说明：Kafka 更像分区日志，适合高吞吐、可回放、消费者组并行处理。开发上最容易踩坑的是自动提交 offset、重试导致乱序、key 设计不稳定、回放产生副作用。

底层链路先记住：生产者根据 record key 或分区器选择 partition，Broker leader 把 record 追加到该分区的 log segment，并分配单调递增的 offset；消费者组内的消费者被分配若干分区，处理完成后把位点提交给 group coordinator，位点最终写入内部的 __consumer_offsets。所以 Kafka 排障永远绕不开 partition、offset、consumer group、rebalance 和 ISR。

直接做法：生产者开启幂等，关键消息用 acks=all，同一业务键使用同一个 record key。

spring:
  kafka:
    bootstrap-servers: kafka.example.internal:9092
    producer:
      acks: all
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5
        delivery.timeout.ms: 120000
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      enable-auto-commit: false
      properties:
        isolation.level: read_committed
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
    listener:
      ack-mode: manual_immediate

这组参数要一起看，不能逐个孤立配置：

参数	建议基线	约束关系	验证方式
acks	关键业务用 all	leader 等 ISR 确认后返回，可靠性更强但延迟受副本状态影响	降低 ISR 或制造 broker 故障，确认生产端超时和告警
enable.idempotence	true	需要与 acks=all、retries>0、max.in.flight.requests.per.connection<=5 不冲突	模拟生产端网络闪断，确认没有因为重试写出重复 record
max.in.flight.requests.per.connection	幂等开启时不超过 5	幂等关闭且大于 1 时，失败重试可能改变同分区顺序	同 key 连续写入后制造首批请求失败，检查 offset 顺序
retries	保持开启，但受 delivery.timeout.ms 约束	它只处理可重试发送异常，不替代业务补偿	停掉 broker 后恢复，确认发送最终成功或进入本地消息表
delivery.timeout.ms	大于等于 request.timeout.ms + linger.ms	send() 返回后的总体投递上限，过小会让慢 ISR 场景提前失败	压测副本抖动时观察超时比例
max.poll.interval.ms	大于单批最坏处理时间	超过后消费者会被认为处理失联，引发 rebalance 和重复处理	故意让业务处理超过该值，确认幂等记录兜住重复
session.timeout.ms	覆盖 GC 和网络抖动	太小会因短抖动频繁 rebalance，太大又延迟故障摘除	观察 rebalance 日志、心跳失败和消费停顿窗口
isolation.level	事务链路消费者用 read_committed	只过滤 Kafka 事务中 abort 的记录，不代表外部数据库事务已提交	生产事务消息并 abort，确认消费者不可见

生产时把订单号、账户号这类局部顺序键作为 key。

kafkaTemplate.send("order-paid", event.bizKey(), event)
    .whenComplete((result, ex) -> {
        if (ex != null) {
            localMessageRepository.markSendFailed(event.eventId(), ex.getMessage());
            return;
        }
        localMessageRepository.markSent(
            event.eventId(),
            result.getRecordMetadata().partition(),
            result.getRecordMetadata().offset()
        );
    });

消费时，业务事务提交后再确认 offset。

@KafkaListener(topics = "order-paid", groupId = "benefit-service")
public void consume(OrderPaidEvent event, Acknowledgment ack) {
    consumeRecordService.runIdempotently(
        "benefit-service",
        event.idempotentKey(),
        () -> benefitService.grantBenefit(event.orderId())
    );
    ack.acknowledge();
}

验证结果：

kafka-consumer-groups.sh \
  --bootstrap-server kafka.example.internal:9092 \
  --describe \
  --group benefit-service

观察 CURRENT-OFFSET、LOG-END-OFFSET 和 LAG。处理成功后 lag 应下降；故意让消费者抛异常时 offset 不应提前提交。

常见坑：

• enable.auto.commit=true 时，业务还没处理完 offset 已提交，消费者宕机后消息不会再被当前组处理。
• enable.idempotence=false 且允许多个 in-flight 请求时，重试可能改变同一分区内的记录顺序。
• 只按随机 UUID 做 key，会把同一订单的事件打散到多个分区，局部顺序自然失效。
• 回放历史消息前不做副作用隔离，容易重复发短信、重复发券、重复推送。

生产建议：Kafka 的 exactly-once 语义主要约束 Kafka 内部生产和流处理链路。只要消费者写数据库、调 HTTP、发短信，业务端仍然需要幂等键、状态机和补偿。

Kafka 底层现象怎么反推

Kafka 的顺序只属于同一个 topic-partition。只要同一业务键被打到不同 partition，业务顺序就没了；只要同一个消费者组发生 rebalance，旧消费者处理到一半、新消费者接手同一分区，就可能出现重复处理。看到重复消费时，先查业务是否在 ack 前提交 offset，再查消费者是否因为 max.poll.interval.ms、进程重启、扩缩容或心跳异常触发了 rebalance。

offset 不是“消息是否处理过”的绝对真相，它只是某个消费者组在某个分区上的消费进度。业务成功但 offset 提交失败，下次还会再消费；offset 提交成功但业务事务失败，这条消息对当前组就可能被跳过。所以 Kafka 消费者要么把业务状态和 offset 交给同一个事务闭环，要么接受至少一次并用业务幂等兜底。

生产端 acks=all 和 ISR 相关。leader 要等同步副本确认后再返回成功，可靠性更强，但吞吐和延迟会受到副本状态影响。生产者幂等可以避免重试在 Kafka 日志中写出重复记录，但它不能替下游数据库、HTTP 调用或短信系统做业务幂等。

回放历史消息时更要谨慎。Kafka 保留的是日志，不是业务任务队列；同一个消费者组改 offset 会影响这个组后续读取位置。生产上回放建议使用独立消费者组、限定时间窗口、先 dry-run 统计数量，再让消费者按业务状态过滤副作用。

本地消息表：让发送失败可恢复

场景说明：最危险的窗口是业务事务提交了，但消息发送失败或进程崩溃。事务消息可以解决一部分场景，但不是所有 MQ 都支持，也不是所有团队都愿意引入。通用做法是本地消息表。

直接做法：业务事务内写业务表和本地消息表，事务提交后异步投递。

这张 Outbox 图专门用来审“本地事务成功、消息发送失败”的窗口。评审时不要只看有没有 outbox_event 表，还要看状态流转、并发抢占、发送成功后更新状态失败时是否允许重发，以及消费者幂等是否能兜住重复投递。

用它排障时，先查 event_id 在业务表、outbox_event、MQ 发送日志和消费幂等表里的状态是否一致。关闭 MQ 做故障演练时，预期不是业务回滚，而是业务事务提交、outbox_event 停在 NEW 或 RETRY；恢复后任务继续投递，消费者靠幂等保证重复发送不会重复扣减、重复发券或重复通知。

CREATE TABLE outbox_event (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  event_id VARCHAR(128) NOT NULL,
  event_type VARCHAR(128) NOT NULL,
  biz_key VARCHAR(191) NOT NULL,
  payload JSON NOT NULL,
  status VARCHAR(32) NOT NULL,
  retry_count INT NOT NULL DEFAULT 0,
  next_retry_at DATETIME NOT NULL,
  last_error VARCHAR(1024),
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  UNIQUE KEY uk_event_id (event_id),
  KEY idx_status_retry (status, next_retry_at)
);

@Transactional
public void paySuccess(String orderId) {
    orderRepository.markPaid(orderId);
    outboxRepository.insert(OrderPaidEvent.from(orderId));
}

@Scheduled(fixedDelay = 3000)
public void flushOutbox() {
    List<OutboxEvent> events = outboxRepository.lockNextBatch("NEW", 100);
    for (OutboxEvent event : events) {
        try {
            mqPublisher.publish(event);
            outboxRepository.markSent(event.eventId());
        } catch (Exception e) {
            outboxRepository.markRetry(event.eventId(), e.getMessage());
        }
    }
}

验证结果：关闭 MQ 后执行业务，业务事务仍能提交，本地消息表为 NEW 或 RETRY；恢复 MQ 后，补偿任务把消息投递出去。

常见坑：补偿任务没有并发锁，多实例同时扫表重复发送。解决方式是按数据库能力使用 select ... for update skip locked、状态抢占或分片任务，消费者侧仍然保持幂等。

生产建议：本地消息表要设置最大重试次数和人工处理入口。无限重试会把数据库和 MQ 都拖慢。

本地消息表的状态也要能表达完整生命周期，建议至少包含 NEW、SENDING、SENT、RETRY、FAILED、TERMINATED。多实例投递时，先用数据库行锁、状态抢占或分片任务拿到处理权，再发消息。发送成功但更新状态失败时，下一轮可能重发，所以消费者幂等仍然不能省。

最小可验证实验：上线前主动制造故障

不要只跑“发一条消息、消费成功”的 happy path。MQ 链路上线前至少做下面五个小实验，实验越早做，线上越少靠运气。

实验	做法	通过标准
重复消费	同一个 idempotentKey 连续投递 10 次，或手动重放同一条消息	业务表只变化一次，幂等表能记录重复拦截
消费中断	消费者执行业务后、ack 或 offset 提交前强制退出	消息会被再次投递，业务不会重复产生副作用
顺序乱投	对同一订单先投 SHIPPED，再投 PAID	状态机拒绝越级流转，补偿任务可恢复
延迟触发	构造 1 分钟延迟和超出产品允许范围的延迟	到达时间、立即投递或拒绝行为与预期一致
积压恢复	暂停消费者制造积压，再按限速恢复	lag 下降，下游 RT 和错误率不被打爆

这些实验不用等到压测平台完备才做。开发环境里能验证语义，测试环境里能验证并发，预发环境里能验证告警和人工处理入口。只要其中一个实验说不清结果，就不要把这条消息链路放进核心交易流程。

重复消费：不要祈祷，只做幂等

场景说明：生产者重试、Broker 重投、消费者宕机、offset 提交失败、ack 丢失、人工回放，都会带来重复消费。

直接做法：消费者入口统一套幂等执行器。

public void runIdempotently(String group, String key, Runnable business) {
    boolean inserted = consumeRecordRepository.tryInsert(group, key, "PROCESSING");
    if (!inserted && consumeRecordRepository.isSuccess(group, key)) {
        return;
    }
    if (!inserted) {
        throw new RetryLaterException("same event is processing: " + key);
    }

    try {
        business.run();
        consumeRecordRepository.markSuccess(group, key);
    } catch (Exception e) {
        consumeRecordRepository.markFailed(group, key, e.getMessage());
        throw e;
    }
}

验证结果：同一 idempotentKey 连续投递 10 次，业务结果只出现一次，消费记录能看到重复拦截。

常见坑：幂等表和业务表不在同一个事务里。业务成功但幂等状态没成功，下一次重复消息又会执行业务。能放同库同事务时尽量放；跨库时用业务唯一约束兜底。

生产建议：幂等不等于直接忽略重复请求。对于支付、库存、权益这类业务，要返回当前业务状态，而不是简单吞掉异常。

顺序消息：先缩小顺序范围

场景说明：订单状态需要按 created -> paid -> shipped -> finished 推进，但不代表全站所有订单都要全局排队。

直接做法：

• RabbitMQ：同一业务键尽量进入同一队列，消费者并发要谨慎；更复杂的局部顺序通常不优先用 RabbitMQ 做。
• RocketMQ：同一业务键设置同一个 messageGroup。
• Kafka：同一业务键作为 record key，保证进入同一分区。

消费者侧再用状态机兜底。

UPDATE order_state
SET status = 'PAID', version = version + 1
WHERE order_id = 'order_10001'
  AND status = 'CREATED';

验证结果：乱序投递 SHIPPED 再投递 PAID，SHIPPED 更新应失败并进入等待补偿，而不是把订单状态直接跳过去。

常见坑：只相信 MQ 顺序，不校验业务前置状态。MQ 只能保证某个队列或分区内的消息顺序，业务状态仍然要自己保护。

生产建议：顺序消息遇到毒消息会拖住同一顺序组。要准备“跳过并补偿”的人工流程，不能让一个坏订单堵住一批正常订单。

延迟消息：适合短期触发，不适合长期业务预约

场景说明：支付超时关单、下单后延迟检查、通知失败后稍后重试，这些适合延迟消息。三个月后的会员续费提醒、长期预约任务，不建议只靠 MQ 延迟。

直接做法：

• RabbitMQ：常见做法是 TTL + DLX 或延迟消息插件，但要确认插件和队列行为。
• RocketMQ 5.x：设置投递时间戳，默认最大延迟范围为 24 小时。
• Kafka：本身不是延迟队列，通常用时间轮、延迟 Topic、数据库任务表或调度服务转发。

验证结果：构造 1 分钟延迟消息，观察消费者收到时间和业务预期误差；再构造超过允许范围的消息，确认是否被立即投递或拒绝。

常见坑：把延迟消息当精确定时器。Broker 负载、队列堆积、Topic 设计都会影响触发时间。

生产建议：延迟消息触发后仍要查业务当前状态。比如关单消息到达时，先查订单是否已支付，不要直接关单。

重试、死信和补偿要分层

场景说明：消费失败可能是下游临时超时，也可能是消息格式错误，还可能是业务状态永久不允许。三类问题不能用同一种重试策略。

直接做法：按错误类型处理。

错误类型	处理方式	验证方式
网络抖动、下游短暂不可用	有间隔重试	下游恢复后自动成功
参数缺失、事件版本不兼容	进入死信	DLQ 有记录，人工修复后可重放
业务状态暂不满足	进入补偿表或延迟重试	状态满足后补偿成功
下游限流	消费端限速，不靠失败重试限流	lag 可控，下游错误率下降

补偿表建议单独建，不要只依赖 MQ 的 DLQ。

下面这张状态机把重复消费、顺序前置、重试、死信和补偿放到同一条业务处理记录上。它适合放进设计评审和排障手册：现场只要知道当前消息卡在哪个状态，就能判断是该 ack、延迟重试、进 DLQ、建补偿任务，还是人工终止。

生产排障时不要只问“这条消息为什么失败”，而要查幂等记录、业务状态、重试次数、DLQ 记录和补偿任务年龄。重复投递如果已经到 ACKED，直接返回当前业务状态；顺序乱投如果落到 WAIT_COMPENSATION，先等前置状态或走补偿；毒消息进入 DEAD_LETTER 后必须有人修复、重放或终止，不能让消费者线程无休止重试。

Kafka 要额外注意 retry topic、DLT 和 offset 的关系。Kafka 本身是分区日志，不会像传统队列那样天然给每条消息挂一个“延迟重试状态”。常见做法是消费者捕获可恢复异常后，把原始消息和失败原因发到 order-paid.retry.1m、order-paid.retry.5m 这类重试 Topic；超过次数后发到 order-paid.dlt。只有当“原始消息已经进入重试/DLT 或业务补偿表”这个动作成功后，才提交原始分区 offset，否则会出现既没重试记录又跳过原始消息的空洞。

手动 seek 重试适合本地短暂异常，不适合长期卡住同一分区。一个毒消息如果一直 seek 回当前 offset，同分区后续消息都会被挡住，lag 会越来越大。更稳的边界是：

场景	Kafka 处理方式	业务要求
下游 1 秒级抖动	当前 listener 内有限次数退避重试	不能超过 max.poll.interval.ms
下游分钟级不可用	发 retry topic，提交原始 offset	retry 消息带原 topic、partition、offset、失败原因
payload 版本不兼容	发 DLT，提交原始 offset	DLT 有负责人、修复入口和重放脚本
业务前置状态未满足	写补偿表或延迟 Topic	重放前重新查业务当前状态

验证方式：构造一条必然失败的 Kafka 消息，观察原 consumer group lag、retry topic 数量、DLT 数量和幂等表状态。成功标准不是“日志里打印了异常”，而是原始 offset 不会无限卡住、失败消息可定位、重放不会绕过幂等。

CREATE TABLE mq_compensation_task (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  event_id VARCHAR(128) NOT NULL,
  task_type VARCHAR(64) NOT NULL,
  biz_key VARCHAR(191) NOT NULL,
  status VARCHAR(32) NOT NULL,
  next_execute_at DATETIME NOT NULL,
  retry_count INT NOT NULL DEFAULT 0,
  last_error VARCHAR(1024),
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  KEY idx_status_next (status, next_execute_at)
);

验证结果：手动制造参数错误，消息进入死信；修复 payload 后从补偿入口重放，业务成功且幂等记录正确。

常见坑：死信队列没人看。没人处理的 DLQ 只是换了个地方堆积。

生产建议：重试次数、死信数量、补偿任务年龄都要有业务告警责任人。监控平台怎么搭归部署运维，但“谁处理这条坏消息”必须由业务团队定义。

积压恢复：先保护下游，再追 lag

场景说明：消费者挂了 2 小时，恢复后发现积压几十万条。最危险的动作是立刻把消费者扩到几十个实例，把数据库打满。

直接做法：按四步恢复。

Kafka 可以用消费者组命令看 lag，RabbitMQ 看 ready/unacked，RocketMQ 看消费进度和 lag 指标。开发排障时至少要能拿到“积压量、消费速率、失败率、下游耗时”四个数。

kafka-consumer-groups.sh --bootstrap-server kafka.example.internal:9092 \
  --describe --group benefit-service

rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers

验证结果：恢复过程中，下游数据库连接池不打满，接口 P95 不持续恶化，lag 稳定下降。

常见坑：

• 只看积压量，不看失败率。毒消息导致的积压，加消费者没用。
• 只看消费者进程存活，不看业务成功数。
• 回放时没有按业务时间窗口和事件类型分批。

生产建议：消费者要有可调并发和限速开关。Spring Kafka 可以调容器并发，RabbitMQ 可以调 prefetch，RocketMQ 要结合消费者线程池、消费组策略和业务限流。开关值由压测给出，不要故障时现猜。

事件契约版本治理：消息也要有 API 生命周期

场景说明：MQ 事故里很大一类不是 broker 坏了，而是生产者把字段改名、删除、换语义，消费者还在按旧结构反序列化；或者历史消息、死信消息、延迟消息在新代码上线后被重新消费，触发了旧版本 payload 与新业务逻辑的冲突。事件一旦发布，就要按 API 契约治理。

直接做法：每个事件都登记契约，至少包括事件名、版本、owner、生产者、消费者、幂等键、schema、保留时间、回放策略和 DLQ 责任人。

{
  "eventType": "order.paid",
  "eventVersion": 2,
  "eventId": "evt-20260710-0001",
  "traceId": "trace-9f2c",
  "occurredAt": "2026-07-10T10:15:30+08:00",
  "bizKey": "order_10001",
  "idempotentKey": "benefit-service:order_10001:paid:v2",
  "payload": {
    "orderId": "order_10001",
    "paidAmount": 9900,
    "currency": "CNY"
  }
}

兼容规则：

• 只兼容新增可选字段；字段改名、删除、类型变更、枚举含义变更都按破坏性变更处理。
• 破坏性变更优先新建 eventType，或者显式提升 eventVersion 并双写一段兼容期。
• 消费者必须忽略未知字段，并对缺失的新字段提供默认逻辑或拒绝进入 DLQ。
• 生产者在兼容期内保留旧字段，直到所有消费者、回放任务和死信重放入口确认完成迁移。
• 契约样例要进 CI：生产者样例消息能被所有消费者反序列化，消费者也要用旧版本样例跑兼容测试。

验证结果：准备 v1、v2 两份样例消息，分别投到测试 Topic 或测试队列；消费者能成功处理兼容新增字段，遇到破坏性变更能进入明确的死信或补偿流程，而不是在业务线程里反复重试。回放前用 dry-run 统计事件版本分布，确认老版本事件仍有处理分支。

常见坑：

• 只给 HTTP API 做版本治理，MQ payload 当成内部对象直接序列化。
• 使用 Java 类全限定名、数据库字段名或枚举 ordinal 作为消息契约，导致重构直接破坏消费。
• 死信重放时使用当前最新代码，但没有覆盖旧版本 payload。
• 用“所有消费者都升级了”作为口头结论，没有消费者清单、版本清单和回放验证。

生产建议：事件契约要有 owner 和下线流程。新增字段、弃用字段、事件重命名、回放窗口调整，都要能找到负责人。核心事件建议保留样例消息、schema、兼容性测试和重放手册；DLQ 告警里要带 eventType、eventVersion、eventId 和失败原因，方便按契约定位。

落地工程深水区

幂等和状态机要一起设计

只做幂等表不能解决所有问题。订单状态推进、库存扣减、权益发放要有状态机和唯一约束。判断标准很简单：重复消息、乱序消息、回放消息同时出现时，业务最终状态仍然正确。

上线检查：

-- 检查同一消费者组是否有重复成功记录。
SELECT consumer_group, idempotent_key, COUNT(*)
FROM mq_consume_record
WHERE status = 'SUCCESS'
GROUP BY consumer_group, idempotent_key
HAVING COUNT(*) > 1;

重试不能拖垮下游

消费失败后立刻高频重试，会把下游故障放大。重试间隔要递增，失败次数要封顶，超过上限进入死信或补偿。RocketMQ 官方也明确把重试定位成业务失败的保护措施，不应用来做业务分流或限速。

积压恢复要有容量上限

提前压测出单消费者成功处理速率、下游数据库最大安全写入速率、单条消息平均耗时。故障恢复时按这些数限速。

安全消费并发 = min(消费者 CPU 可承载并发, 下游 QPS 上限 / 单条消息下游调用次数)
恢复耗时 = 积压消息数 / 每秒成功消费数

事件版本要兼容

消息一旦发出去，就可能被延迟消费、死信重放、历史回放。字段只能做兼容新增，不要随意改名或删除。消费者要按 eventVersion 做兼容处理，并把旧版本样例纳入回归测试。

补偿要能人工接管

最终一致不是“最终会好”。它需要对账、补偿、人工终止和复盘入口。核心链路至少要能按 eventId、bizKey、traceId 查到当前在哪一步失败。

排障手册

现象	优先检查	常见原因	修复方式
业务成功但下游没收到	本地消息表、confirm 回调、发送异常日志	发送失败、路由错误、Topic/Queue 名错	重投本地消息，修正路由
消费重复执行	幂等表、业务唯一索引	ack/offset 提交失败、生产者重试、回放	补幂等键和状态机
顺序错乱	key/messageGroup/routing key、消费者并发	同一业务键打到不同分区或队列	固定业务键路由，状态机兜底
消费一直失败	消费者异常栈、DLQ、消息版本	毒消息、字段不兼容、下游不可用	修复后重放或进入补偿
积压暴涨	lag、消费速率、下游 RT	消费慢、下游慢、毒消息阻塞	限速恢复、隔离坏消息、分批追赶
回放后副作用重复	幂等记录、业务状态	回放前未过滤已完成动作	回放前按状态筛选，消费者幂等

命令速查

# Kafka：查看消费者组 lag
kafka-consumer-groups.sh --bootstrap-server kafka.example.internal:9092 \
  --describe --group benefit-service

# Kafka：按时间定位 offset，具体参数按 Kafka 版本和脚本帮助确认
kafka-consumer-groups.sh --bootstrap-server kafka.example.internal:9092 \
  --group benefit-service --topic order-paid --reset-offsets --to-datetime 2026-07-09T10:00:00.000 --dry-run

# RabbitMQ：查看队列 ready/unacked
rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers

# RabbitMQ：查看绑定关系
rabbitmqctl list_bindings source_name destination_name routing_key

# RocketMQ：查看消费进度和堆积，参数以当前版本 mqadmin help 为准
mqadmin consumerProgress -n nameserver.example.internal:9876 -g benefit-service

# RocketMQ：按业务 key 查消息，适合确认消息是否写入 Broker
mqadmin queryMsgByKey -n nameserver.example.internal:9876 \
  -t order_paid_fifo -k order_10001

落地检查清单

• 事件名、版本、业务主键、幂等键是否明确。
• 生产端是否有 confirm、事务消息或本地消息表兜底。
• 消费端是否关闭自动确认，业务成功后才 ack 或提交 offset。
• 重复消费是否不会重复扣减、重复发券、重复通知。
• 顺序要求是否缩小到业务键局部顺序。
• 延迟消息是否只用于短期触发，到达后是否二次校验业务状态。
• 重试是否有间隔、次数上限和死信出口。
• 死信是否有人工修复、重放和终止入口。
• Kafka retry topic / DLT 是否和 offset 提交边界一起设计，是否不会跳过失败消息。
• 积压恢复是否有下游限速和分批方案。
• 回放是否先评估副作用，是否按事件版本兼容。
• 部署、权限、集群和监控平台是否交给部署运维主线，不在业务代码里硬编码。

官方文档入口

• RabbitMQ Publishers
• RabbitMQ Consumers
• RabbitMQ Queues
• RabbitMQ Consumer Acknowledgements and Publisher Confirms
• RabbitMQ Dead Letter Exchanges
• RabbitMQ Time-To-Live and Expiration
• Apache RocketMQ Concepts
• Apache RocketMQ Parameter Constraints and Suggestions
• Apache RocketMQ Basic Best Practices
• Apache RocketMQ Ordered Message
• Apache RocketMQ Delay Message
• Apache RocketMQ Transaction Message
• Apache RocketMQ Consumption Retry
• Apache Kafka 4.3 Introduction
• Apache Kafka 4.3 Producer Configs
• Apache Kafka 4.3 Consumer and Share Consumer Configs
• Apache Kafka 4.3 Log Implementation
• Apache Kafka 4.3 Distribution and Consumer Offset Tracking
• Spring for Apache Kafka：Handling Exceptions