MQ 开发篇

高清原图

先把业务消息语义和 Broker 运维分开

开发侧真正要解决的是怎样把 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 运维和流计算交给各自主线

不写 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_demo_0001",
  "eventType": "order.paid",
  "eventVersion": 1,
  "bizKey": "order_10001",
  "idempotentKey": "order_10001:paid:v1",
  "occurredAt": "<EVENT_TIMESTAMP>",
  "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-<RUN_ID>",
  "traceId": "trace-9f2c",
  "occurredAt": "<EVENT_TIMESTAMP>",
  "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 <EVENT_TIMESTAMP> --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