Day18 消息队列 MQ：RabbitMQ / RocketMQ，异步解耦、削峰、事务消息、死信队列

📖 基础知识讲解【基础原理 + 高级深挖知识点】

基础原理

1. MQ 到底解决了什么问题：异步、解耦、削峰

消息队列的本质是在生产者和消费者之间插入一个可靠的中间缓冲区，把「同步的、强依赖的、瞬时的」调用变成「异步的、弱依赖的、可堆积的」消息传递。它带来的收益只有三条，面试被问到「为什么用 MQ」时，答案不要超出这三条：

作用	本质	典型场景	不用 MQ 的后果
异步	把非核心链路从主流程摘出去，缩短响应时间	下单后发短信、推送、加积分、写日志	主流程 RT 被下游拖长，下游超时直接影响下单成功率
解耦	生产者只认消息格式，不认下游有几个、是谁	订单创建事件被库存、营销、风控、数仓同时消费	每接一个下游就要改订单服务代码并重新发版
削峰	用队列堆积把瞬时洪峰摊平成平稳流量	秒杀、抢券、大促开门红	瞬时并发直接打到数据库，连接池耗尽、服务雪崩

三者的区别可以用一句话记住：异步是为了「快」，解耦是为了「省改动」，削峰是为了「不挂」。

text

同步调用：
用户 ──► 订单服务 ──► 库存服务(80ms) ──► 积分服务(50ms) ──► 短信服务(200ms)
         总 RT = 20 + 80 + 50 + 200 = 350ms，任一下游挂掉则下单失败

引入 MQ：
用户 ──► 订单服务 ──► 写库(20ms) ──► 发消息(2ms) ──► 返回
                                        │
                                        ▼
                                    ┌───────┐
                                    │  MQ   │
                                    └───┬───┘
                        ┌───────────────┼───────────────┐
                        ▼               ▼               ▼
                     库存服务        积分服务        短信服务
         总 RT = 22ms，下游挂掉只是消息堆积，不影响下单

⚠️ 削峰不是把流量变小，而是把流量「拉长」。原本 1 秒 10 万请求，削峰后变成 100 秒每秒 1000 请求。如果业务要求用户 1 秒内必须看到结果，削峰是做不到的——这时只能扩容或者限流，不能指望 MQ。

面试加分表述：「我判断一个场景能不能用 MQ，看的是调用方是否接受最终一致。能接受就异步化，不能接受（比如扣库存判断是否超卖）就必须保持同步。MQ 是用一致性的时效换可用性和吞吐，不是万能的性能药。」

2. 引入 MQ 的代价：必须能回答的四个问题

MQ 是典型的「引入一个组件，带来四个新问题」。面试官问「为什么用 MQ」之后，80% 的概率紧接着问代价：

系统可用性降低：MQ 本身成了核心链路的一部分，MQ 挂了全链路瘫痪。所以生产环境 MQ 必须集群部署，且发送失败要有本地降级兜底（落库后定时补偿）。
系统复杂度上升：原来一个同步调用能解决的事，现在要考虑消息丢失、重复、乱序、堆积，还要维护死信、补偿、对账。
一致性问题：主流程写库成功、发消息失败，或者消息发了但消费者执行失败，都会导致上下游数据不一致。这是事务消息 / 本地消息表存在的根本原因。
运维与排障成本：链路从「一条调用栈」变成「两段异步」，排查问题必须依赖消息轨迹和全链路 traceId。

面试加分表述：「我们团队的规范是能不用 MQ 就不用。只有满足三个条件之一才引入：下游超过 2 个且还会继续增加、下游 RT 超过主流程的 30%、或者存在明确的瞬时洪峰。因为 MQ 带来的隐性成本（对账、补偿、幂等）通常比写几个异步线程高得多。」

3. 四大主流 MQ 选型对比

维度	RabbitMQ	RocketMQ	Kafka	Pulsar
开发语言	Erlang	Java	Scala / Java	Java
单机吞吐	万级	十万级	十万级（最高）	十万级
时延	微秒级（最低）	毫秒级	毫秒级（批量聚合）	毫秒级
可用性	镜像队列 / Quorum	主从 + DLedger	ISR 多副本	BookKeeper 多副本
消息可靠性	高（confirm + 持久化）	非常高（可零丢失）	高（acks=all）	非常高
功能完备度	路由灵活、插件丰富	事务消息 / 延时 / 顺序 / 回溯齐全	流处理生态强	存算分离、多租户
顺序消息	单队列单消费者才有序	原生支持分区顺序	分区内有序	分区内有序
延时消息	需 DLX 或插件	原生支持	不支持（需外部实现）	原生支持任意延时
事务消息	有事务但性能极差	原生半消息 + 回查	有事务（面向流）	支持
消息回溯	不支持	支持（按时间 / offset）	支持（按 offset）	支持
Topic 数量影响	Topic 多性能下降	几万 Topic 性能稳定	Topic / 分区多时性能明显下降	分区多较稳定
典型场景	业务解耦、复杂路由、金融级低延时	电商交易、订单、事务一致性	日志、埋点、大数据管道	云原生、多租户

选型结论（面试直接给判断，不要罗列）：

业务系统（订单、交易、库存）优先 RocketMQ：事务消息、延时消息、顺序消息、消息回溯四件套原生支持，Java 栈排查方便。
路由规则复杂、消息量不大、要求极低时延 → RabbitMQ：Exchange 的四种路由模式是它最大的差异化能力。
日志 / 埋点 / 流计算 → Kafka：批量 + 顺序写 + 零拷贝带来最高吞吐，但它的设计目标不是「每条消息都不能丢的业务消息」。
Kafka 的致命短板是 Topic 数量：Kafka 每个分区一个物理文件，分区数上千后随机 IO 严重；RocketMQ 所有 Topic 共用一个 CommitLog，所以能撑几万 Topic。

⚠️ 不要说「Kafka 会丢消息」。Kafka 配好 acks=all + min.insync.replicas=2 + 关闭 unclean 选举同样不丢。差异在于默认配置的取向和功能完备度，而不是能力上限。

4. AMQP 协议与 RabbitMQ 的核心模型

RabbitMQ 是 AMQP 0-9-1 协议的实现，它的模型比其他 MQ 多了一层 Exchange（交换机），这是理解 RabbitMQ 的关键。

text

                         ┌──────────────── Broker（RabbitMQ 服务端）────────────────┐
                         │                                                          │
                         │   ┌─────────── Virtual Host（逻辑隔离单元）──────────┐    │
 Producer ──publish──────┼──►│  Exchange ──binding(routingKey)──► Queue ────────┼───┼──► Consumer
   │                     │   │     │                                 │          │    │       │
   │  Connection(TCP)    │   │     └──binding──► Queue(另一个)        │          │    │       │
   │   └─ Channel 1      │   │                                       │          │    │    ACK/NACK
   │   └─ Channel 2      │   └───────────────────────────────────────┘          │    │       │
   │                     └──────────────────────────────────────────────────────┘    │
   └─────────────────────────────── confirm 回调 ◄──────────────────────────────────┘

核心组件逐个说清：

组件	职责	必须记住的点
Producer	发消息，只把消息投给 Exchange	生产者不知道队列存在，这是解耦的来源
Connection	和 Broker 之间的 TCP 长连接	建连成本高（TCP + TLS + 认证），应用内复用
Channel	Connection 上的逻辑信道	真正的操作单元，所有 publish / consume 都走 Channel；Channel 非线程安全
Virtual Host	逻辑隔离，各自独立的 Exchange / Queue / 权限	多环境或多业务线隔离用，默认是 `/`
Exchange	按 routingKey 和 binding 规则把消息路由到队列	不存消息，路由不到就丢弃（或走 alternate-exchange）
Binding	Exchange 与 Queue 的绑定关系，带 bindingKey	一个 Exchange 可绑多个队列，一个队列可被多个 Exchange 绑
Queue	真正存消息的地方，FIFO	消息只会被同一队列上的一个消费者拿到（集群内竞争）
Consumer	订阅队列，拿到消息并 ACK	ACK 决定消息是否从队列删除

面试加分表述：「RabbitMQ 和 Kafka / RocketMQ 最本质的模型差异是：RabbitMQ 的消息路由发生在服务端的 Exchange，所以能做复杂的路由和过滤；而 Kafka / RocketMQ 的 Topic 是直通的，过滤要么在 Broker 用 TAG，要么在消费端做。这决定了 RabbitMQ 更适合「一条消息要按不同规则分发到不同系统」的业务编排场景。」

5. Connection 与 Channel：为什么不是一个线程一个连接

这是 RabbitMQ 最常被问、也最容易在生产上踩坑的点。

Connection 是 TCP 连接，建立成本高：三次握手 + TLS 握手 + AMQP 协议协商 + 认证，一次大约几十毫秒，而且每个连接在 Broker 端都要占用文件描述符和一组 Erlang 进程。
Channel 是 Connection 之上的轻量级逻辑信道，多个 Channel 的数据帧通过 channel id 复用同一条 TCP 连接，创建一个 Channel 几乎没有网络开销。

text

     应用进程
  ┌───────────────────────────────┐
  │  Thread-1 ──► Channel-1 ──┐   │
  │  Thread-2 ──► Channel-2 ──┼──►│══════ 一条 TCP Connection ══════► Broker
  │  Thread-3 ──► Channel-3 ──┘   │       （帧头带 channel id 区分）
  └───────────────────────────────┘

必须记住的三条规则：

Channel 不是线程安全的：一个 Channel 绝对不能被多个线程同时 publish，否则会出现帧交错导致连接被 Broker 强制关闭（UNEXPECTED_FRAME）。Spring AMQP 的 RabbitTemplate 内部用 CachingConnectionFactory 做了 Channel 缓存，每次操作借一个、用完还回去，所以 RabbitTemplate 本身是线程安全的。
Channel 不能无限开：Channel 太多会导致 Broker 端 Erlang 进程暴涨、内存上升。channel-cache-size 默认 25，高并发发送时要调大，否则线程会在借 Channel 时排队。
一个 Connection 承载的 Channel 数量有上限（channel_max，默认 2047）。超过会报 channel_max exceeded。真正的高吞吐场景应该用多 Connection + 每 Connection 多 Channel，而不是单 Connection 死扛。

⚠️ 一个经典事故：开发为了「保证线程安全」，在每次发消息时 createConnection()，压测时 QPS 刚到 500，Broker 的文件描述符就被打满，日志里全是 too many open files。正确做法是复用 Connection、按需取 Channel。

6. 四种 Exchange 的路由规则与选型

类型	路由规则	routingKey 是否生效	典型场景
direct	bindingKey 与 routingKey 完全相等才投递	是	精确路由：日志级别、订单类型分流
fanout	忽略 routingKey，广播给所有绑定队列	否	一个事件通知多个系统，性能最好
topic	用通配符匹配，`*` 匹配一个单词，`#` 匹配零个或多个单词	是	多维度订阅：`order.create.vip`、`order.#`
headers	按消息 header 的键值对匹配，忽略 routingKey	否	极少用，性能最差，能被 topic 替代

topic 的通配符规则要能当场举例（面试高频）：

text

绑定键            能匹配的 routingKey                 不能匹配的
order.*          order.create、order.pay            order.create.vip（多了一段）
order.#          order、order.create、order.a.b.c   orders.create（第一段不等）
*.create.*       order.create.vip、item.create.v2   order.create（段数不够）
#                任意 routingKey                     无

两个关键细节：

分隔符固定是点号，所谓「单词」是指被点号分隔的一段。
* 必须匹配恰好一段，# 可以匹配零段。因此 order.# 能匹配 order 本身，而 order.* 不能。

⚠️ 路由不到队列的消息会被静默丢弃。这是 RabbitMQ 最阴险的丢消息方式：Exchange 存在、消息发送成功、confirm 也返回 ack，但消息根本没进任何队列。解决办法有两个：开启 mandatory + ReturnCallback 感知，或者给 Exchange 配置备份交换机 alternate-exchange 兜底。

面试加分表述：「我们线上统一用 topic，哪怕当前只有一个精确路由。原因是 topic 是 direct 的超集，后续要加一个按前缀订阅的下游时，不需要改生产端的 Exchange 类型，只要新加一个 binding。代价是 topic 的匹配要走 trie 树，比 direct 的哈希查找慢一点点，但在万级 QPS 下完全看不出差异。」

7. RabbitMQ 的持久化三要素：少一个都会丢

很多人以为「队列设了 durable 就不会丢」，这是错的。RabbitMQ 要在 Broker 重启后还能找回消息，必须同时满足三个条件：

#	要素	怎么配	不配的后果
1	Exchange 持久化	声明时 `durable = true`	重启后 Exchange 消失，新消息无处可投，老消息还在队列里
2	Queue 持久化	声明时 `durable = true`	重启后队列直接消失，队列里的消息全没，这是最严重的
3	Message 持久化	`deliveryMode = 2`（Spring Boot 默认即为 PERSISTENT）	队列还在，但消息只在内存，重启后消息全丢

java

// 声明持久化的交换机与队列（Spring Boot 配置类）
@Bean
public TopicExchange orderExchange() {
    // 参数：name、durable、autoDelete
    return new TopicExchange("order.topic", true, false);
}

@Bean
public Queue orderCreateQueue() {
    return QueueBuilder.durable("order.create.queue")          // durable = true
            .withArgument("x-dead-letter-exchange", "order.dlx")      // 绑定死信交换机
            .withArgument("x-dead-letter-routing-key", "order.dead")
            .withArgument("x-message-ttl", 1800000)                   // 队列级 TTL：30 分钟
            .withArgument("x-max-length", 100000)                     // 队列最大长度，超了按策略淘汰
            .build();
}

⚠️ 持久化 ≠ 不丢。消息写入磁盘是先进 PageCache，由操作系统异步刷盘，默认每 200ms 一次（queue_index_embed_msgs_below 相关机制）。这 200ms 内断电，消息照样丢。要做到真正零丢失，必须靠 publisher confirm + 消费端手动 ACK + 业务侧对账三层兜底，而不是指望持久化本身。

⚠️ 另一个坑：已经声明过的队列不能改属性。如果线上队列是 durable=false，想改成 true，直接改代码重启会报 PRECONDITION_FAILED - inequivalent arg，必须先删队列（会丢掉队列里的消息）再重建。所以队列属性必须在上线前一次定好。

8. 发送端可靠性：publisher confirm 与 return callback

RabbitMQ 的发送端有两套回调，职责完全不同，面试经常混为一谈：

机制	回答的问题	触发时机	失败含义
ConfirmCallback	消息有没有到达 Exchange	Broker 收到并（持久化消息）落盘后	nack：Broker 内部错误或 Exchange 不存在
ReturnCallback	消息有没有从 Exchange 路由到 Queue	路由失败且开启了 `mandatory`	路由键写错、binding 不存在

二者组合起来才是完整的发送端保障：

text

发送 ──► 到达 Exchange？──否──► ConfirmCallback(ack=false) ──► 重发 / 落库补偿
            │是
            ▼
       路由到 Queue？──否──► ReturnCallback ──► 告警 + 落库（路由键配置错误，重发也没用）
            │是
            ▼
       ConfirmCallback(ack=true) ──► 删除本地消息表记录

yaml

spring:
  rabbitmq:
    publisher-confirm-type: correlated   # 开启 confirm（异步回调，推荐）
    publisher-returns: true              # 开启 return
    template:
      mandatory: true                    # 路由失败时触发 return 而不是静默丢弃

publisher-confirm-type 的三个取值要能区分：

none：关闭，默认值。
simple：发完调用 waitForConfirms() 同步阻塞等结果。可靠但吞吐极低，相当于把异步发送变成同步 RPC。
correlated：异步回调，通过 CorrelationData 的 id 关联是哪条消息。生产唯一推荐。

⚠️ confirm 回调是在 IO 线程里执行的，回调里绝对不能做耗时操作（查库、发 HTTP），否则会阻塞整个连接的收发。正确做法是回调里只更新本地消息表状态或者丢到线程池。

⚠️ confirm 返回 ack 不代表消息一定没丢。消息落在 PageCache 里就可能返回 ack，此时断电仍然丢。这也是为什么重要业务要搭配 本地消息表 + 定时补偿：以本地库的事务为准，MQ 只是投递通道。

9. 消费端可靠性：三种 ACK 模式与 prefetch

模式	配置值	语义	风险
自动 ACK	`acknowledge-mode: none`	Broker 投递出去立刻删除消息	消费端一崩消息就没了，绝对不要用于业务消息
自动 ACK（方法返回）	`acknowledge-mode: auto`（默认）	监听方法正常返回则 ack，抛异常则 nack 并重回队列	抛异常默认无限重回队列，形成死循环刷满 CPU
手动 ACK	`acknowledge-mode: manual`	业务代码里显式 `basicAck` / `basicNack` / `basicReject`	忘了 ack 会导致消息一直是 unacked，连接断开才重新投递

三个确认 API 的区别必须说清：

java

// deliveryTag：当前 Channel 上的递增投递序号
// multiple：是否批量确认小于等于该 tag 的所有消息
channel.basicAck(deliveryTag, false);

// requeue = true  → 消息重回队列头部，会被立刻重投（容易死循环）
// requeue = false → 消息进入死信队列（若配了 DLX），否则丢弃
channel.basicNack(deliveryTag, false, false);

// 和 basicNack 的唯一区别：reject 不支持 multiple 批量
channel.basicReject(deliveryTag, false);

prefetch（basic.qos）是消费端最重要的调优参数：它限制一个 Channel 上「已投递但未 ack」的消息条数上限。

yaml

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 10            # 未 ack 消息上限，达到后 Broker 停止向该 Channel 投递
        concurrency: 5          # 初始消费者线程数
        max-concurrency: 20     # 最大消费者线程数
        retry:
          enabled: false        # 关掉框架重试，由业务 + 死信控制

prefetch 的取值逻辑：

prefetch = 1：严格的「处理完一条再拿一条」，吞吐最低但负载最均衡，顺序消费场景必须用 1。
prefetch 过大（或不设，默认无限）：Broker 会把队列里的消息一次性全推给第一个连上的消费者，造成消费者之间负载极度不均，还可能把消费端内存打爆。
经验公式：prefetch ≈ 单条消息处理耗时(ms) 的倒数级别。处理快（1ms 级）设 100~250，处理慢（100ms 级）设 5~10。

⚠️ 最危险的组合是 auto 模式 + 业务抛异常。消息 nack 后 requeue 回队列头，立刻又被投递，再次抛异常……每秒上万次循环，CPU 飙满、日志把磁盘写爆。正确做法：要么关掉 requeue 让它进死信，要么用 max-attempts 限制重试次数后进死信。

10. RocketMQ 的四大组件与整体架构

text

                    ┌─────────────── NameServer 集群（无状态、互不通信）────────────────┐
                    │   NameServer-1      NameServer-2      NameServer-3               │
                    └────▲──────────────────────▲─────────────────────▲────────────────┘
      30s 心跳注册路由    │                      │   拉取路由（30s）     │
            ┌────────────┴──────┐               │            ┌─────────┴────────┐
            │                   │               │            │                  │
   ┌────────┴────────┐  ┌───────┴─────────┐     │      ┌─────┴──────┐   ┌───────┴──────┐
   │ Broker-a Master │  │ Broker-b Master │     │      │  Producer  │   │   Consumer   │
   │   ├ CommitLog   │  │   ├ CommitLog   │     │      └─────┬──────┘   └───────┬──────┘
   │   ├ ConsumeQueue│  │   ├ ConsumeQueue│     │            │ send             │ pull
   │   └ IndexFile   │  │   └ IndexFile   │◄────┴────────────┴──────────────────┘
   └────────┬────────┘  └───────┬─────────┘
            │ 同步/异步复制       │
   ┌────────┴────────┐  ┌───────┴─────────┐
   │ Broker-a Slave  │  │ Broker-b Slave  │
   └─────────────────┘  └─────────────────┘

组件	职责	关键特性
NameServer	轻量级注册中心，保存 Topic 到 Broker 的路由表	节点之间完全不通信，是 AP 设计；数据靠 Broker 的 30s 心跳重建，所以挂一半也能用
Broker	存储和转发消息，分 Master / Slave	每 30s 向所有 NameServer 注册路由；120s 收不到心跳则被摘除
Producer	发消息，从 NameServer 拉路由后直连 Broker Master	客户端做负载均衡（轮询 MessageQueue）与故障规避
Consumer	拉消息，可连 Master 也可连 Slave	默认从 Master 拉，Master 压力大或不可用时会被引导到 Slave

四个必须理解的设计取舍：

为什么不用 ZooKeeper 而自研 NameServer：MQ 的路由信息只需要最终一致，不需要强一致。去掉 ZK 既减少了外部依赖，又避免了 ZK 在网络分区时不可写的问题。代价是路由变更有最长 30 秒的感知延迟。
Broker 向所有 NameServer 注册：任何一个 NameServer 都有完整路由，所以 NameServer 可以无脑水平扩展，挂到只剩一个仍然可用。
Broker 摘除有延迟：Master 宕机后，NameServer 要 120 秒才剔除它，这期间生产者仍可能把消息发到这台已死的 Broker 上，表现为发送超时。客户端的 故障规避（latencyFaultTolerance） 就是为了填这段空窗。
消息发送是同步 RPC：Producer 直连 Broker Master，不经过 NameServer 转发。NameServer 只提供路由，不在数据链路上。

11. RocketMQ 的存储模型：CommitLog、ConsumeQueue、IndexFile

这是 RocketMQ 最核心、也是和 Kafka 差异最大的设计。

text

Producer 发送 Topic-A / Topic-B / Topic-C 的消息
                │
                ▼  全部顺序追加写入同一个文件（混合存储）
┌──────────────────────── CommitLog ────────────────────────┐
│ 00000000000000000000 (1GB)  00000000001073741824 (1GB) …  │
│ [A的msg][B的msg][C的msg][A的msg][B的msg] ……              │
└───────────────────────────┬───────────────────────────────┘
                            │ 异步构建索引（ReputMessageService）
          ┌─────────────────┴────────────────┐
          ▼                                  ▼
┌──────── ConsumeQueue ────────┐    ┌─────── IndexFile ───────┐
│ Topic-A/queue-0/00000000     │    │ 哈希索引：key → offset  │
│ 每条 20 字节定长：            │    │ 支持按 msgKey / msgId   │
│  ├ 8B  CommitLog 物理偏移    │    │ 查询单条消息            │
│  ├ 4B  消息长度              │    └─────────────────────────┘
│  └ 8B  TagsCode（过滤用）    │
└──────────────────────────────┘

三个文件的分工：

文件	作用	大小	为什么这样设计
CommitLog	存消息本体，所有 Topic 混合顺序写	单文件固定 1GB，写满滚动新文件	顺序写磁盘可达 600MB/s，接近内存随机写速度；Topic 再多也是一个文件，所以能撑几万 Topic
ConsumeQueue	消费索引，按 Topic + QueueId 组织	每条定长 20 字节，单文件存 30 万条约 5.72MB	定长意味着可以按下标直接定位：`offset × 20` 就是文件位置，O(1) 查找
IndexFile	哈希索引，支持按消息 key 或 msgId 查单条消息	固定 400MB 左右	运维排查用，业务消费不依赖它

读写路径的区别是关键：

写路径是纯顺序的：消息只追加到 CommitLog 尾部，这是高吞吐的来源。
读路径是「两次读」：先读 ConsumeQueue 拿到物理偏移（顺序读，且文件小、几乎全在 PageCache），再按偏移读 CommitLog（随机读）。
所以 RocketMQ 的读性能依赖 PageCache 命中率。消费正常（读的是刚写入的热数据）时几乎全命中 PageCache，速度极快；一旦消息堆积导致消费者去读几小时前的冷数据，就会触发大量磁盘随机读，表现为 Broker 磁盘 IO 飙升、所有生产者发送变慢。

面试加分表述：「Kafka 是每个分区一个独立的 log 文件，写入是「分区内顺序、全局随机」；RocketMQ 是全局一个 CommitLog，写入是真正的全局顺序。所以分区数增长时，Kafka 的磁盘会退化成随机写而 RocketMQ 不会——这就是 RocketMQ 能支撑几万 Topic 而 Kafka 在几千分区就吃力的根因。代价是 RocketMQ 读消息要多一跳 ConsumeQueue 索引，且读 CommitLog 是随机读。本质是用读的复杂度换写的吞吐和 Topic 规模。」

12. 刷盘机制与主从复制：两组配置决定可靠性等级

RocketMQ 把「消息安全」拆成了两个正交的维度：单机落盘和跨机复制。

第一组：刷盘方式（flushDiskType）

取值	行为	丢数据窗口	性能
ASYNC_FLUSH（默认）	消息写入 PageCache 即返回成功，后台线程定时 fsync	机器断电丢失未刷盘部分（默认最多 500ms 或 64KB）	高
SYNC_FLUSH	必须 fsync 到磁盘才返回成功	几乎不丢（除磁盘损坏）	下降一个数量级

第二组：主从复制（brokerRole）

取值	行为	丢数据窗口	性能
ASYNC_MASTER（默认）	写入 Master 成功即返回，异步复制给 Slave	Master 整机故障且数据未复制完则丢	高
SYNC_MASTER	必须至少一个 Slave 确认收到才返回成功	Master 挂了 Slave 有全量数据	下降约 20%~30%
SLAVE	从节点角色	—	—

四种组合的生产选择：

text

ASYNC_FLUSH + ASYNC_MASTER  → 性能最高，可接受极少量丢失：日志、埋点、通知
ASYNC_FLUSH + SYNC_MASTER   → 【生产推荐】机器断电可能丢一点，整机故障不丢：订单、交易
SYNC_FLUSH  + SYNC_MASTER   → 最可靠，性能最差：资金、账务（通常还会叠加业务对账）
SYNC_FLUSH  + ASYNC_MASTER  → 组合不合理，单机安全但整机故障仍丢，性能却已牺牲

⚠️ SYNC_MASTER 不等于强一致。它只保证「至少一个 Slave 收到」，但 RocketMQ 4.5 之前的主从不能自动切换：Master 挂了只能读 Slave，不能写，必须人工介入。要自动故障转移必须用 DLedger（Raft） 模式，代价是至少 3 个节点且写入要过半确认。

面试加分表述：「我们线上用的是 ASYNC_FLUSH + SYNC_MASTER。理由是：机房有 UPS，整机断电的概率远低于单机故障，所以同步复制的价值大于同步刷盘；而 SYNC_FLUSH 会让 TPS 从 8 万掉到 8 千，这个代价换来的只是「断电不丢」这个极低概率事件。真正的资金一致性我们是靠本地消息表 + 日终对账保证的，而不是靠 MQ 的刷盘配置。」

13. Topic、Queue 与消费者数量的关系

RocketMQ 的 Topic 下有多个 MessageQueue（队列 / 分区），这是并发度的基本单位。

text

Topic: ORDER_TOPIC，writeQueueNums = 4，readQueueNums = 4

Broker-a            Broker-b
 ├ queue-0           ├ queue-0        实际是 4 × 2 = 8 个 MessageQueue
 ├ queue-1           ├ queue-1
 ├ queue-2           ├ queue-2
 └ queue-3           └ queue-3

同一消费者组内 3 个消费者，平均分配策略下：
 Consumer-1 → a:q0, a:q1, a:q2
 Consumer-2 → a:q3, b:q0, b:q1
 Consumer-3 → b:q2, b:q3

三条必须记住的规则：

一个 MessageQueue 在同一消费者组内，只会被一个消费者消费。所以消费者数量超过队列总数时，多出来的消费者完全空转。这是扩容消费者却没有提速的最常见原因。
消费并发度上限 = 队列总数 × 单消费者线程数。队列数不够时，加机器无效，必须先扩队列（updateTopic -w -r）。
读写队列数可以不等：缩容时先把 writeQueueNums 降下来，等旧队列消费完再降 readQueueNums，否则会出现「有消息但没人消费」的残留队列。

⚠️ RocketMQ 的队列数只能增不能随意减。减队列会导致该队列上未消费的消息无人消费（消费者的 Rebalance 不再分配到它），必须先确认队列堆积为 0。

14. RocketMQ 的五种消息类型

类型	用途	关键 API	代价
普通消息	绝大多数场景	`send(msg)`	无序、无事务
顺序消息	同一业务主键的操作必须按序执行	`send(msg, selector,shardingKey)`	吞吐下降、单队列阻塞风险
延时消息	定时任务、超时关单	`msg.setDelayTimeLevel(n)`	4.x 只支持 18 个固定等级
事务消息	本地事务与消息发送的原子性	`sendMessageInTransaction`	半消息 + 回查，链路复杂
批量消息	提升吞吐	`send(List<Message>)`	单批总大小不能超 4MB，且不能是延时 / 事务消息

四种发送方式也要分清（和消息类型是两个维度）：

发送方式	是否等 Broker 响应	可靠性	场景
同步 `send`	等	最高，返回 SendResult	核心业务：订单、支付
异步 `send(msg, callback)`	不等，回调通知	高，失败可在回调重试	对 RT 敏感但要可靠：通知、推送
单向 `sendOneway`	不等，不要结果	最低，丢了不知道	日志、心跳收集
批量 `send(List)`	等	高	高吞吐数据同步

java

// 发送结果的四种状态，必须判断！
SendResult result = producer.send(msg);
switch (result.getSendStatus()) {
    case SEND_OK:                       // 真正成功
        break;
    case FLUSH_DISK_TIMEOUT:            // SYNC_FLUSH 下刷盘超时，消息可能丢
    case FLUSH_SLAVE_TIMEOUT:           // SYNC_MASTER 下从节点同步超时
    case SLAVE_NOT_AVAILABLE:           // 从节点不可用
        // 这三种都表示「Broker 收到了但不保证持久」，需要业务侧记录并补偿
        log.warn("消息投递存在风险，status={}, msgId={}", result.getSendStatus(), result.getMsgId());
        break;
}

⚠️ 最常见的生产事故是「只调用 send 不看返回值」。只要没抛异常就认为成功，结果 FLUSH_SLAVE_TIMEOUT 的消息悄悄丢了。正确做法：只有 SEND_OK 才算成功，其他状态要么重试要么落补偿表。

15. Push 还是 Pull：长轮询的真相

RocketMQ 提供 DefaultMQPushConsumer 和 DefaultMQPullConsumer，但底层都是 Pull。所谓的 Push 是长轮询（Long Polling） 模拟出来的。

text

Consumer                                    Broker
   │── pullMessage(offset, timeout=15s) ──────►│
   │                                           │ 有新消息？
   │                                           │   有 → 立即返回
   │                                           │   没有 → 【挂起请求】不立即返回
   │                                           │        （PullRequestHoldService 挂起，默认最长 15s）
   │                                           │        新消息到达时被唤醒立即返回
   │◄────────── 返回消息 / 超时返回空 ──────────│
   │── 立刻发起下一次 pull ───────────────────►│

三种模型的对比：

模型	实时性	服务端压力	消费端压力	用在哪
纯 Push	最好	服务端要维护消费者状态和推送速率	可能被打爆（无法反压）	RabbitMQ（靠 prefetch 反压）
纯 Pull	差（取决于轮询间隔）	小	空轮询浪费资源	Kafka（靠批量 + poll 间隔缓解）
长轮询	接近 Push	要挂起连接，占内存	可控，消费端决定拉多少	RocketMQ

面试加分表述：「长轮询的精髓是把「空轮询的等待」从客户端挪到了服务端。客户端发一次请求就阻塞在那里，服务端挂起这个请求直到有消息或超时。这样既有 Push 的实时性（消息到了立刻唤醒），又有 Pull 的流控能力（消费速率由消费者自己的拉取节奏决定），避免了纯 Push 把慢消费者打爆的问题。」

⚠️ DefaultMQPushConsumer 的「Push」只是编程模型上的 Push（你注册一个 Listener，消息会被回调给你），底层仍是客户端主动拉。所以消费端的流控参数（pullBatchSize、consumeThreadMin/Max、pullThresholdForQueue）才是真正决定消费速率的东西。

16. 消费模式：集群消费与广播消费

模式	语义	位点存储位置	典型场景
CLUSTERING（集群，默认）	一条消息只被组内一个消费者处理	Broker 端（`consumerOffset.json`）	业务处理：扣库存、发券、落库
BROADCASTING（广播）	一条消息被组内每个消费者都处理一次	消费者本地文件（`~/.rocketmq_offsets`）	刷新本地缓存、刷新本地配置、本地限流规则下发

java

consumer.setMessageModel(MessageModel.BROADCASTING);

三个关键差异：

位点存储位置不同导致的运维差异：广播模式的位点存在消费者本机，换机器部署就等于位点丢失，会从 consumeFromWhere 配置的位置重新开始。所以广播模式必须保证消费逻辑幂等。
广播模式不支持重试：消费失败返回 RECONSUME_LATER 在广播模式下无效，消息直接被跳过，也不会进死信队列。这是最容易踩的坑。
广播模式下 consumeFromWhere 才真正生效：集群模式下，只有全新的消费者组第一次启动时 consumeFromWhere 才生效（因为 Broker 上没有位点记录），老组重启一律从已存位点继续。

⚠️ 用广播模式刷新本地缓存是正确用法；但绝对不要用广播模式做「需要可靠执行」的业务，因为它既没有重试也没有死信，失败就是静默丢失。真要让多个实例都处理且要求可靠，应该给每个实例一个独立的消费者组（集群模式 + 不同 group），这样每组都有自己的位点和重试。

17. Rebalance：消费者负载均衡机制

Rebalance 决定「哪个消费者消费哪些队列」，是消费端最容易出问题的地方。

触发时机：

消费者上线 / 下线 / 崩溃（心跳超时）
Topic 的队列数量变更（扩容队列）
消费者每 20 秒主动触发一次 RebalanceService
Broker 发现消费者组成员变化，主动通知组内所有消费者

五种分配策略：

策略	规则	适用
`AllocateMessageQueueAveragely`（默认）	连续平均分配，`q0-q3 → C1`，`q4-q7 → C2`	绝大多数场景
`AllocateMessageQueueAveragelyByCircle`	轮询分配，`q0,q2,q4 → C1`，`q1,q3,q5 → C2`	希望跨 Broker 打散
`AllocateMessageQueueByMachineRoom`	优先分配同机房的 Broker 队列	多机房部署，降低跨机房流量
`AllocateMessageQueueConsistentHash`	一致性哈希	希望扩缩容时减少队列迁移
`AllocateMessageQueueByConfig`	手动配置固定队列	灰度 / 隔离指定流量

⚠️ Rebalance 期间会出现重复消费。因为队列从 C1 迁到 C2 时，C1 可能已经拉取并处理了部分消息但位点还没提交，C2 从旧位点重新拉取，这些消息会被消费两次。这是 MQ 无法避免的，只能靠消费端幂等兜底——这也是「为什么必须做幂等」最硬的理由。

⚠️ 滚动发布时每重启一个实例就触发一次 Rebalance，一个 10 实例的服务滚动发布会触发 20 次 Rebalance，期间重复消费量不小。所以发布窗口要避开业务高峰，且幂等必须是强幂等（唯一索引级别），不能只靠 Redis 的 setnx。

18. 消息重试与延时等级

RocketMQ 的重试机制（集群模式下）：

text

消费失败（返回 RECONSUME_LATER 或抛异常）
        │
        ▼
消息被发回 Broker，进入重试 Topic：%RETRY%消费者组名
        │
        ▼  按延时等级递增投递，共 16 次
第 1 次 10s → 第 2 次 30s → 1m → 2m → 3m → 4m → 5m → 6m
→ 7m → 8m → 9m → 10m → 20m → 30m → 1h → 2h
        │
        ▼ 16 次仍失败
进入死信 Topic：%DLQ%消费者组名（保留 3 天，默认不可消费，需手动开启权限）

RocketMQ 的 18 个延时等级（messageDelayLevel 配置，4.x）：

text

级别:  1   2    3    4   5   6   7   8   9   10  11  12  13  14  15  16  17  18
时间: 1s  5s  10s  30s  1m  2m  3m  4m  5m  6m  7m  8m  9m 10m 20m 30m  1h  2h

java

Message msg = new Message("ORDER_TOPIC", "订单超时检查".getBytes());
msg.setDelayTimeLevel(16);   // 等级 16 = 30 分钟后投递
producer.send(msg);

延时消息的实现原理（4.x）：

消息发送时，Broker 发现 delayTimeLevel > 0，就把原 Topic 和 QueueId 存进消息属性，然后把消息的 Topic 改写成内部 Topic SCHEDULE_TOPIC_XXXX，QueueId 改成 level - 1。
这样 18 个等级就对应 SCHEDULE_TOPIC_XXXX 下的 18 个队列，每个队列内的消息天然按到期时间有序（因为同一队列延时相同，按写入顺序就是按到期顺序）。
Broker 有一个 ScheduleMessageService，为每个等级起一个定时任务，到点后把消息的 Topic 还原成原始 Topic 再重新写入 CommitLog，消费者就能拉到了。

面试加分表述：「RocketMQ 4.x 只支持固定等级而不是任意时间，根本原因是它用「等级分队列」把排序问题消掉了。如果支持任意延时，Broker 就必须维护一个按到期时间排序的数据结构（时间轮或堆），这在百万级延时消息下成本很高。RocketMQ 5.0 引入了基于时间轮 + TimerWheel 文件的方案，才支持了任意精度延时。」

RabbitMQ 没有原生延时消息，两种实现方式：

方式	原理	致命缺陷
TTL + DLX	消息设 TTL，过期后进死信队列，消费死信队列	队头阻塞：队列 TTL 是按队头判断的，队头消息 30 分钟不过期，后面 1 分钟的消息也出不来
延时插件 `rabbitmq_delayed_message_exchange`	插件实现的延时交换机，消息存在 Erlang 的 Mnesia 中按时间排序	延时消息存在内存中，量大时内存压力大；插件非官方核心组件，重启可能丢

⚠️ 「TTL + DLX 做延时队列」的队头阻塞是面试必问。解法是给每个延时档位建一个独立队列（5 分钟队列、30 分钟队列、2 小时队列），同一队列内所有消息 TTL 相同就不会阻塞。但这样档位就固定了，和 RocketMQ 的延时等级是一个思路。

19. 死信队列：进入条件与处理规范

RabbitMQ 的死信交换机 DLX，三种进入条件：

#	条件	触发场景
1	消息被 `basicNack` / `basicReject` 且 `requeue = false`	业务判定为脏数据，主动拒绝
2	消息 TTL 过期（队列级 `x-message-ttl` 或消息级 `expiration`）	延时队列、超时未处理
3	队列达到最大长度 `x-max-length` 或最大字节数 `x-max-length-bytes`	堆积超限，按 `x-overflow` 策略淘汰队头

java

// 配置死信：业务队列指定 DLX 和 DLK
@Bean
public Queue businessQueue() {
    return QueueBuilder.durable("order.business.queue")
            .withArgument("x-dead-letter-exchange", "order.dlx.exchange")
            .withArgument("x-dead-letter-routing-key", "order.dead")
            .build();
}

// 死信队列本身是一个普通队列，需要有消费者去处理
@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable("order.dead.queue").build();
}

RocketMQ 的死信队列 DLQ，进入条件只有一个：重试 16 次后仍然失败。

特性	RabbitMQ DLX	RocketMQ DLQ
命名	自己指定	固定 `%DLQ%消费者组名`
粒度	按队列配置	按消费者组
进入条件	拒绝 / TTL / 超长	重试次数耗尽
默认可读	是（普通队列）	否，默认权限是 2（只写），要 `updateTopicPerm` 改成 6 才能消费
保留时间	同普通队列	3 天，之后自动删除

死信的处理规范（这是体现工程素养的地方）：

死信队列必须有消费者，否则就是个黑洞。最低标准是消费后落库 + 告警。
死信不要自动重投回原队列，否则可能形成「原队列 → 死信 → 原队列」的无限循环。必须人工或半自动审核后重投。
死信消息要带上失败原因。RabbitMQ 的死信消息 header 里有 x-death 数组，记录了进入死信的原因、次数、原队列、时间，必须打到日志里。
死信量要做监控告警，死信突增通常意味着下游依赖挂了或者消息格式变更了。

⚠️ RocketMQ 的死信 Topic 默认不可见不可消费，很多人以为消息「彻底丢了」。实际是在 %DLQ%组名 里躺着，可以用控制台或 mqadmin topicPerm 开权限后用一个新的消费者组去消费。

20. 消息可靠性全链路：三处可能丢失

text

         ①生产者 → Broker 丢失          ②Broker 内部丢失         ③Broker → 消费者丢失
Producer ──────────────────► Broker ──────────────────► Broker ──────────────────► Consumer
   网络抖动 / 进程崩溃          PageCache 未刷盘断电        自动 ACK 后业务异常
   未判断 SendResult           主从未同步 Master 宕机       消费端崩溃但位点已提交

阶段	RabbitMQ 方案	RocketMQ 方案	兜底
①生产者丢	publisher confirm + mandatory/ReturnCallback + 重试	同步发送 + 判断 `SEND_OK` + 重试 3 次 + 事务消息	本地消息表：先入库再发消息，定时扫描未确认记录重发
②Broker 丢	Exchange / Queue / Message 三重持久化 + 镜像队列或 Quorum Queue	`SYNC_FLUSH`（或至少 `SYNC_MASTER`）+ 主从 / DLedger	多副本 + 跨机房部署
③消费者丢	手动 ACK，业务执行成功后才 ack；失败 nack 进死信	消费成功返回 `CONSUME_SUCCESS`，失败返回 `RECONSUME_LATER` 走重试和死信	死信队列 + 对账补偿

面试加分表述：「我回答「怎么保证消息不丢」时会先定一个前提：单靠 MQ 的配置做不到 100% 不丢，只能把丢失概率降到足够低，再用业务对账兜住剩下的。所以我们的方案是三层：第一层是 MQ 的可靠性配置（confirm + 持久化 + 手动 ACK）；第二层是本地消息表，让「业务数据」和「消息待发记录」在同一个数据库事务里，以数据库为准做补偿；第三层是 T+1 对账任务，比对上下游数据量，差异走人工处理。生产上真正救过我们的是第二层和第三层。」

⚠️ 「消息零丢失」的代价是极低的吞吐和极高的复杂度。面试时如果一上来就说「配 SYNC_FLUSH + SYNC_MASTER 就不丢了」，会被反问性能数据。正确姿势是先讲清丢失的三个位置，再按业务重要性给分级方案。

21. 消费幂等：为什么 MQ 做不到 Exactly Once

消息投递有三种语义：

语义	含义	实现难度	现状
At Most Once	最多一次，可能丢	简单（自动 ACK / oneway）	日志场景可接受
At Least Once	至少一次，可能重复	中等（手动 ACK + 重试）	所有主流 MQ 的实际保证
Exactly Once	恰好一次	极难	MQ 层面做不到，只能靠「At Least Once + 消费端幂等」等价实现

为什么做不到 Exactly Once：消费端「执行业务」和「提交位点」是两个独立的操作，无法原子化。

text

顺序 A：先提交位点，再执行业务 → 位点提交后崩溃 ⇒ 消息丢失（At Most Once）
顺序 B：先执行业务，再提交位点 → 业务完成后崩溃 ⇒ 消息重复（At Least Once）

MQ 选择了 B，把「重复」这个较轻的问题交给业务侧的幂等解决。

重复消费的六个真实来源（面试可以直接列）：

生产者重试：发送超时但 Broker 实际已收到，重试导致同一业务产生两条消息。
消费端位点未提交就崩溃：业务执行完、ACK 未到 Broker。
Rebalance：队列迁移，新消费者从旧位点重新拉取。
消费超时：RabbitMQ 中消费者长时间不 ack，连接断开后消息重新投递。
主从切换：Slave 的位点落后于 Master，切换后从旧位点重新消费。
人工重投：运维从死信队列重放消息。

幂等的四种实现方案：

方案	原理	可靠性	适用
数据库唯一索引	用业务唯一键建唯一索引，重复插入报错即忽略	最高（强幂等）	插入类操作：订单、流水
乐观锁 / 状态机	`update ... where status = 待支付`，影响行数为 0 则已处理	高	更新类操作：订单状态流转
Redis 去重表	`SET msgKey 1 NX EX 86400`，已存在则跳过	中（Redis 丢数据或过期后会重复）	高并发前置过滤
消费记录表	单独一张表记录已消费的 msgId，与业务在同一事务	高	通用方案

⚠️ 幂等键绝对不要用 msgId。RocketMQ 的 msgId 在生产者重试时会变（重试会生成新的 msgId），而且 Broker 端还有 offsetMsgId 的概念。正确的幂等键是业务唯一键：订单号、支付流水号、业务类型 + 业务 ID。

面试加分表述：「我们的标准做法是两级幂等：第一级用 Redis 的 SET NX 做快速拦截，挡住 99% 的重复（成本低、不占数据库连接）；第二级用数据库唯一索引做最终保证，捕获 DuplicateKeyException 后直接返回成功。单靠 Redis 不行，因为 Redis 的 key 过期或者主从切换丢 key 之后重复消息就会漏过去；单靠数据库也不好，因为高并发重复会产生大量无效的数据库写入和异常栈。」

✨ 高级深挖知识点

1. 消息乱序的根因：三个并发点

text

乱序点①：生产端负载均衡
   同一订单的「创建、支付、取消」轮询发到 q0、q1、q2 → 三个队列并行消费，顺序完全丢失

乱序点②：消费端多队列并行
   一个消费者分到 q0 和 q1，两个线程同时拉取消费 → 跨队列无序

乱序点③：消费端线程池并发
   同一队列拉到 32 条消息，丢进 consumeThread 线程池 → 队列内部也乱了

所以要保证顺序必须同时锁住这三个点：

生产端：同一业务主键的消息必须路由到同一个队列（用 MessageQueueSelector 按 orderId.hashCode() % queueSize）。
消费端：同一队列只能被一个消费者消费（MQ 本身保证）。
消费端：同一队列的消息必须串行处理（RocketMQ 用 MessageListenerOrderly，RabbitMQ 用 prefetch = 1 + 单线程）。

顺序级别	实现方式	吞吐	适用
全局顺序	Topic 只设 1 个队列 + 单消费者单线程	极低（几千 TPS）	几乎没有真实场景需要
分区顺序（局部顺序）	按业务 key 哈希到固定队列，队列内串行	高（可随队列数扩展）	99% 的业务场景

面试加分表述：「被问到顺序消息，我一定会先反问「需要的是全局顺序还是分区顺序」。因为业务上真正需要全局顺序的场景极少，绝大多数是「同一个订单 / 同一个用户的操作要有序」，这就是分区顺序。分区顺序的实现成本和性能损失都可以接受，而全局顺序意味着整个 Topic 退化成单队列单线程，吞吐直接掉到几千，还丧失了横向扩展能力。」

2. 顺序消息在消费端的三个坑

java

// 生产端：按订单号选队列，保证同一订单进同一队列
producer.send(msg, (mqs, message, arg) -> {
    long orderId = (Long) arg;
    // 必须用稳定的哈希，不能用 Object.hashCode()（String 的 hashCode 稳定，但自定义对象不稳定）
    int index = (int) (Math.abs(orderId) % mqs.size());
    return mqs.get(index);
}, orderId);

// 消费端：必须用 MessageListenerOrderly，内部对每个队列加锁串行消费
consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        try {
            handle(msg);
        } catch (Exception e) {
            // ⚠️ 顺序消费失败只能返回 SUSPEND_CURRENT_QUEUE_A_MOMENT
            // 它会让【整个队列】暂停一会儿重试，而不是跳过这一条
            context.setSuspendCurrentQueueTimeMillis(3000);
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    }
    return ConsumeOrderlyStatus.SUCCESS;
});

三个坑：

一条消息卡住，整个队列全部阻塞。顺序消费失败后不能跳过（跳过就乱序了），只能原地重试。如果是「永久失败」（比如消息体格式错误导致反序列化失败），这个队列就永远卡死，后面的消息全部堆积。解法：设置最大重试次数，超过后记录到异常表 + 告警 + 放过这条消息，由人工介入，牺牲单条的顺序保住整个队列。
队列数变更会打破顺序。队列从 4 个扩到 8 个时，orderId % 4 变成 orderId % 8，同一订单的新消息会被路由到新队列，而旧队列里还有老消息没消费完，顺序就断了。解法：顺序 Topic 的队列数一旦定下不要改，必须改时要先把存量消息消费完再切换。
Rebalance 期间顺序会被打破。队列从 C1 迁到 C2 时，C1 可能还有消息在处理中，C2 已经开始拉下一批。RocketMQ 用队列级分布式锁（向 Broker 申请 lockBatchMQ）缓解：C2 必须先拿到队列锁才能消费，而 C1 释放锁前会等待本地处理完。但锁有 30 秒过期时间，极端情况仍可能重叠。

⚠️ 顺序消息和延时消息不能混用。RocketMQ 的延时消息要经过 SCHEDULE_TOPIC_XXXX 中转再重新投递，重新投递时队列选择是重新计算的，顺序必然被打破。

3. RocketMQ 事务消息：半消息 + 事务回查

事务消息解决的是**「本地事务」与「发送消息」的原子性**问题：要么本地事务提交且消息投出去，要么本地事务回滚且消息不可见。

text

      ┌──────────────── 正常流程 ────────────────┐
①Producer ──发送半消息(Half Message)──► Broker
   │                                      │ 写入内部 Topic：RMQ_SYS_TRANS_HALF_TOPIC
   │◄────────── 返回发送成功 ──────────────┘ 【消费者看不到】
   │
②执行本地事务（扣款、写订单）
   │
③根据本地事务结果发送二次确认：
   ├ COMMIT   ──► Broker 把半消息恢复成原 Topic，写入 CommitLog ──► 消费者可见
   ├ ROLLBACK ──► Broker 丢弃半消息（写 OP 消息标记已处理）
   └ UNKNOW   ──► 不做处理，等回查

      ┌──────────────── 异常补偿（回查）────────────────┐
④Broker 定时扫描 Half Topic，发现超过 6 秒没有二次确认的半消息
   │
⑤Broker 向 Producer 组内【任意一台】机器发起回查：checkLocalTransaction
   │
⑥Producer 查本地事务状态（查订单表是否有这条记录），返回 COMMIT / ROLLBACK / UNKNOW
   │
⑦最多回查 15 次（transactionCheckMax），仍为 UNKNOW 则【默认丢弃】并打日志

java

TransactionMQProducer producer = new TransactionMQProducer("order_tx_group");
producer.setTransactionListener(new TransactionListener() {

    /** 第②步：半消息发送成功后，执行本地事务 */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            String txId = msg.getTransactionId();
            // 本地事务里必须记录 txId，否则回查时无法判断状态
            orderService.createOrder((OrderDTO) arg, txId);
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    /** 第⑥步：Broker 回查，必须是幂等且无副作用的查询 */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String txId = msg.getTransactionId();
        Integer status = orderService.getTxStatus(txId);
        if (status == null)        return LocalTransactionState.UNKNOW;      // 还没写进去，下次再查
        if (status == 1)           return LocalTransactionState.COMMIT_MESSAGE;
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
});

三个关键实现细节：

半消息为什么消费者看不到：Broker 收到半消息后，把它的真实 Topic 和 QueueId 存到属性里，然后改写 Topic 为 RMQ_SYS_TRANS_HALF_TOPIC。消费者订阅的是业务 Topic，自然拉不到。
Rollback 不是删除：CommitLog 是顺序追加的，不能修改已写入的数据。Rollback 的实现是往 RMQ_SYS_TRANS_OP_HALF_TOPIC 写一条 OP 消息标记「这条半消息已处理」。回查线程扫描时跳过有 OP 标记的半消息。
回查只能解决「Producer 宕机导致二次确认丢失」。它依赖本地事务留下了可查询的痕迹（所以必须把 transactionId 落到业务库），否则回查永远返回 UNKNOW，15 次后消息被丢弃。

⚠️ 事务消息只保证「本地事务 + 消息投递」的原子性，不保证消费者一定成功。消费者失败走的还是重试和死信那一套。所以事务消息不等于分布式事务，它实现的是最终一致性，不是 ACID。

面试加分表述：「事务消息的本质是 2PC 的变体：半消息是 prepare 阶段，二次确认是 commit/rollback 阶段，回查是超时补偿。它比传统 2PC 优秀的地方在于不需要协调者持有全局锁，Broker 只是被动等待确认超时后主动问一次，所以可用性高、吞吐不受影响。但它的代价是业务侧必须实现一个可靠的回查接口，这个接口如果写错（比如查的是缓存而不是数据库），整个机制就失效了。」

4. 分布式事务三种方案的对比：本地消息表、事务消息、Seata

维度	本地消息表	RocketMQ 事务消息	Seata AT 模式
一致性	最终一致	最终一致	准强一致（有全局锁）
对 MQ 的要求	无特殊要求，任何 MQ 都行	必须支持事务消息	不依赖 MQ
额外存储	需要一张消息表	不需要（Broker 承担）	需要 `undo_log` 表
侵入性	中（要写消息表 + 定时任务）	低（实现两个回调）	低（加注解）
性能	较好（一次本地事务 + 定时扫表）	最好（无额外库操作）	较差（全局锁，热点行串行）
下游失败怎么办	MQ 重试 + 死信 + 人工	MQ 重试 + 死信 + 人工	自动回滚上游
典型场景	通用、技术栈不受限	电商下单扣库存、支付通知	需要回滚的强一致场景

本地消息表的完整流程（这是最通用、面试最稳的答案）：

text

┌─────────── 同一个本地数据库事务 ───────────┐
│  ①INSERT INTO t_order (...)               │
│  ②INSERT INTO t_mq_message (状态=待发送)  │
└───────────────────┬────────────────────────┘
                    │ 事务提交后
                    ▼
            ③发送消息到 MQ
                    │
        ┌───────────┴───────────┐
     成功                     失败/超时
        │                         │
  ④UPDATE 状态=已发送        不处理，保持待发送
        │                         │
        └───────────┬─────────────┘
                    ▼
     ⑤定时任务每 30s 扫描「待发送 且 创建时间 > 1 分钟」的记录重发
       重发超过 5 次则标记为「异常」并告警，转人工

本地消息表的两个细节：

为什么要「创建时间 > 1 分钟」这个条件：刚插入的记录可能正在被第③步发送，立即扫描会导致重复发送。加一个延迟窗口能大幅减少重复。
消息表要做归档：已发送成功的记录保留 7 天后归档到历史表或删除，否则这张表会变成最大的表，扫描会越来越慢。

面试加分表述：「我们线上两种都在用，但主力是本地消息表。原因是：事务消息的回查接口本质上还是要查「本地事务有没有成功」，这就要求业务表里有能唯一定位的 txId——既然都要改表结构了，不如直接做消息表，还能顺带获得消息发送记录、重发次数、失败原因这些可观测性，排查问题时特别有用。事务消息更适合那种「不想额外维护表和定时任务」的轻量场景。」

5. 零拷贝：mmap 与 sendfile 的本质差异

传统 IO 的四次拷贝、四次上下文切换：

text

read(file) + write(socket)：

磁盘 ──DMA拷贝①──► 内核 PageCache ──CPU拷贝②──► 用户缓冲区
                                                      │
Socket 网卡 ◄──DMA拷贝④── Socket 缓冲区 ◄──CPU拷贝③──┘

4 次拷贝（2 次 DMA + 2 次 CPU），4 次用户态/内核态切换

mmap + write（RocketMQ 使用）：

text

磁盘 ──DMA拷贝①──► 内核 PageCache （用户态直接映射，无需拷贝到用户空间）
                        │ CPU拷贝②
                        ▼
                  Socket 缓冲区 ──DMA拷贝③──► 网卡

3 次拷贝（2 DMA + 1 CPU），4 次切换。优势是【用户态可以直接读写文件内容】

sendfile（Kafka 使用）：

text

磁盘 ──DMA拷贝①──► 内核 PageCache ──（仅传递描述符）──► Socket 缓冲区 ──DMA拷贝②──► 网卡

2 次拷贝（全 DMA，支持 SG-DMA 时 CPU 完全不参与），2 次切换。但【用户态拿不到数据】

维度	mmap（RocketMQ）	sendfile（Kafka）
拷贝次数	3 次	2 次
用户态能否处理数据	能（映射到进程地址空间）	不能（数据不经过用户态）
适合的文件大小	小文件（单次映射上限约 1.5GB~2GB）	任意大小
为什么选它	RocketMQ 要在用户态解析消息、构建 ConsumeQueue 索引、做 TAG 过滤	Kafka 消费时整段转发，不需要解析内容

面试加分表述：「RocketMQ 选 mmap 而不是 sendfile，不是因为 mmap 更快——恰恰相反，sendfile 的拷贝次数更少。真正的原因是RocketMQ 需要在用户态操作消息内容：它要给每条消息建 ConsumeQueue 索引、要支持 Broker 端的 TAG 和 SQL92 过滤、要做半消息的 Topic 改写。这些都必须让 JVM 看到数据。Kafka 不需要，它的消费就是把一段连续的 log 原样推给客户端，所以能用 sendfile 做到完全不经过用户态。这也顺带解释了为什么 CommitLog 单文件固定 1GB——因为 mmap 一次映射的大小有限，1GB 是一个安全且对齐的选择。」

6. PageCache 与刷盘抖动：RocketMQ 的性能毛刺从哪来

RocketMQ 的高性能完全建立在 PageCache 上，它的性能毛刺也来自 PageCache：

text

写入路径（一切正常时）：
  消息 ──► mmap 映射的内存地址（即 PageCache）──► 返回成功（微秒级）
                                                 │
                                        后台 flush 线程 fsync 到磁盘

毛刺来源：
  ① PageCache 被占满 → 内核触发页回收 → 写入线程被阻塞（表现为 TPS 突然掉零）
  ② 消费冷数据 → 大量缺页中断 + 磁盘随机读 → 挤占 PageCache，连带影响写入
  ③ fsync 时磁盘繁忙 → flush 线程卡住 → PageCache 脏页堆积超过 dirty_ratio → 全局写阻塞

RocketMQ 的四个对抗手段：

手段	配置	作用
预分配文件（MappedFileQueue 预热）	`warmMapedFileEnable=true`	新建 CommitLog 时提前写满 0，让 PageCache 页先分配好，避免写入时缺页中断
文件锁定内存	`transientStorePoolEnable=true`	用堆外内存池做写缓冲（读写分离）：写进 DirectByteBuffer，由 commit 线程批量刷进 PageCache，避免读写争抢同一片 PageCache
mlock 锁页	`warmMapedFileEnable` 配合 `mlock`	防止 CommitLog 的 PageCache 被内核换出到 swap
过期文件清理	`fileReservedTime=72`（小时）、`deleteWhen=04`	凌晨 4 点删除 3 天前的 CommitLog，释放磁盘和 PageCache

⚠️ transientStorePoolEnable 开启后，消息先进堆外内存再进 PageCache，这中间多了一个丢失窗口：进程被 kill -9 时堆外内存的数据会丢。所以它只建议在「异步刷盘 + 主从同步」的组合下开启，纯异步单机不要开。

⚠️ 磁盘使用率超过 85%（diskMaxUsedSpaceRatio）时，Broker 会拒绝写入，报 service not available now, maybe disk full。超过 90% 会强制删除最早的文件。所以磁盘水位监控必须在 75% 就告警。

7. Broker 端消息过滤：TAG 与 SQL92

消息过滤的价值是减少网络传输和消费端的无效处理。RocketMQ 支持两种：

TAG 过滤（最常用）：

java

// 生产端：一条消息只能有一个 TAG
Message msg = new Message("ORDER_TOPIC", "PAID", body);

// 消费端：可以订阅多个 TAG，用 || 分隔
consumer.subscribe("ORDER_TOPIC", "PAID || REFUNDED");
consumer.subscribe("ORDER_TOPIC", "*");           // 订阅全部

TAG 过滤的实现原理（这是加分点）：

ConsumeQueue 的每条索引项有 8 字节的 TagsCode，存的是 TAG 字符串的 hashCode。
Broker 读 ConsumeQueue 时先比对 hashCode，不匹配就直接跳过，不去读 CommitLog——这是过滤能提升性能的根本原因。
但 hashCode 会哈希冲突，所以消费端 SDK 拿到消息后还会再比对一次 TAG 字符串做二次过滤。

text

Broker 端（粗筛）：比对 TagsCode 哈希 ──► 可能有误放行
        ▼
Consumer 端（精筛）：比对 TAG 字符串 ──► 保证准确

SQL92 过滤：

java

// 生产端：把过滤字段放进消息属性
msg.putUserProperty("orderAmount", "5000");
msg.putUserProperty("region", "HZ");

// 消费端：用 SQL 表达式过滤
consumer.subscribe("ORDER_TOPIC",
        MessageSelector.bySql("orderAmount > 1000 AND region IN ('HZ', 'SH')"));

维度	TAG 过滤	SQL92 过滤
性能	高（只比 8 字节哈希）	较低（要读消息属性并执行表达式）
表达能力	只能等值 / 或关系	支持 >、<、BETWEEN、IN、AND/OR/NOT、IS NULL
默认是否开启	开启	默认关闭，需 `enablePropertyFilter=true`
一条消息的维度	只能 1 个 TAG	可以多个属性

⚠️ 不要用过滤代替 Topic 划分。如果一个 Topic 里 95% 的消息都被过滤掉，说明 Topic 设计有问题——这些消息仍然占用了 CommitLog 的存储和 ConsumeQueue 的扫描。正确做法是拆成多个 Topic。过滤适合的是「同一业务实体的不同状态」这种 TAG 分布比较均匀的场景。

⚠️ RabbitMQ 的过滤是在 Exchange 路由阶段完成的（topic 通配符），消息根本不会进入不匹配的队列，这比 RocketMQ 的「进队列后过滤」更彻底，但灵活性不如 SQL92。

8. 消息堆积：排查顺序与应急扩容方案

第一步：先判断是「生产变快」还是「消费变慢」

bash

# RocketMQ：看消费者组的堆积量（diffTotal）和消费 TPS
sh mqadmin consumerProgress -n localhost:9876 -g order_consumer_group

# 输出关键列：
# BROKER_NAME  QID  BROKER_OFFSET  CONSUMER_OFFSET  DIFF  LAST_TIMESTAMP
# 看 DIFF 分布：是所有队列均匀堆积，还是只有个别队列堆积

现象	根因	处理
所有队列均匀堆积，生产 TPS 暴涨	上游流量洪峰	扩容消费者 + 临时提升消费线程数
所有队列均匀堆积，生产 TPS 不变	消费逻辑变慢（下游 RT 上升、慢 SQL、GC）	优化消费逻辑，先定位慢在哪
只有个别队列堆积	顺序消费被卡住，或该队列所在消费者挂了 / 卡死	看该消费者的线程栈，大概率是死锁或下游卡住
堆积量不变、`LAST_TIMESTAMP` 不更新	消费者完全没在消费（订阅关系不一致、Rebalance 异常、消费者未启动）	查订阅关系、查消费者注册情况

第二步：定位消费慢的原因（按概率排序）

下游依赖变慢：数据库慢 SQL、RPC 超时、第三方接口 RT 上升。看消费耗时监控的 P99。
消费线程数不够：consumeThreadMin/Max 默认 20，IO 密集型业务要调到 50~100。
批量大小不合理：consumeMessageBatchMaxSize 默认 1，能批量处理的业务（比如批量写库）调到 10~32 能有数量级提升。
队列数不够：消费者数 > 队列数，多余消费者空转。
GC 问题：消费端频繁 Full GC，拉取线程被 STW 阻塞。
顺序消费阻塞：某条消息永久失败，整个队列卡死。

第三步：应急扩容方案（面试必考）

当堆积量已经到百万级、且扩容消费者受队列数限制时，标准答案是临时转发扩容：

text

原架构（队列数 4，无法通过加机器提速）：
  ORDER_TOPIC(4 queues) ──► 4 个消费者（已到上限）

应急方案：
  ① 新建一个临时 Topic：ORDER_TOPIC_TEMP，设置 40 个队列
  ② 上线一个「搬运工」消费者：只做一件事——从 ORDER_TOPIC 拉消息，不做业务处理，
     直接转发到 ORDER_TOPIC_TEMP。它的消费速度极快（纯转发，无业务逻辑）
  ③ 上线 40个临时消费者消费 ORDER_TOPIC_TEMP，执行真正的业务逻辑
  ④ 堆积消化完后，下线临时链路，恢复原架构

     ORDER_TOPIC(4) ──► 搬运工(4) ──► ORDER_TOPIC_TEMP(40) ──► 临时消费者(40)

⚠️ 转发扩容的前提是业务可以乱序 + 幂等。顺序消息不能这么做，因为转发会打破分区顺序。顺序消息堆积的唯一解是优化单条消费耗时或者提前把队列数设大。

面试加分表述：「堆积处理我会分三个动作并行做：止血、提速、复盘。止血是先让业务可用——如果是非核心消息（比如发短信），直接跳过一部分（重置位点到最新） 比死扛更明智，因为用户已经不需要半小时前的短信了；提速是扩容和转发；复盘是看为什么监控没有提前告警。最关键的判断是这些堆积的消息还值不值得消费，很多团队花几小时消化完的消息，业务上其实早已失去意义。」

9. 消费者端限流与反压

消费端限流的目标是保护下游，不能因为 MQ 里堆了 100 万消息就用 100 万的并发去打数据库。

RocketMQ 的四层流控参数：

参数	默认值	作用
`pullThresholdForQueue`	1000	单队列本地缓存消息条数上限，超过则暂停拉取该队列
`pullThresholdSizeForQueue`	100 (MB)	单队列本地缓存消息大小上限
`consumeConcurrentlyMaxSpan`	2000	单队列中「最大 offset - 最小 offset」的跨度上限，防止某条消息处理太久导致位点无法推进
`pullInterval`	0	两次拉取的间隔（毫秒），设大了能降低消费速率

java

consumer.setConsumeThreadMin(20);
consumer.setConsumeThreadMax(64);
consumer.setPullBatchSize(32);                    // 单次从 Broker 拉取条数
consumer.setConsumeMessageBatchMaxSize(1);        // 单次回调给 Listener 的条数
consumer.setPullThresholdForQueue(1000);          // 本地缓存条数阈值，触发后暂停该队列拉取

RabbitMQ 的反压就是 prefetch：未 ack 消息达到 prefetch 后，Broker 停止向该 Channel 投递。这是比 RocketMQ 更直接的反压机制。

业务级限流（当 MQ 参数不够用时）：

java

// 用 Guava RateLimiter 在消费入口做令牌桶限流，保护下游数据库
private final RateLimiter rateLimiter = RateLimiter.create(500.0);   // 每秒 500 条

public ConsumeConcurrentlyStatus consume(List<MessageExt> msgs, ...) {
    // 拿不到令牌就阻塞等待（阻塞会占用消费线程，等价于降低消费速率）
    rateLimiter.acquire(msgs.size());
    // 注意：如果等待时间过长，RocketMQ 会认为消费超时（默认 15 分钟）
    return doConsume(msgs);
}

⚠️ 用「返回 RECONSUME_LATER 来限流」是错误做法。它会让消息进入重试 Topic，不仅增加 Broker 压力，还会让消息的投递时间变得不可控（10s → 30s → 1m 递增），最终大量消息进死信。限流应该用「阻塞消费线程」或「降低拉取速率」，而不是靠重试。

面试加分表述：「消费端限流的本质是让反压传导到 Broker。RocketMQ 的做法是「本地缓存满了就停止拉取」，RabbitMQ 的做法是「未 ack 超过 prefetch 就停止投递」——两者都是让消息安全地堆在 Broker 而不是堆在消费端内存里。堆在 Broker 是可控的（磁盘便宜），堆在消费端内存是会 OOM 的，这个取舍必须说清。」

10. RabbitMQ 高可用的三代方案

方案	原理	问题
普通集群（Classic Cluster）	节点间同步元数据（Exchange / Queue 定义），但队列内容只在一个节点上	队列所在节点宕机，该队列的消息不可用（非持久化则直接丢）；访问其他节点会被转发到队列宿主节点，有额外网络开销
镜像队列（Mirrored Queue）	队列内容在多个节点间同步，一主多从；主节点挂了从节点升主	全量同步，网络开销大；基于非正式的 GM 协议，网络分区时可能丢数据；3.8 后官方不再推荐
仲裁队列（Quorum Queue，3.8+）	基于 Raft 协议的强一致队列，过半写入成功才算成功	不支持优先级队列、不支持懒队列的部分特性、内存占用更高；性能略低于镜像队列

bash

# 配置镜像队列（已不推荐，但面试常问）
rabbitmqctl set_policy ha-all "^order\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'
# ha-mode: all（所有节点）/ exactly（指定数量）/ nodes（指定节点）
# ha-sync-mode: automatic（自动同步，新节点加入会阻塞队列）/ manual（手动）

java

// 声明 Quorum Queue（推荐）
@Bean
public Queue orderQuorumQueue() {
    return QueueBuilder.durable("order.queue")
            .quorum()                       // 等价于 x-queue-type = quorum
            .build();
}

⚠️ 镜像队列的 ha-sync-mode: automatic 有个大坑：新节点加入或节点重启后要同步全量数据，同步期间整个队列会被阻塞，队列里有几百万消息时可能阻塞几分钟。改成 manual 则新节点不会有历史消息（只同步新消息），需要人工触发 rabbitmqctl sync_queue。

⚠️ RabbitMQ 集群要求节点间网络延迟极低，官方明确说明不支持跨 WAN 组集群（跨机房要用 Federation 或 Shovel 插件做消息转发，而不是组集群）。Erlang 的分布式心跳对网络抖动极度敏感，跨机房很容易触发脑裂。

面试加分表述：「RabbitMQ 的高可用演进本质上是从「最终一致的 GM 协议」走向「强一致的 Raft」。镜像队列的问题在于它没有共识算法，网络分区恢复后只能靠 cluster_partition_handling 策略（pause_minority / autoheal / ignore）粗暴处理，autoheal 会直接丢掉少数派节点上的消息。Quorum Queue 用 Raft 把这个问题从根上解决了，代价是不再支持一些老特性。新项目我会直接用 Quorum Queue。」

11. RabbitMQ 的惰性队列 Lazy Queue

RabbitMQ 默认把消息尽可能保存在内存中（即使设了持久化，也会在内存留一份副本），这样投递最快。但堆积量大时内存会被打爆，触发内存告警（默认物理内存的 40%）后 Broker 会阻塞所有生产者连接。

java

@Bean
public Queue lazyQueue() {
    return QueueBuilder.durable("order.lazy.queue")
            .lazy()                     // x-queue-mode = lazy
            .build();
}

维度	默认队列（default）	惰性队列（lazy）
消息存放	优先内存，内存不足才换出到磁盘	直接写磁盘，消费时才读进内存
支持堆积量	百万级就有内存压力	千万级无压力
投递延迟	低	略高（每次都要读磁盘）
内存告警风险	高	极低

⚠️ RabbitMQ 触发内存告警（vm_memory_high_watermark，默认 0.4）后会阻塞所有生产者的 publish，表现为生产端线程全部卡在发送上，连锁导致上游服务线程池耗尽。所以预计会有大量堆积的队列必须设成 lazy，这是 RabbitMQ 最重要的一条生产规范。

⚠️ 同样还有磁盘告警（disk_free_limit，默认 50MB），磁盘剩余空间低于阈值时也会阻塞生产者。生产环境应该把它设成内存大小的 1~2 倍，比如 {mem_relative, 2.0}。

12. 消息去重的工程实现：三层防线

text

第一层：Redis 布隆过滤器（可选，超高并发场景）
    ├─ 作用：快速判断「肯定没处理过」，内存占用极小
    └─ 局限：有假阳性，判断为「可能处理过」时要走第二层确认

第二层：Redis SETNX（主力拦截）
    ├─ SET idempotent:{bizKey} {状态} NX EX 86400
    ├─ 返回 OK  → 首次处理，继续执行业务
    └─ 返回 nil → 已有记录，查状态：处理中 → 返回重试；已完成 → 直接 ack

第三层：数据库唯一索引（最终保证）
    ├─ UNIQUE KEY uk_biz (biz_type, biz_id)
    └─ 捕获 DuplicateKeyException → 视为成功，直接 ack

三个状态很关键，不能只存「已处理」：

java

public boolean tryConsume(String bizKey) {
    String key = "mq:idem:" + bizKey;
    // NX + 60 秒短过期：标记「处理中」
    Boolean first = redis.opsForValue().setIfAbsent(key, "PROCESSING", 60, TimeUnit.SECONDS);
    if (Boolean.TRUE.equals(first)) {
        return true;                                   // 拿到处理权
    }
    String status = redis.opsForValue().get(key);
    if ("DONE".equals(status)) {
        return false;                                  // 已完成，直接 ack，不再执行
    }
    // PROCESSING：说明另一个线程正在处理，或者上次处理崩了还没过期
    // 必须返回「稍后重试」而不是「直接 ack」，否则崩溃那次的业务就永久丢失了
    throw new RetryLaterException("消息正在处理中");
}

public void markDone(String bizKey) {
    // 处理成功后改成 DONE 并延长过期时间
    redis.opsForValue().set("mq:idem:" + bizKey, "DONE", 7, TimeUnit.DAYS);
}

⚠️ 只用 SETNX 标记「已处理」是有漏洞的：如果 SETNX 成功后业务执行失败，消息重试时会发现 key 已存在而被当成重复消息跳过，业务永久丢失。所以必须区分「处理中」和「已完成」两个状态，并且「处理中」要设短过期时间以便崩溃后能重入。

⚠️ 幂等 key 的过期时间要大于消息的最大重试周期。RocketMQ 的 16 次重试跨度是 2 小时多，如果幂等 key 只存 1 小时，最后几次重试就失去幂等保护了。建议至少 24 小时，重要业务 7 天。

面试加分表述：「最彻底的幂等是让业务操作本身天然幂等，根本不需要额外的去重表。比如「把订单状态改成已支付」写成 UPDATE t_order SET status = 2 WHERE order_no = ? AND status = 1，执行多少次结果都一样，返回影响行数为 0 就说明已经处理过了。能用状态机解决的就不要引入 Redis 和去重表——少一个组件就少一类故障。」

13. 重试与死信的边界：哪些异常该重试

把所有异常都丢进重试是新手最常见的错误。正确的做法是先对异常分类：

异常类型	例子	该不该重试	处理方式
瞬时异常	数据库连接超时、下游 503、网络抖动、乐观锁冲突	该重试	返回 `RECONSUME_LATER`，走 MQ 重试
业务异常	订单不存在、用户已注销、金额校验不过	不该重试（重试一万次也一样）	记录日志 + 直接 ack，必要时落异常表
数据异常	消息体反序列化失败、字段缺失	绝对不该重试	直接进死信，告警，等待修数据或改代码
幂等命中	消息已经处理过	不该重试	直接 ack

java

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext ctx) {
    MessageExt msg = msgs.get(0);
    try {
        handle(msg);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    } catch (BizException e) {
        // 业务异常：重试无意义，落异常表后 ack
        log.warn("业务异常，不重试，msgId={}, reason={}", msg.getMsgId(), e.getMessage());
        exceptionRecordService.save(msg, e);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    } catch (JsonParseException e) {
        // 数据异常：直接丢死信，不占用重试资源
        log.error("消息格式错误，msgId={}", msg.getMsgId(), e);
        deadLetterService.save(msg, e);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    } catch (Exception e) {
        // 瞬时异常：走重试，但要限制次数
        if (msg.getReconsumeTimes() >= 3) {
            log.error("重试 3 次仍失败，转人工，msgId={}", msg.getMsgId(), e);
            alarmService.send("消息消费失败", msg);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;   // ack 掉，避免继续重试 13 次
        }
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
}

⚠️ 不要让消息跑满 16 次重试再进死信。16 次重试跨度 2 小时多，这期间业务已经完全不可用了，而且重试会放大 Broker 压力。合理的做法是自己控制重试次数（3~5 次），超过就主动落异常表并告警。

面试加分表述：「我见过一个真实事故：下游接口改了字段名，导致所有消息反序列化失败。因为代码里是无脑 catch (Exception e) { return RECONSUME_LATER; }，结果 50 万条消息每条都重试 16 次，产生了 800 万次无效消费，Broker 的重试 Topic 被打爆，连带影响了同一个 Broker 上的其他业务。从那以后我们的消费模板强制要求按异常类型分支处理，并且重试次数由业务代码控制而不是依赖 MQ 的默认 16 次。」

14. 消费位点管理与位点重置

位点（offset）就是「这个消费者组在这个队列上消费到哪了」。

存储位置	RocketMQ	RabbitMQ
集群模式	Broker 端 `consumerOffset.json`，默认每 5 秒持久化一次	没有位点概念，用 ACK 删消息的模型
广播模式	消费者本地 `~/.rocketmq_offsets`	—

consumeFromWhere 的三个取值（只对新消费者组首次启动生效）：

取值	行为	风险
`CONSUME_FROM_LAST_OFFSET`（默认）	从队列最新位点开始，忽略历史消息	新上线的消费者收不到上线前的消息
`CONSUME_FROM_FIRST_OFFSET`	从队列最早的消息开始	新组上线会瞬间消费几天的历史消息，极易打爆下游
`CONSUME_FROM_TIMESTAMP`	从指定时间点开始（配合 `consumeTimestamp`）	需要精确指定时间

bash

# 位点重置（应急操作）：把消费者组的位点重置到某个时间点
sh mqadmin resetOffsetByTime -n localhost:9876 \
   -g order_consumer_group -t ORDER_TOPIC -s now        # now = 跳过所有堆积
sh mqadmin resetOffsetByTime -n localhost:9876 \
   -g order_consumer_group -t ORDER_TOPIC -s 1698400000000   # 回溯到指定毫秒时间戳

位点重置的两个典型用途：

应急跳过堆积：非核心业务堆积几百万条且已无业务价值时，-s now 直接跳到最新，秒级恢复。
消息回溯重放：代码 bug 导致一批消息处理错了，修复后把位点回拨到出问题的时间点重新消费（前提是业务幂等）。

⚠️ 位点重置是高危操作：-s now 会让堆积的消息永久不被消费（消息还在 CommitLog 里，但位点跳过去了）；回拨位点则会导致大量重复消费。两者都必须经过评审，并且重置前要先把消费者停掉（运行中重置可能被消费者的位点上报覆盖）。

⚠️ CONSUME_FROM_LAST_OFFSET 的坑：新消费者组上线时默认不消费历史消息。如果是「下游系统新接入，需要补历史数据」的场景，必须显式设成 CONSUME_FROM_FIRST_OFFSET 或 CONSUME_FROM_TIMESTAMP，否则会漏掉上线前的所有数据。

15. 延时消息的三种实现与 RocketMQ 5.0 的任意精度延时

方案	实现原理	精度	缺点
RocketMQ 延时等级	18 个固定等级，消息先投到 `SCHEDULE_TOPIC_XXXX` 的对应队列，定时任务扫描后转投真实 Topic	只能是 18 个固定值	不支持任意时间
RabbitMQ TTL + DLX	消息设 TTL 进入无消费者的队列，过期后通过 DLX 转到真实队列	任意	队头阻塞：队列级 TTL 下，队头消息没过期，后面先到期的也出不来
RabbitMQ 延时插件	`rabbitmq_delayed_message_exchange`，Broker 内部用 Erlang 的定时器	任意	消息存在内存中，不支持镜像，节点挂了消息丢
时间轮 / 外部调度	自己用 Netty `HashedWheelTimer` 或 Redis ZSet 存到期时间，到点再投 MQ	任意	需要自己保证高可用和持久化
RocketMQ 5.0 定时消息	基于时间轮 + 定时日志（TimerLog），支持任意秒级精度	任意（秒级）	需要 5.x 版本

java

// RocketMQ 4.x：只能用等级，level 3 = 10s
msg.setDelayTimeLevel(3);

// RocketMQ 5.x：任意时间戳
Message msg = messageBuilder
        .setTopic("ORDER_TIMEOUT")
        .setBody(body.getBytes())
        .setDeliveryTimestamp(System.currentTimeMillis() + 30 * 60 * 1000L)   // 30 分钟后
        .build();

⚠️ RabbitMQ 队列级 TTL 的队头阻塞是最容易踩的坑。假设队列里先进了一条 TTL=30 分钟的消息，后进了一条 TTL=1 分钟的消息，后者不会在 1 分钟后出队——因为 RabbitMQ 只检查队头消息是否过期。解决方案：要么用延时插件，要么为每个延时档位建一个独立队列（队列级统一 TTL，不混用消息级 TTL）。

面试加分表述：「我们的延时关单最早用 RabbitMQ 的 TTL+DLX，因为关单时间统一是 30 分钟，所以用的是队列级 TTL，天然不会队头阻塞。后来活动订单要支持 15 分钟/1 小时/24 小时多档，就改成了 RocketMQ 的延时等级 + 二次延时（超过最大等级 2h 的，先延 2h 再重新投递一条延时消息，循环直到到期）。现在新集群升到 5.x，直接用 setDeliveryTimestamp 一行搞定。」

16. 消息轨迹与可观测性

RocketMQ 消息轨迹（Message Trace）会把每条消息的生产、存储、消费全链路记录到一个内置 Topic RMQ_SYS_TRACE_TOPIC 中：

java

// 生产者：开启轨迹
DefaultMQProducer producer = new DefaultMQProducer("order_producer", true);
// 消费者：开启轨迹
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
        "order_consumer", null, new AllocateMessageQueueAveragely(), true, null);

轨迹记录的关键信息：

text

生产侧：msgId / topic / tags / keys / 发送耗时 / 发送结果 / 目标 Broker 和 Queue
存储侧：存储时间 / 存储 Broker / queueOffset / commitLogOffset
消费侧：消费者组 / 客户端 IP / 消费耗时 / 消费结果 / 重试次数

自己做全链路追踪的最小实现——把 traceId 放进消息属性，消费端取出来塞回 MDC：

java

// 生产端
msg.putUserProperty("traceId", MDC.get("traceId"));

// 消费端
String traceId = msg.getUserProperty("traceId");
MDC.put("traceId", traceId != null ? traceId : UUID.randomUUID().toString());
try {
    handle(msg);
} finally {
    MDC.clear();   // 线程池复用，必须清理，否则 traceId 会串
}

⚠️ 消费线程是线程池复用的，MDC 不清理会导致 traceId 串号，排查问题时看到的日志链路完全是错的。一定要在 finally 里 MDC.clear()。

⚠️ 开启轨迹会额外产生一倍左右的消息量（每条业务消息对应多条轨迹记录），对 Broker 的 IO 和磁盘有明显压力。生产环境建议只给核心 Topic 开启，或者用独立的轨迹 Topic 并单独配置保留时长。

17. 大消息与消息压缩

限制项	RocketMQ 默认值	说明
单条消息最大	4MB（`maxMessageSize`）	超过直接报 `MESSAGE_ILLEGAL`
压缩阈值	4KB（`compressMsgBodyOverHowmuch`）	超过自动用 zip 压缩 body
批量消息总大小	4MB	`send(List<Message>)` 的总和不能超

java

// 批量发送时必须自己做分片，否则总大小超限会整批失败
public void sendBatch(List<Message> messages) {
    ListSplitter splitter = new ListSplitter(messages, 4 * 1024 * 1024);
    while (splitter.hasNext()) {
        producer.send(splitter.next());
    }
}

大消息的正确处理方式——消息里只放引用：

text

❌ 错误：把 10MB 的 Excel 文件 base64 后放进消息体
    → 超过 4MB 直接发送失败；即使不超，也会严重拖慢 Broker 的刷盘和网络传输

✅ 正确：文件先传 OSS，消息里只放 URL + 文件元信息
    {
      "taskId": "T20261028001",
      "fileUrl": "https://oss.xxx.com/import/T20261028001.xlsx",
      "fileSize": 10485760,
      "md5": "a1b2c3..."
    }
    → 消息体几百字节，消费端按 URL 去拉文件

⚠️ 消息体越大，Broker 的吞吐下降越快。RocketMQ 的高性能建立在「消息小、顺序写」的前提上，一条 1MB 的消息相当于 1000 条 1KB 消息的 IO 量，还会挤占 PageCache，导致其他 Topic 的读也变慢。生产规范：单条消息建议控制在 4KB 以内，超过 64KB 就要考虑是不是设计有问题。

面试加分表述：「判断消息该不该大的标准是：消息是「事件通知」还是「数据传输」。MQ 的定位是事件通知，消息里应该只放「发生了什么 + 怎么找到数据」，而不是把数据本身搬过去。我们团队的规范是消息体只允许放业务主键和必要的轻量字段，需要完整数据的场景让消费端回查接口或走 OSS。」

18. Kafka 与 RocketMQ 的核心差异

维度	Kafka	RocketMQ
存储模型	每个 Partition 一个独立文件（分区级顺序写）	所有 Topic 共用 CommitLog（Broker 级顺序写）
分区/队列多了之后	文件数 = Topic × Partition，几千个分区后随机 IO 明显，吞吐暴跌	永远只写一个 CommitLog，万级队列性能不衰减
读取方式	直接读 Partition 文件，完全顺序读	先读 ConsumeQueue 拿偏移，再随机读 CommitLog（靠 PageCache 弥补）
延时消息	原生不支持	原生支持（18 等级 / 5.0 任意精度）
事务消息	支持（Exactly-Once 语义，偏流处理）	支持（半消息 + 事务回查，偏业务最终一致）
消息过滤	不支持 Broker 端过滤	支持 TAG 和 SQL92
消息重试/死信	需要自己建重试 Topic	原生支持
消息查询	不支持按 key 查	支持（IndexFile，按 msgId / key 精确查）
定位	日志、埋点、流式计算，超高吞吐	业务消息，功能丰富 + 低延迟稳定

text

Kafka 的取舍：为了极致吞吐，牺牲了业务功能和分区扩展性
    分区级顺序写 → 单分区读写都是纯顺序 → 吞吐极高
    但分区数一多，文件句柄和随机 IO 就爆了

RocketMQ 的取舍：为了功能完备和队列可扩展，牺牲了部分读性能
    Broker 级顺序写 → 队列数无上限 → 支持海量 Topic
    读 CommitLog 是随机的 → 靠 mmap + PageCache 预读弥补

面试加分表述：「选型我会看业务消息还是数据管道。日志采集、用户行为埋点、数据同步到数仓这类场景，量大、不要求单条可靠、不要求延时和事务，用 Kafka；订单、支付、库存这类业务消息，需要事务消息、延时消息、死信队列、消息查询，用 RocketMQ。两者并不互斥，我们公司实际是两套都在跑，按场景分流。」

19. 生产配置清单（Spring Boot 落地）

RocketMQ 关键配置：

yaml

rocketmq:
  name-server: 10.0.0.1:9876;10.0.0.2:9876       # 多个 NameServer，分号分隔
  producer:
    group: order_producer_group
    send-message-timeout: 3000                   # 发送超时，默认 3s
    retry-times-when-send-failed: 2              # 同步发送重试次数（会换 Broker）
    retry-times-when-send-async-failed: 2
    max-message-size: 4194304                    # 4MB
    compress-message-body-threshold: 4096        # 超过 4KB 压缩
  consumer:
    group: order_consumer_group
    consume-thread-min: 20                       # 消费线程池
    consume-thread-max: 64
    consume-message-batch-max-size: 1            # 一次回调给几条,幂等难做时保持 1
    pull-batch-size: 32                          # 一次从 Broker 拉几条
    max-reconsume-times: 3                       # 自己控制重试次数,别用默认 16

RabbitMQ 关键配置：

yaml

spring:
  rabbitmq:
    addresses: 10.0.0.1:5672,10.0.0.2:5672
    publisher-confirm-type: correlated           # 开启 confirm 回调
    publisher-returns: true                      # 开启 return 回调
    template:
      mandatory: true                            # 路由失败时触发 return 而不是丢弃
      retry:
        enabled: true
        initial-interval: 1000ms
        max-attempts: 3
    listener:
      simple:
        acknowledge-mode: manual                 # 手动 ACK
        prefetch: 10                             # 预取数,防止单消费者吃光消息
        concurrency: 5                           # 最小消费者数
        max-concurrency: 20
        retry:
          enabled: true
          max-attempts: 3
          default-requeue-rejected: false        # 重试耗尽后不重回队列,进死信

⚠️ default-requeue-rejected 默认是 true，这是 RabbitMQ 最凶险的默认值。消息消费失败后 nack 会被立刻重新放回队列队头，然后马上又被同一个消费者取到、又失败——形成毫秒级死循环，CPU 和日志瞬间打满。必须设成 false 并配置死信队列。

⚠️ prefetch 不设就是无限预取：一个消费者会把队列里所有消息都拉到本地内存，导致其他消费者空转、本地内存溢出，而且这些消息在消费者崩溃时才会重投，故障恢复极慢。生产必须设置，一般 10~50。

20. 监控告警指标清单

分类	指标	告警阈值建议	说明
堆积	消费组堆积量 `diffTotal`	核心 > 1 万，非核心 > 10 万	最重要的指标
堆积	消费延迟（最后一条消息存储时间到现在）	> 5 分钟	比条数更能反映业务影响
生产	发送失败率	> 0.1%	看 Broker 健康
生产	发送 RT P99	> 500ms	刷盘抖动的先兆
消费	消费失败率 / 重试消息量	重试量突增 2 倍	下游异常的早期信号
消费	死信队列消息数	> 0 就告警	死信代表确定性的业务损失
Broker	CommitLog 磁盘使用率	> 75%	超过 `diskMaxUsedSpaceRatio`（默认 88%）会拒绝写入
Broker	PageCache 写入耗时（`PCRT`）	> 100ms	磁盘性能告急
Broker	主从同步落后字节数	> 100MB	主挂了会丢这部分数据
RabbitMQ	内存水位	> 0.35（阈值 0.4）	到阈值会阻塞生产者
RabbitMQ	`unacked` 消息数	持续增长不回落	消费者卡死或忘记 ack
RabbitMQ	队列消息数 `messages_ready`	按队列设阈值	配合 lazy queue 判断

bash

# RocketMQ 日常排查三条命令
sh mqadmin consumerProgress -n localhost:9876 -g order_consumer_group   # 看堆积
sh mqadmin topicStatus -n localhost:9876 -t ORDER_TOPIC                 # 看各队列位点
sh mqadmin queryMsgById -n localhost:9876 -i <msgId>                    # 查单条消息全链路

⚠️ 只监控「堆积条数」是不够的。一个 QPS 1 万的 Topic 堆积 1 万条只是 1 秒的延迟，而一个 QPS 10 的 Topic 堆积 1 万条意味着延迟 16 分钟。「消费延迟时间」才是和业务影响强相关的指标，告警应该以时间为主、条数为辅。

面试加分表述：「MQ 的监控我分三层：资源层（Broker 磁盘、PageCache 耗时、主从落后）、链路层（发送失败率、发送 RT、消费失败率）、业务层（消费延迟时间、死信数量）。其中死信队列只要有消息就必须告警，因为死信代表的是「确定性的业务数据损失」，而堆积只是「暂时的延迟」，两者的紧急程度完全不同。」

❓ 高频面试题【基础简答 + 连环追问链 + 精简口述答案 + 项目关联话术】

题 1：为什么要用消息队列？引入 MQ 会带来哪些问题？

精简口述答案（90 秒版）

用 MQ 核心是三件事：异步、解耦、削峰。异步是把非核心链路（发短信、推送、同步数仓）从主流程里摘出去，下单接口从 800ms 降到 200ms；解耦是上游只管发事件，下游爱怎么接怎么接，新增下游不用改上游代码；削峰是秒杀时把瞬时 10 万 QPS 的请求先堆在 MQ 里，后端按自己 2000 QPS 的能力匀速消费。

但引入 MQ 的代价也很明确，四个问题：可用性降低（MQ 挂了全链路瘫，必须集群部署）、复杂度上升（要处理重复消费、消息丢失、顺序性、堆积）、一致性问题（上游以为发成功了，下游其实处理失败）、排查成本上升（链路从一次 RPC 变成异步多跳，必须做消息轨迹）。

所以判断该不该用 MQ 的标准是：这个操作是不是「必须同步拿到结果」。必须同步的（扣库存、扣款）不能用 MQ；不必须的（通知、统计、同步）才用。

连环追问链

追问	答题要点
异步化之后怎么知道下游处理成功了？	下游处理完回发一条「结果消息」或写状态表，上游做状态轮询/回调；关键业务要有对账补偿任务
削峰时消息堆在 MQ 里，用户一直等着怎么办？	削峰只适合「可异步返回结果」的场景。秒杀是先返回「排队中」，前端轮询结果页，不是让用户阻塞等待
MQ 挂了业务要不要降级？	要。发送失败时降级写本地消息表，MQ 恢复后由定时任务补发；核心链路绝不能因为 MQ 挂了就报错
不用 MQ 能不能解耦？	能用事件总线/本地事件，但只在单进程内有效；跨服务解耦必须有中间存储，MQ 是最成熟的选择
什么场景不适合用 MQ？	强一致要求的事务操作、需要立即返回结果的查询、调用链只有一跳且本身很快的场景
一个请求里既要同步又要异步怎么设计？	同步部分走 RPC 并在本地事务内完成，异步部分在事务提交后发 MQ（或用事务消息/本地消息表保证不丢）

✅ 项目关联话术

我们的下单链路原来是串行调用「扣库存 → 创建订单 → 发短信 → 推送 App → 同步数仓 → 加积分」，接口 P99 到了 900ms。重构时我把链路切成两段：扣库存和创建订单必须同步（强一致，失败要回滚），后面四个全部改成发一条 ORDER_CREATED 事件到 RocketMQ，由四个独立消费者组各自订阅。改完下单接口 P99 降到 180ms，而且后来新增「发优惠券」的需求时，上游代码一行没改，只加了一个消费者组。

题 2：怎么保证消息不丢失？

精简口述答案（90 秒版）

消息会在三个地方丢，必须三处都兜住：

第一处，生产者到 Broker。网络可能抖、Broker 可能挂。解决方案：RocketMQ 用同步发送 + 失败重试（send() 返回 SEND_OK 才算成功，默认重试 2 次且会切换 Broker）；RabbitMQ 用 publisher confirm（异步回调确认）+ return callback（路由失败回调）。更彻底的是本地消息表：业务和消息记录在同一个事务里落库，再由定时任务投递。

第二处，Broker 自身。消息在 PageCache 里还没落盘时宕机就丢了。解决方案：同步刷盘（flushDiskType=SYNC_FLUSH）+ 主从同步复制（brokerRole=SYNC_MASTER）。RabbitMQ 对应的是队列持久化 + 消息持久化 + 镜像/Quorum 队列，三个都要开。

第三处，Broker 到消费者。消费者拿到消息还没处理完就崩了。解决方案：手动 ACK，业务真正处理完才确认；RocketMQ 返回 CONSUME_SUCCESS，RabbitMQ 调 basicAck。失败则返回 RECONSUME_LATER / basicNack 进重试，重试耗尽进死信。

最后还要有兜底对账：定时任务扫「应该有下游结果但没有」的数据，主动补偿。因为无论配置多严，100% 不丢是不存在的。

连环追问链

追问	答题要点
同步刷盘 + 同步复制性能损失多大？	同步刷盘 RT 从 1ms 升到 5~10ms，吞吐降 50% 左右；一般只有资金类业务开，普通业务用异步刷盘 + 同步复制折中
异步刷盘 + 同步复制为什么是最优折中？	消息至少在两台机器的内存里，单机宕机不丢（另一台能顶）；只有两台同时断电才丢，概率极低
publisher confirm 是同步还是异步？	推荐异步 confirm（`correlated` 模式），吞吐最高；同步 `waitForConfirms` 会阻塞，吞吐只有异步的 1/10
消费者手动 ACK 后业务回滚了怎么办？	ACK 要放在事务提交之后；如果业务是分布式事务，用本地消息表 + 对账补偿，不要指望 MQ 帮你回滚
本地消息表和事务消息怎么选？	本地消息表不依赖 MQ 特性、任何 MQ 都能用、可靠性最高但要自己写定时任务；事务消息代码更简洁但只有 RocketMQ 支持
为什么说 100% 不丢做不到？	任何确认机制都存在「确认丢失」的中间态（发了但没收到 ack），系统只能做到不丢 + 可能重复，所以必须配合幂等和对账

✅ 项目关联话术

我们支付回调通知的链路是「MQ 配置 + 对账」双保险。MQ 侧：生产端同步发送 + 本地消息表（支付成功和消息记录在同一个事务里 insert），Broker 侧异步刷盘 + 同步双副本，消费端手动 ACK + 3 次重试 + 死信告警。对账侧：每 5 分钟扫一次「支付成功超过 10 分钟但订单状态还是待支付」的记录，主动触发补偿。上线一年多，MQ 侧丢过 2 次（都是 Broker 主从切换时的边界情况），但都被对账任务在 5 分钟内兜住了，业务零感知。

题 3：怎么保证消息不被重复消费（幂等）？

精简口述答案（90 秒版）

首先要明确一个前提：MQ 做不到 Exactly Once，重复消费是必然的。原因有三个：生产端重试会产生内容相同的两条消息、消费端 ACK 丢失会导致重投、Rebalance 时位点还没提交就被别人接管了。所以不要指望 MQ，必须在消费端做幂等。

幂等的核心是找到一个业务唯一标识。注意不要用 msgId，因为生产端重试产生的是两个不同 msgId 的消息，用 msgId 根本挡不住。要用业务 key，比如订单号、支付流水号，或者「订单号 + 操作类型」的组合。

实现上我用三层防线：第一层 Redis SETNX，SET idem:{bizKey} PROCESSING NX EX 60，拿到才执行，并且要区分「处理中」和「已完成」两个状态——只标记「已处理」的话，业务执行失败后重试会被误判为重复而永久丢失。第二层数据库唯一索引，UNIQUE KEY (biz_type, biz_id)，捕获 DuplicateKeyException 视为成功。第三层状态机，UPDATE t_order SET status=2 WHERE order_no=? AND status=1，影响行数 0 说明已处理过。

最优雅的方案其实是第三层：让业务操作本身天然幂等，就不需要额外组件了。

连环追问链

追问	答题要点
为什么不能用 msgId 做幂等 key？	生产端重试会生成新的 msgId，内容相同但 ID 不同，msgId 挡不住；只能挡 Broker 重投那一种重复
Redis 幂等 key 该设多久过期？	必须大于消息最大重试周期。RocketMQ 16 次重试跨度 2 小时多，建议至少 24 小时，资金类 7 天
Redis 挂了幂等就失效了吗？	是，所以数据库唯一索引是最终防线，不能只靠 Redis；Redis 的作用是拦住 99% 的重复，减轻 DB 压力
「处理中」状态为什么不能直接 ack？	直接 ack 会让「上次处理崩了」的消息永久丢失。应该抛异常走重试，等「处理中」的短过期到了就能重入
分布式下两个消费者同时拿到同一条消息？	`SETNX` 本身是原子的，只有一个能成功；另一个会看到 `PROCESSING` 并走重试
批量消费怎么做幂等？	要么拆成单条处理（`consumeMessageBatchMaxSize=1`），要么整批共用一个批次 key 并保证整批在一个事务里

✅ 项目关联话术

我们加积分的消费者最早只用 Redis SETNX 标记「已处理」，结果出过一次问题：某次下游积分服务超时，SETNX 已经成功但积分没加上，消息重试时被判定为重复直接跳过，用户积分永久丢失，客诉了十几单。复盘后改成了三层方案：Redis 存 PROCESSING（60 秒）/ DONE（7 天）两个状态，积分流水表加 UNIQUE KEY (user_id, order_no, type)，并且加积分的 SQL 改成带状态条件的更新。之后再没出过重复或丢失。

题 4：死信队列是什么？哪些场景会进死信？

精简口述答案（90 秒版）

死信队列是消息的最终垃圾桶：消费失败到重试耗尽、或者因为各种原因无法被正常投递的消息，会被转到一个专门的队列里，等人工介入。

RabbitMQ 的死信（DLX）有三个触发条件：消息被 basicNack/basicReject 且 requeue=false；消息 TTL 过期；队列达到 x-max-length 上限后溢出。配置方式是在队列上声明 x-dead-letter-exchange 和 x-dead-letter-routing-key。有意思的是，TTL 过期进 DLX 这个特性被反过来用作延时队列的实现。

RocketMQ 的死信（DLQ）只有一个条件：重试达到 maxReconsumeTimes（默认 16 次）。死信 Topic 的命名是 %DLQ%消费者组名，而且默认权限是只写不可读，要用 mqadmin updateTopicPerm 改成 6（读写）才能消费。

死信的正确用法是：只要有消息进死信就立刻告警。因为死信代表「确定性的业务数据损失」，和「堆积」这种暂时的延迟完全不是一个紧急程度。我们的做法是死信 Topic 单独挂一个消费者，收到就落异常表 + 企微告警 + 提供后台重投按钮。

连环追问链

追问	答题要点
死信里的消息怎么处理？	先定位根因（数据问题/代码 bug/下游故障），修复后重投；重投前必须确认消费端幂等，否则会造成重复业务
RocketMQ 死信 Topic 读不到怎么办？	`sh mqadmin updateTopicPerm -t %DLQ%组名 -p 6`，默认 perm=2 只写
死信消息会一直保留吗？	RocketMQ 死信消息保留 3 天（跟 CommitLog 的 `fileReservedTime`）；RabbitMQ 的死信队列就是普通队列，不设 TTL 会一直存
要不要让消息跑满 16 次重试？	不要。16 次跨度 2 小时多，期间业务已经不可用了。应该业务代码控制 3~5 次，超过主动落异常表告警
死信队列本身也消费失败怎么办？	死信消费者只做「落库 + 告警」这类绝对不会失败的操作，不要在死信消费者里再调业务接口
RabbitMQ 为什么 `requeue=true` 很危险？	消息会被立刻放回队头并马上被同一消费者再取到，形成毫秒级死循环，CPU 和日志瞬间打满

✅ 项目关联话术

我们踩过 default-requeue-rejected 默认 true 的坑。一条脏数据导致反序列化失败，nack 后消息立刻回到队头又被取到，每秒循环上万次，10 分钟打出 40GB 日志把磁盘写满，连带整个 Pod 挂了。之后我们统一了消费模板：default-requeue-rejected: false、所有业务队列必须声明 DLX、死信队列统一接到一个 dead-letter-collector 服务落异常表并告警，运营后台提供重投入口。

题 5：RocketMQ 事务消息的完整流程是什么？

精简口述答案（90 秒版）

事务消息解决的是**「本地事务和发消息」的原子性**问题：不能出现「本地事务提交了但消息没发出去」或者「消息发了但本地事务回滚了」。

完整流程六步：

第一步，生产者发**半消息（half message）**到 Broker。半消息会被存到一个内部 Topic RMQ_SYS_TRANS_HALF_TOPIC 里，消费者看不到。

第二步，Broker 返回半消息发送成功。

第三步，生产者执行本地事务（在 executeLocalTransaction 回调里）。

第四步，生产者根据本地事务结果向 Broker 发送二次确认：COMMIT_MESSAGE 或 ROLLBACK_MESSAGE。

第五步，Broker 收到 COMMIT 后把半消息从 half Topic 恢复到真实 Topic，消费者就能消费了；收到 ROLLBACK 就把半消息标记为删除（写一条到 RMQ_SYS_TRANS_OP_HALF_TOPIC）。

第六步，事务回查（关键）。如果第四步的二次确认丢了（生产者崩了、网络断了），Broker 会定时扫描 half Topic 里超时未确认的消息，主动回调生产者的 checkLocalTransaction 方法问「这个事务到底成没成」，根据返回结果再决定提交还是回滚。默认每 60 秒回查一次，最多回查 15 次，超过就直接回滚。

所以事务消息的本质是两阶段提交 + 回查补偿，而 checkLocalTransaction 是整个机制的兜底，必须实现且必须可靠。

连环追问链

追问	答题要点
回查方法怎么实现才靠得住？	必须能独立查到本地事务状态，通常是查业务表（订单是否存在）或查专门的事务状态表；不能依赖内存变量，因为生产者可能已经重启了
回查 15 次还确定不了怎么办？	Broker 直接 ROLLBACK 丢弃半消息。所以回查逻辑一定要在15 分钟内能给出确定答案，否则消息就没了，要靠对账兜
事务消息保证下游一定消费成功吗？	不保证。它只保证「本地事务成功 ⇒ 消息一定能被投递」，下游消费失败要靠重试和死信；最终一致还要靠下游幂等
事务消息和本地消息表的区别？	事务消息把「消息暂存」放在 Broker（半消息），本地消息表放在自己的 DB；前者代码简洁，后者不依赖 MQ 特性且可观测性更好
事务消息支持顺序吗？	不支持。半消息恢复到真实 Topic 的顺序和发送顺序无关，需要顺序就不能用事务消息
性能开销大吗？	一次业务要写两次 Broker（半消息 + 二次确认恢复），吞吐大约是普通消息的一半，且回查会额外增加 Broker 负载

✅ 项目关联话术

我们「创建订单 + 通知库存预占」这个链路用的是事务消息。executeLocalTransaction 里执行订单入库，成功返回 COMMIT_MESSAGE；checkLocalTransaction 里按消息属性带的 orderNo 查订单表，查到就 COMMIT、查不到就 ROLLBACK、查到但状态是「创建中」就返回 UNKNOW 等下次回查。这里有个细节很重要：回查靠的是「订单表里有没有这条记录」而不是任何内存状态，这样哪怕生产者 Pod 被重建了，回查依然能给出正确答案。

题 6：怎么保证消息的顺序性？

精简口述答案（90 秒版）

乱序有三个并发点，必须三个都解决：

第一个，发送端并发。多线程发送时到达 Broker 的顺序就乱了 → 解决：同一业务 key 的消息串行发送，或者直接用同步发送。

第二个，队列分散。默认轮询投递会把同一订单的消息打到不同队列 → 解决：用 MessageQueueSelector 按业务 key 哈希，保证同一订单的消息永远进同一个队列。

第三个，消费端并发。即使在同一队列，默认的 MessageListenerConcurrently 会用线程池并发消费 → 解决：换成 MessageListenerOrderly，它会对队列加锁，保证同一队列同一时刻只有一个线程在消费。

关键认知是：RocketMQ 只能做到「分区有序」，做不到「全局有序」。全局有序要求 Topic 只有 1 个队列、1 个消费者、单线程消费，吞吐会降到几百 QPS，几乎没有生产价值。而实际业务要的从来都是分区有序——同一个订单的状态变更有序就够了，不同订单之间不需要有序。

RabbitMQ 没有分区概念，要实现顺序只能一个队列对一个单并发消费者（prefetch=1 + concurrency=1），想提高吞吐就得按业务 key 哈希到多个队列，每个队列一个消费者。

连环追问链

追问	答题要点
顺序消费时一条消息卡住了会怎样？	整个队列全部阻塞，后面的消息都消费不了。必须设置重试上限，超过就跳过（落异常表）而不是无限重试
顺序消息的吞吐怎么提升？	加队列数 + 按 key 哈希，队列之间并行、队列内部有序；但队列数一旦改变，哈希结果就变了，需要停机或双写过渡
扩容队列数会破坏顺序吗？	会。`hash % queueCount` 的结果变了，同一订单的新旧消息可能在不同队列，存量消息消费完之前不能扩容
顺序消费 + Rebalance 会冲突吗？	会有短暂阻塞。顺序消费靠的是 Broker 端的队列锁，Rebalance 时旧消费者要先释放锁，新消费者才能拿到，中间有秒级空档
业务上怎么避免依赖顺序？	消息里带版本号/状态机。消费时判断「当前状态是否允许这个变更」，乱序到的旧消息直接丢弃，这样就不需要 MQ 保证顺序了
如果必须全局有序呢？	Topic 设 1 个队列 + 1 个消费者实例 + Orderly 监听器。要接受吞吐只有几百 QPS，而且该消费者是单点

✅ 项目关联话术

我们的订单状态同步最早用的是顺序消息（按 orderNo 哈希到 32 个队列 + MessageListenerOrderly），但出过一次事故：一条消息因为下游接口超时反复重试，把整个队列堵了 40 分钟，这个队列上 1/32 的订单状态全部停止更新。复盘后我们改成了乱序 + 状态机校验：消息体里带 version 字段和 fromStatus，消费时执行 UPDATE t_order SET status=?, version=? WHERE order_no=? AND version < ?，乱序到的旧消息影响行数为 0 自然被丢弃。这样既不怕乱序，也不会因为一条消息卡死整个队列。

题 7：线上堆积了几百万条消息怎么办？

精简口述答案（90 秒版）

分两步：先定位原因，再选应急方案。

定位三步：第一步看是全部队列堆积还是部分队列堆积（mqadmin consumerProgress）。部分队列堆积说明是数据倾斜或某个消费者实例挂了/卡死；全部队列堆积说明是消费能力整体不足或下游故障。第二步看消费者日志和线程栈（jstack），确认线程是卡在数据库、卡在 RPC，还是根本没在消费。第三步看生产端是否流量突增（对比发送 TPS 曲线）。

应急方案按原因选：

如果是消费者挂了/卡死，重启就能恢复，堆积会自然追上。

如果是消费能力不足且队列数够（比如 32 个队列只有 4 个消费者），直接水平扩容消费者实例到队列数，这是最快的办法。

如果是队列数不够（队列数就是并发上限，加机器没用），用搬运工方案：临时写一个消费者，只做「读出消息 → 转发到一个队列数更多的新 Topic」不做任何业务处理，再用大量消费者消费新 Topic。

如果是下游是瓶颈（比如数据库写不动），扩消费者只会把下游压死，应该改成批量处理 + 攒批写库，把 1 条 1 次 insert 变成 500 条 1 次 batch insert。

如果是非核心业务且消息已无价值，评审后用 mqadmin resetOffsetByTime -s now 直接跳过堆积，秒级恢复。

事后一定要做两件事：加堆积告警（以消费延迟时间为主指标）和做限流/批量优化，避免再次发生。

连环追问链

追问	答题要点
为什么加消费者实例不一定有效？	消费者数超过队列数时，多出来的消费者分配不到队列，完全空转。并发上限由队列数决定
搬运工方案的新 Topic 队列数怎么定？	按「需要的并发数」定，一般设成原来的 4~8 倍；注意新 Topic 也要能被足够多的消费者实例覆盖
搬运时顺序会乱吗？	会。搬运工方案不适用于顺序消息，顺序 Topic 只能扩容队列数（且要等存量消费完）或优化单条处理耗时
跳过堆积有什么风险？	消息还在 CommitLog 里但位点跳过了，相当于永久不消费。必须评审确认业务可接受，核心业务绝对不能跳
消费延迟和堆积条数哪个更重要？	延迟时间。QPS 1 万的 Topic 堆 1 万条只延迟 1 秒，QPS 10 的 Topic 堆 1 万条延迟 16 分钟，后者才是事故
怎么预防堆积？	容量规划（队列数 = 峰值 QPS / 单消费者 QPS × 2）、消费端限流、下游熔断降级、堆积分级告警

✅ 项目关联话术

有次大促，我们的「订单同步数仓」消费者堆积了 400 万条。排查发现是数仓的 ClickHouse 侧限流，单条 insert RT 从 10ms 涨到 300ms。因为这个 Topic 只有 8 个队列，扩消费者实例到 8 个之后就到顶了，吞吐只有 260 QPS，追不上。我们的处理是：改批量——把 consumeMessageBatchMaxSize 调到 100，消费端攒批后一次 batch insert，单批 RT 400ms，等效 QPS 提到 2000，2 小时追平了堆积。事后把这个 Topic 的队列数从 8 扩到 32，并且把批量消费固化到代码里。

题 8：延时消息怎么实现？延时关单你是怎么做的？

精简口述答案（90 秒版）

四种实现，各有取舍：

RocketMQ 延时等级：4.x 只支持 18 个固定等级（1s/5s/10s/30s/1m/2m/3m/4m/5m/6m/7m/8m/9m/10m/20m/30m/1h/2h）。原理是消息先投到内部 Topic SCHEDULE_TOPIC_XXXX 的第 level-1 号队列，Broker 的定时任务每秒扫描，到期后恢复 Topic 再投递。缺点是不支持任意时间，而且最大只有 2 小时。超过 2 小时要用「二次延时」——先延 2 小时，到期后再发一条延时消息，循环到期。

RocketMQ 5.0 定时消息：基于时间轮 + TimerLog，setDeliveryTimestamp 支持任意秒级精度，这是最优解。

RabbitMQ TTL + DLX：消息设 TTL 进一个没有消费者的队列，过期后通过死信交换机转到真实队列。这里有个致命坑：消息级 TTL 会队头阻塞——RabbitMQ 只检查队头消息是否过期，如果队头是 TTL=30 分钟的消息，后面 TTL=1 分钟的消息也要等 30 分钟。所以必须用队列级 TTL（一个延时档位一个队列）。

RabbitMQ 延时插件：rabbitmq_delayed_message_exchange，支持任意时间，但消息存在节点内存里、不支持镜像队列，节点挂了消息就丢，不适合可靠性要求高的场景。

我们的延时关单最早用 RabbitMQ 队列级 TTL（固定 30 分钟），后来支持多档位就迁到了 RocketMQ 延时等级。还有个重要细节：延时消息只是触发器，不能直接关单。消费时必须回查订单当前状态，只有还是「待支付」才关，否则已支付的订单会被误关。

连环追问链

追问	答题要点
为什么不用 Redis ZSet 做延时？	可以，`score` 存到期时间戳 + 定时任务 `ZRANGEBYSCORE` 扫描。但要自己保证高可用、防重复扫描（分布式锁）、持久化，运维成本高于直接用 MQ
为什么不用数据库定时扫表？	量小可以。但订单表上千万后，`WHERE status=1 AND create_time < ?` 的扫表会很慢，而且扫描周期决定了精度（分钟级）
延时消息会不会不准？	会有秒级抖动。RocketMQ 的定时任务是每秒扫描，加上投递和消费耗时，实际延迟比设定值晚 1~3 秒，业务要容忍
关单消息丢了订单就关不掉了？	所以必须有兜底扫表任务，每小时扫一次「超时未支付」的订单，MQ 是主路径、扫表是兜底
支付和关单并发怎么办？	关单 SQL 必须带状态条件 `UPDATE ... SET status=5 WHERE order_no=? AND status=1`，靠数据库行锁 + 状态条件保证互斥
大量延时消息对 Broker 有压力吗？	有。延时消息都集中在 `SCHEDULE_TOPIC_XXXX` 的 18 个队列里，同一秒到期的消息过多会造成投递毛刺，建议对到期时间做随机抖动打散

✅ 项目关联话术

我们的延时关单是「MQ 主路径 + 扫表兜底 + 状态机互斥」三件套。下单成功后发一条延时 30 分钟的 RocketMQ 消息；消费时先按 orderNo 查订单，状态不是「待支付」直接 ack 返回；是「待支付」则执行 UPDATE t_order SET status=5 WHERE order_no=? AND status=1，影响行数 0 说明被支付抢先了，也直接 ack。另外有个每小时跑一次的兜底任务，扫「创建超过 35 分钟还待支付」的订单补偿关单。还有个细节：大促时同一秒下单量很大，关单消息会集中到期，我们给延时时间加了 0~60 秒的随机抖动，把投递峰值打散了。

题 9：RabbitMQ 有哪几种 Exchange？分别什么场景用？

精简口述答案（90 秒版）

四种：

Direct（直连）：routingKey 和 bindingKey 完全相等才路由。最简单，适合点对点的明确投递，比如「订单服务 → 库存服务」。

Topic（主题）：支持通配符，* 匹配一个单词，# 匹配零个或多个单词。比如 order.* 能匹配 order.create 但匹配不了 order.create.vip，而 order.# 两个都能匹配。适合按业务维度灵活订阅，是生产中最常用的。

Fanout（扇出）：忽略 routingKey，广播给所有绑定的队列。性能最好（不需要路由计算），适合一个事件通知多个下游，比如「用户注册 → 发券 + 发短信 + 建画像」。

Headers（头匹配）：按消息 header 的键值对匹配，x-match=all 要求全部匹配、x-match=any 要求任一匹配。性能最差，生产基本不用，Topic 能满足的场景都用 Topic。

还有一个概念要知道：默认交换机（Default Exchange），名字是空字符串的 Direct 交换机，所有队列都自动用队列名作为 bindingKey 绑定到它上面。所以 convertAndSend("queueName", msg) 这种只传队列名的写法能工作，就是走了默认交换机。

选型上我的经验是：不确定就用 Topic。Topic 可以退化成 Direct（不用通配符）、也能模拟 Fanout（用 # 绑定），灵活性最高，后期加下游不用改交换机类型。

连环追问链

追问	答题要点
`*` 和 `#` 的区别？	`` 严格匹配一个单词（点分隔），`#` 匹配零个或多个单词。`order.#` 能匹配 `order` 本身，`order.` 不能
Fanout 比 Topic 快多少？	Fanout 不做路由键计算，差距在 10%~30% 之间；但实际瓶颈通常在网络和磁盘，这个差距一般可以忽略
消息路由不到任何队列会怎样？	默认静默丢弃。必须设 `mandatory=true` 并实现 `ReturnCallback` 才能感知，否则是无声的消息丢失
一个队列能绑多个交换机吗？	能，一个队列可以绑定到多个交换机、也可以用多个 bindingKey 绑到同一个交换机
多个 bindingKey 都匹配会收到多条吗？	不会。同一队列即使多个绑定都匹配，也只会收到一条消息，RabbitMQ 会去重
备份交换机是什么？	`alternate-exchange`，路由失败的消息会转到备份交换机而不是丢弃，比 `mandatory` 回调更可靠的兜底方案

✅ 项目关联话术

我们的事件总线统一用 Topic 交换机，routingKey 规范是 {领域}.{实体}.{动作}，比如 trade.order.created、trade.order.paid、user.account.registered。下游按需绑定：订单搜索服务绑 trade.order.# 收全部订单事件，短信服务只绑 trade.order.paid。这套规范的好处是新增事件和新增下游都不需要改交换机配置。另外我们把 mandatory=true 和 ReturnCallback 设成了全局默认，routingKey 写错时会立刻在日志里报警，而不是消息无声消失。

题 10：RocketMQ 为什么用 CommitLog 混合存储？和 Kafka 的分区存储比有什么取舍？

精简口述答案（90 秒版）

Kafka 是「分区级顺序写」：每个 Partition 一个独立的日志文件，写入和读取都是纯顺序，性能极高。但文件数 = Topic 数 × 分区数，分区数到几千之后，多个文件的「顺序写」在磁盘层面就退化成了随机写，吞吐会暴跌。

RocketMQ 是「Broker 级顺序写」：所有 Topic 的所有消息都混合追加写到同一个 CommitLog（1GB 一个文件，顺序滚动）。这样无论有多少 Topic 和队列，磁盘上永远只有一个文件在被顺序写，队列数到万级性能也不衰减。

代价是读变复杂了。消费者要先读 ConsumeQueue（每条记录固定 20 字节：commitLogOffset 8 + size 4 + tagHashCode 8）拿到消息在 CommitLog 里的物理偏移，再随机读 CommitLog。RocketMQ 靠两个手段弥补：一是 mmap 把文件映射到用户空间，读取走 PageCache 不经过内核拷贝；二是 PageCache 的预读机制，读一个位置会把周围 128KB 一起读进来，而混合存储下相邻的消息往往是同一时间段写入的，局部性其实不差。

所以这是一个明确的取舍：Kafka 用「牺牲队列可扩展性」换「极致吞吐」，RocketMQ 用「牺牲部分读性能」换「海量 Topic + 功能完备」。RocketMQ 之所以敢这么设计，是因为它的定位是业务消息——业务场景的 Topic 数量多（每个业务一个 Topic）、但单 Topic 吞吐没有日志场景那么极端。

连环追问链

追问	答题要点
ConsumeQueue 为什么是固定 20 字节？	固定长度可以用 `index × 20` 直接算出偏移量，不需要遍历，O(1) 定位；这也是它能做「位点即下标」的基础
ConsumeQueue 丢了能恢复吗？	能。CommitLog 是唯一的真实数据源，ConsumeQueue 和 IndexFile 都是索引，可以通过重放 CommitLog 完整重建
IndexFile 是干什么的？	支持按 msgId / 消息 key 精确查询单条消息。结构是 Hash 槽 + 链表，这是 Kafka 完全不具备的能力
混合存储会不会导致一个 Topic 影响另一个？	会。所有 Topic 共用 CommitLog 和 PageCache，一个 Topic 的大消息或大量冷读会挤占 PageCache，影响其他 Topic。所以核心业务建议独立集群
RocketMQ 为什么用 mmap 而不是 sendfile？	mmap 让用户态能访问消息内容，这是实现 TAG/SQL92 过滤、消息查询、事务消息的前提；sendfile 虽然零拷贝更彻底但数据不经过用户态
mmap 的限制是什么？	单次映射有大小限制（这也是 CommitLog 定为 1GB 的原因之一），而且 `page fault` 会导致首次访问慢，所以 RocketMQ 有文件预热机制（写满一遍 0 并 `mlock`）

✅ 项目关联话术

我们的 RocketMQ 集群上有 200 多个 Topic、总共 3000 多个队列，这个规模下 Kafka 是跑不动的（文件句柄和随机 IO 都撑不住），这正是我们选 RocketMQ 的核心原因。但混合存储的「互相影响」我们也踩过：一个做文件导入的业务往消息里塞了 2MB 的 body，持续写入把 PageCache 挤爆，导致同一 Broker 上另外十几个 Topic 的消费延迟全部从毫秒级涨到秒级。之后我们定了两条规范：单条消息体不超过 4KB（大数据走 OSS 只传 URL）、核心交易 Topic 迁到独立 Broker 集群做物理隔离。

题 11：RabbitMQ、RocketMQ、Kafka 怎么选型？

精简口述答案（90 秒版）

我按三个维度判断：吞吐量级、功能需求、团队技术栈。

RabbitMQ：Erlang 写的，单机吞吐万级。优势是延迟最低（微秒级）、路由能力最灵活（四种 Exchange + 通配符）、管理界面好用、社区成熟。劣势是吞吐上限低、消息堆积能力弱（默认存内存，要用 lazy queue）、Erlang 栈二次开发基本不可能、集群扩展性差。适合数据量不大、路由逻辑复杂、对延迟敏感的中小型业务系统。

RocketMQ：Java 写的，单机吞吐十万级。优势是功能最全（事务消息、延时消息、顺序消息、死信队列、消息查询、Broker 端过滤）、Java 栈可读可改、阿里大规模验证过、海量 Topic 不衰减。劣势是生态不如 Kafka（大数据组件对接要自己写）、社区活跃度比 Kafka 低。适合电商、金融这类业务消息场景，尤其是需要事务消息和延时消息的。

Kafka：Scala/Java 写的，单机吞吐百万级，是三者里最高的。优势是吞吐极高、生态最完善（Flink/Spark/Hive 原生对接）、天然的流处理平台。劣势是分区数多了性能下降、不支持延时消息、消息查询能力缺失、运维（尤其是 ZK 时代）较复杂。适合日志采集、行为埋点、数据管道、流式计算。

我的结论是：业务消息选 RocketMQ，数据管道选 Kafka，小型系统且路由复杂选 RabbitMQ。实际上大公司往往是 Kafka + RocketMQ 两套并行，按场景分流，不是二选一。

连环追问链

追问	答题要点
Pulsar 值得上吗？	架构更先进（存算分离，Broker 无状态、BookKeeper 存储，扩容最优雅），但社区和运维经验不足、组件多（还要维护 ZK + BookKeeper）。新团队慎用
已有 RabbitMQ 要不要迁 RocketMQ？	看瓶颈。如果是吞吐和堆积能力不足、或者需要事务/延时消息，值得迁；如果只是路由需求，RabbitMQ 更合适，别为了技术而迁移
Kafka 现在还依赖 ZooKeeper 吗？	2.8 开始支持 KRaft 模式（自己用 Raft 管元数据），3.3 后 KRaft 生产可用，4.0 已彻底移除 ZK 依赖
为什么说 Kafka 分区多了会降性能？	文件数 = Topic × Partition，每个分区都在顺序写自己的文件，多个文件同时写在磁盘层面就是随机 IO；加上每个分区都有独立的 fetch 请求，网络和内存开销也线性增长
三者的消息可靠性配置对应关系？	RabbitMQ：持久化 + confirm + 手动 ack；RocketMQ：同步刷盘 + 同步复制 + 手动 ack；Kafka：`acks=all` + `min.insync.replicas>=2` + 手动 commit
选型时最容易被忽略的因素？	运维能力和故障处置经验。一个你们能救活的次优选择，远好过一个挂了没人会修的最优选择

✅ 项目关联话术

我们公司是 Kafka + RocketMQ 双栈。数据侧（用户行为埋点、业务库 binlog 同步数仓、日志采集）全走 Kafka，因为量在百万 TPS 级别、单条丢一点无所谓、而且下游是 Flink 和 Hive，生态对接零成本。业务侧（订单、支付、库存、营销）全走 RocketMQ，因为需要事务消息保证下单和库存的最终一致、需要延时消息做关单、需要按 msgId 查消息来排查客诉。还有个历史遗留的 RabbitMQ 集群跑着几个老的管理后台任务，量很小、路由规则很绕，评估过迁移成本后决定不动它——没有收益的迁移本身就是风险。

题 12：MQ 怎么实现分布式事务的最终一致性？

精简口述答案（90 秒版）

MQ 做分布式事务的核心思路是用「最终一致」替代「强一致」：不追求所有服务同时成功，而是保证「上游成功后，下游一定会在有限时间内达到一致状态」。

要解决两个原子性问题：

第一个，本地事务和发消息的原子性。有两种方案：

本地消息表——在业务库里建一张 t_message 表，业务数据和消息记录在同一个本地事务里 insert，事务提交后立刻尝试发送（发成功就更新状态为已发送），同时有定时任务扫描「超过 1 分钟还是待发送」的记录补发。优点是不依赖任何 MQ 特性、可观测性最好（直接查表就知道哪条消息没发出去）、可靠性最高。缺点是侵入业务库、要自己写定时任务。

RocketMQ 事务消息——半消息 + 二次确认 + 事务回查，前面第 5 题讲过。优点是代码简洁、不侵入业务库。缺点是只有 RocketMQ 支持、吞吐减半。

第二个，下游消费的可靠性。靠手动 ACK + 重试 + 死信 + 消费端幂等。这里幂等是必需的，因为重试一定会带来重复。

最后必须有对账兜底。无论方案多严密，都存在「回查超过 15 次被回滚」「消息进死信没人处理」这类边界情况。所以关键业务一定要有定时对账任务，扫出不一致的数据主动补偿。

和 Seata 这类框架对比：Seata AT 模式是「伪强一致」（全局锁 + 反向补偿），适合链路短、需要立刻看到一致结果的场景，但全局锁影响并发；MQ 最终一致性无锁、吞吐高、链路可以很长，适合「允许秒级不一致」的异步场景，这也是互联网业务的主流选择。

连环追问链

追问	答题要点
本地消息表的定时任务怎么避免重复发送？	消息表记录状态（待发送/已发送/已确认），定时任务只扫「待发送且创建超过 N 秒」的；即使重复发了，消费端幂等会兜住
本地消息表会不会越来越大？	会，必须有归档/清理任务。一般保留 7 天的已确认记录，之后归档到历史表或直接删除
下游一直失败怎么办？	重试耗尽进死信 + 告警 + 人工介入；同时上游的对账任务会发现不一致并走补偿或回滚流程
需要回滚上游怎么做？	MQ 最终一致性没有自动回滚，只能做业务补偿（比如创建一笔反向的退款单），这需要业务层面设计补偿接口
为什么不直接用 2PC/XA？	XA 要求所有参与方长时间持有资源锁，可用性和性能都极差，而且跨服务跨语言的 XA 几乎没法落地
TCC 和 MQ 最终一致怎么选？	TCC 需要业务实现 Try/Confirm/Cancel 三个接口，侵入大但能做到准实时一致；MQ 侵入小但只有最终一致。对实时性没要求就用 MQ

✅ 项目关联话术

我们「下单扣减优惠券」用的是本地消息表方案。下单时在订单库的事务里同时 insert 订单和一条 t_message 记录（内容是优惠券核销指令，状态=待发送）；事务提交后立刻 send，成功则 UPDATE status='SENT'。另有一个每 30 秒跑一次的补偿任务，扫「状态=待发送且创建超过 60 秒」的记录重新投递，连续失败 5 次就告警。优惠券服务消费时按 (couponId, orderNo) 做唯一索引幂等。为什么没用事务消息？因为本地消息表让我们能直接用 SQL 查到「哪些消息没投出去」，排查和手动补偿都非常直观，而事务消息的半消息在 Broker 内部，出问题时只能看 Broker 日志。这是可观测性上的主动选择。

✍️ 手撕代码【核心实现 + 边界 Case 自测清单 + 变种拓展题】

题 1：实现一个可靠的 RabbitMQ 消费者（手动 ACK + 幂等 + 死信）

核心实现

java

// ========== 1. 声明队列、交换机、死信 ==========
@Configuration
public class RabbitConfig {

    public static final String BIZ_EXCHANGE  = "order.exchange";
    public static final String BIZ_QUEUE     = "order.pay.queue";
    public static final String DLX_EXCHANGE  = "order.dlx.exchange";
    public static final String DLQ_QUEUE     = "order.pay.dlq";

    @Bean
    public TopicExchange bizExchange() {
        return ExchangeBuilder.topicExchange(BIZ_EXCHANGE).durable(true).build();
    }

    @Bean
    public Queue bizQueue() {
        return QueueBuilder.durable(BIZ_QUEUE)
                .deadLetterExchange(DLX_EXCHANGE)          // x-dead-letter-exchange
                .deadLetterRoutingKey("order.pay.dead")    // x-dead-letter-routing-key
                .lazy()                                    // 预计有堆积，必须惰性队列
                .build();
    }

    @Bean
    public Binding bizBinding() {
        return BindingBuilder.bind(bizQueue()).to(bizExchange()).with("order.pay.*");
    }

    @Bean
    public DirectExchange dlxExchange() {
        return ExchangeBuilder.directExchange(DLX_EXCHANGE).durable(true).build();
    }

    @Bean
    public Queue dlqQueue() {
        return QueueBuilder.durable(DLQ_QUEUE).build();   // 死信队列本身不再配 DLX
    }

    @Bean
    public Binding dlqBinding() {
        return BindingBuilder.bind(dlqQueue()).to(dlxExchange()).with("order.pay.dead");
    }
}

java

// ========== 2. 消费者：手动 ACK + 幂等 + 异常分类 ==========
@Slf4j
@Component
public class OrderPayConsumer {

    private static final int MAX_RETRY = 3;

    @Resource private IdempotentService idempotent;
    @Resource private OrderService orderService;
    @Resource private AlarmService alarm;

    @RabbitListener(queues = RabbitConfig.BIZ_QUEUE, ackMode = "MANUAL")
    public void onMessage(Message message, Channel channel) throws IOException {
        long tag = message.getMessageProperties().getDeliveryTag();
        String body = new String(message.getBody(), StandardCharsets.UTF_8);
        String traceId = header(message, "traceId");
        MDC.put("traceId", traceId != null ? traceId : UUID.randomUUID().toString());

        PayEvent event = null;
        try {
            // 2.1 反序列化失败属于数据异常，绝不能重试
            try {
                event = JSON.parseObject(body, PayEvent.class);
            } catch (Exception e) {
                log.error("消息格式错误，直接丢死信，body={}", body, e);
                channel.basicNack(tag, false, false);   // requeue=false → 进 DLX
                return;
            }

            // 2.2 幂等前置判断（PROCESSING / DONE 两态）
            String bizKey = "pay:" + event.getPayNo();
            if (!idempotent.tryLock(bizKey)) {
                log.info("重复消息，直接 ack，payNo={}", event.getPayNo());
                channel.basicAck(tag, false);
                return;
            }

            // 2.3 真正的业务处理（内部是带状态条件的 UPDATE，天然幂等）
            orderService.handlePaySuccess(event);

            // 2.4 先标记完成，再 ack
            idempotent.markDone(bizKey);
            channel.basicAck(tag, false);

        } catch (BizException e) {
            // 业务异常：重试无意义，落异常表后 ack
            log.warn("业务异常不重试，payNo={}, msg={}", safeNo(event), e.getMessage());
            alarm.record(body, e);
            channel.basicAck(tag, false);

        } catch (Exception e) {
            // 瞬时异常：自己控制重试次数，而不是依赖 MQ 无限重投
            int retried = retryCount(message);
            if (retried >= MAX_RETRY) {
                log.error("重试 {} 次仍失败，转死信，payNo={}", retried, safeNo(event), e);
                alarm.send("支付消息消费失败", body);
                channel.basicNack(tag, false, false);          // 进死信
            } else {
                log.warn("第 {} 次失败，延迟重投，payNo={}", retried + 1, safeNo(event), e);
                republishWithDelay(message, retried + 1);      // 自己重投，带退避
                channel.basicAck(tag, false);                  // 原消息 ack 掉
            }
        } finally {
            MDC.clear();   // 线程池复用，必须清理，否则 traceId 串号
        }
    }

    private int retryCount(Message m) {
        Object v = m.getMessageProperties().getHeaders().get("x-retry-count");
        return v == null ? 0 : Integer.parseInt(v.toString());
    }

    private String header(Message m, String key) {
        Object v = m.getMessageProperties().getHeaders().get(key);
        return v == null ? null : v.toString();
    }

    private String safeNo(PayEvent e) {
        return e == null ? "unknown" : e.getPayNo();
    }
}

java

// ========== 3. 幂等服务：PROCESSING / DONE 两态 ==========
@Service
public class IdempotentService {

    @Resource private StringRedisTemplate redis;

    /** 返回 true 表示拿到处理权；false 表示已完成，可直接 ack */
    public boolean tryLock(String bizKey) {
        String key = "mq:idem:" + bizKey;
        Boolean first = redis.opsForValue()
                .setIfAbsent(key, "PROCESSING", 60, TimeUnit.SECONDS);
        if (Boolean.TRUE.equals(first)) {
            return true;
        }
        String status = redis.opsForValue().get(key);
        if ("DONE".equals(status)) {
            return false;
        }
        // PROCESSING：别人在处理，或上次处理崩了还没过期 → 抛异常走重试，绝不能 ack
        throw new RetryLaterException("消息正在处理中：" + bizKey);
    }

    public void markDone(String bizKey) {
        redis.opsForValue().set("mq:idem:" + bizKey, "DONE", 7, TimeUnit.DAYS);
    }
}

边界 Case 自测清单

#	Case	期望行为
1	消息体是非法 JSON	不重试，直接 `basicNack(requeue=false)` 进死信并告警
2	同一条消息被重复投递 2 次	第二次命中 `DONE`，直接 ack，业务只执行一次
3	业务执行到一半进程被 kill	消息未 ack，Broker 重新投递；幂等 key 处于 `PROCESSING`，60 秒后过期可重入
4	下游数据库连接超时	识别为瞬时异常，按退避重投，第 4 次失败进死信
5	订单号在业务上不存在	识别为 `BizException`，不重试，落异常表后 ack
6	Redis 整体不可用	`tryLock` 抛异常 → 走瞬时异常重试分支；DB 唯一索引作为最终防线
7	`requeue` 误设为 `true`	会产生毫秒级死循环，必须保证所有 `nack` 都传 `requeue=false`
8	消费线程复用后 traceId 串号	`finally` 中 `MDC.clear()`，每条消息日志链路独立
9	队列堆积 500 万条	队列已声明 `lazy()`，消息落盘不占内存，不触发 RabbitMQ 内存告警
10	死信队列消费者自己也异常	死信消费者只做落库 + 告警，不调业务接口，保证不会二次失败

🚀 变种拓展题

把「自己重投带退避」改造成基于 TTL + DLX 的延迟重试队列（retry.5s / retry.30s / retry.5m 三级），要求不同重试次数进不同延迟队列。
给消费者加上消费耗时埋点和失败率监控，超过阈值自动触发熔断（暂停 SimpleMessageListenerContainer）。
实现死信消息的后台重投功能：从异常表取出原始消息，重新发到业务交换机，并记录重投次数与操作人。
把幂等从 Redis 改为数据库唯一索引 + 本地缓存两级，要求 Redis 完全不可用时功能不降级。
支持批量消费：改用 BatchMessageListener，一次处理 100 条并整批 ack，同时保证部分失败时能精确定位到失败条目。

题 2：实现一个本地消息表的最终一致性组件

核心实现

sql

-- ========== 1. 消息表 ==========
CREATE TABLE t_local_message (
    id            BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
    msg_key       VARCHAR(128) NOT NULL COMMENT '业务唯一键，用于幂等',
    topic         VARCHAR(64)  NOT NULL,
    tags          VARCHAR(64)  DEFAULT NULL,
    body          TEXT         NOT NULL,
    status        TINYINT      NOT NULL DEFAULT 0 COMMENT '0待发送 1已发送 2已确认 3已失败',
    retry_times   INT          NOT NULL DEFAULT 0,
    next_retry_at DATETIME     NOT NULL COMMENT '下次重试时间，用于退避',
    create_time   DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    update_time   DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    PRIMARY KEY (id),
    UNIQUE KEY uk_msg_key (msg_key),
    KEY idx_scan (status, next_retry_at)
) ENGINE=InnoDB COMMENT='本地消息表';

java

// ========== 2. 业务侧：业务数据与消息记录在同一个本地事务 ==========
@Service
public class OrderCreateService {

    @Resource private OrderMapper orderMapper;
    @Resource private LocalMessageService messageService;
    @Resource private TransactionTemplate txTemplate;

    public void createOrder(OrderDTO dto) {
        Long msgId = txTemplate.execute(status -> {
            // 2.1 业务入库
            Order order = orderMapper.insertOrder(dto);
            // 2.2 消息入库（同一个事务，要么都成功要么都回滚）
            CouponUseEvent event = CouponUseEvent.of(order);
            return messageService.save(
                    "coupon:use:" + order.getOrderNo(),   // msg_key 作为下游幂等依据
                    "COUPON_TOPIC", "USE", JSON.toJSONString(event));
        });

        // 2.3 事务提交后立刻尝试发送（失败也没关系，定时任务会补）
        messageService.sendAsync(msgId);
    }
}

java

// ========== 3. 消息服务：保存 + 发送 + 状态流转 ==========
@Slf4j
@Service
public class LocalMessageService {

    @Resource private LocalMessageMapper mapper;
    @Resource private RocketMQTemplate mqTemplate;

    /** 必须在外层事务内调用 */
    public Long save(String msgKey, String topic, String tags, String body) {
        LocalMessage m = new LocalMessage();
        m.setMsgKey(msgKey);
        m.setTopic(topic);
        m.setTags(tags);
        m.setBody(body);
        m.setStatus(0);
        m.setNextRetryAt(new Date());
        mapper.insert(m);
        return m.getId();
    }

    @Async("mqSendExecutor")
    public void sendAsync(Long id) {
        try {
            doSend(mapper.selectById(id));
        } catch (Exception e) {
            log.warn("首次发送失败，交由定时任务补偿，id={}", id, e);
        }
    }

    public void doSend(LocalMessage m) {
        if (m == null || m.getStatus() != 0) {
            return;   // 已发送或已确认，跳过
        }
        Message<String> msg = MessageBuilder.withPayload(m.getBody())
                .setHeader(RocketMQHeaders.KEYS, m.getMsgKey())   // 便于按 key 查消息
                .build();
        SendResult result = mqTemplate.syncSend(
                m.getTopic() + ":" + m.getTags(), msg, 3000);

        if (SendStatus.SEND_OK == result.getSendStatus()) {
            // CAS 更新：只有 status 还是 0 才更新，防止并发重复流转
            mapper.updateStatusCas(m.getId(), 0, 1);
        } else {
            throw new MqSendException("发送结果异常：" + result.getSendStatus());
        }
    }

    /** 下游回执后调用，置为已确认 */
    public void confirm(String msgKey) {
        mapper.updateStatusByKey(msgKey, 2);
    }
}

java

// ========== 4. 定时补偿任务：扫描未发送的消息重投 ==========
@Slf4j
@Component
public class LocalMessageCompensateJob {

    private static final int BATCH = 200;
    private static final int MAX_RETRY_TIMES = 5;
    /** 指数退避：10s / 30s / 2m / 10m / 30m */
    private static final int[] BACKOFF_SECONDS = {10, 30, 120, 600, 1800};

    @Resource private LocalMessageMapper mapper;
    @Resource private LocalMessageService messageService;
    @Resource private RedissonClient redisson;
    @Resource private AlarmService alarm;

    @Scheduled(fixedDelay = 30_000)
    public void run() {
        RLock lock = redisson.getLock("job:local-message-compensate");
        // 多实例部署必须加分布式锁，否则同一条消息会被多个实例同时重投
        if (!lock.tryLock()) {
            return;
        }
        try {
            List<LocalMessage> list = mapper.selectPending(BATCH, new Date());
            for (LocalMessage m : list) {
                try {
                    messageService.doSend(m);
                } catch (Exception e) {
                    int times = m.getRetryTimes() + 1;
                    if (times >= MAX_RETRY_TIMES) {
                        mapper.updateStatusCas(m.getId(), 0, 3);   // 标记失败
                        alarm.send("本地消息投递彻底失败", m.getMsgKey());
                    } else {
                        Date next = DateUtils.addSeconds(new Date(),
                                BACKOFF_SECONDS[Math.min(times - 1, BACKOFF_SECONDS.length - 1)]);
                        mapper.updateRetry(m.getId(), times, next);
                    }
                }
            }
        } finally {
            lock.unlock();
        }
    }
}

sql

-- ========== 5. 扫描 SQL：必须走 idx_scan 索引 ==========
SELECT id, msg_key, topic, tags, body, status, retry_times
FROM t_local_message
WHERE status = 0 AND next_retry_at <= #{now}
ORDER BY id
LIMIT #{batch};

-- CAS 更新，防止并发重复流转
UPDATE t_local_message SET status = #{to}
WHERE id = #{id} AND status = #{from};

边界 Case 自测清单

#	Case	期望行为
1	业务入库成功但发送 MQ 失败	消息记录状态仍为 0，定时任务 30 秒内补发
2	业务事务回滚	消息记录随事务一起回滚，不会产生幽灵消息
3	发送成功但更新状态前进程崩了	状态仍为 0，定时任务会重发 → 下游收到重复消息，靠消费端幂等兜住
4	同一业务重复调用 `createOrder`	`uk_msg_key` 唯一索引拦截，第二次抛 `DuplicateKeyException`
5	多实例同时跑补偿任务	分布式锁保证只有一个实例扫描，避免重复投递
6	下游持续不可用	5 次指数退避后置为 status=3 并告警，不再无限重试占用资源
7	消息表数据量涨到千万	`idx_scan(status, next_retry_at)` 保证扫描只命中待发送的少量行
8	已确认的消息长期堆积	需配合归档任务，保留 7 天后清理 status=2 的记录
9	刚 insert 的消息被定时任务抢先发送	`doSend` 内部有 `status != 0` 判断 + CAS 更新，重复发送最多一次
10	MQ 整体宕机 1 小时	所有消息滞留在表中，MQ 恢复后由定时任务逐批补发，业务不受影响

🚀 变种拓展题

增加下游回执确认链路：下游消费成功后发一条回执消息，上游消费回执把状态从 1 改成 2，并对「已发送超过 10 分钟仍未确认」的消息告警。
把消息表从业务库拆到独立的消息库，此时业务和消息就不在同一个事务里了，设计一个仍能保证原子性的方案（提示：事务消息或 CDC）。
基于 MySQL binlog + Canal 实现无侵入版本：业务只写业务表，由 Canal 监听 binlog 产生 MQ 消息，对比两种方案的优劣。
实现消息表的分表（按 create_time 按月分表）和自动归档，保证扫描 SQL 不受历史数据影响。
给补偿任务加上消息优先级：资金类消息优先扫描、退避间隔更短，普通消息次之。

题 3：实现 RocketMQ 事务消息的完整收发链路

核心实现

java

// ========== 1. 事务生产者：本地事务 + 回查 ==========
@Slf4j
@Component
public class OrderTransactionProducer {

    @Resource private RocketMQTemplate mqTemplate;
    @Resource private OrderMapper orderMapper;

    public void createOrderWithMq(OrderDTO dto) {
        String orderNo = IdGenerator.nextOrderNo();
        dto.setOrderNo(orderNo);

        StockLockEvent event = StockLockEvent.of(dto);
        Message<String> msg = MessageBuilder.withPayload(JSON.toJSONString(event))
                .setHeader(RocketMQHeaders.KEYS, orderNo)      // 回查时用它查业务状态
                .setHeader("traceId", MDC.get("traceId"))
                .build();

        // 发送半消息，arg 作为 executeLocalTransaction 的入参
        TransactionSendResult result = mqTemplate.sendMessageInTransaction(
                "STOCK_TOPIC:LOCK", msg, dto);

        if (result.getLocalTransactionState() != LocalTransactionState.COMMIT_MESSAGE) {
            throw new BizException("下单失败，orderNo=" + orderNo);
        }
    }
}

java

// ========== 2. 事务监听器：两个方法都必须实现 ==========
@Slf4j
@RocketMQTransactionListener
public class StockTransactionListener implements RocketMQLocalTransactionListener {

    @Resource private OrderMapper orderMapper;

    /** 阶段一：执行本地事务 */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        OrderDTO dto = (OrderDTO) arg;
        try {
            orderMapper.insertOrder(dto);
            log.info("本地事务成功，提交消息，orderNo={}", dto.getOrderNo());
            return RocketMQLocalTransactionState.COMMIT;
        } catch (DuplicateKeyException e) {
            // 订单已存在 → 说明是重复请求，业务上视为成功，消息也该提交
            log.warn("订单已存在，视为成功，orderNo={}", dto.getOrderNo());
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("本地事务失败，回滚消息，orderNo={}", dto.getOrderNo(), e);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 阶段二：事务回查（兜底，必须可靠）
     * 关键：只依赖数据库状态，绝不依赖任何内存变量，
     * 因为回查时生产者可能已经是重启后的新实例了。
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        String orderNo = (String) msg.getHeaders().get(RocketMQHeaders.KEYS);
        if (orderNo == null) {
            orderNo = (String) msg.getHeaders().get("rocketmq_KEYS");
        }
        Order order = orderMapper.selectByOrderNo(orderNo);

        if (order == null) {
            // 查不到 → 本地事务没成功，回滚
            log.info("回查：订单不存在，回滚，orderNo={}", orderNo);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        if (order.getStatus() == OrderStatus.CREATING.code()) {
            // 中间态 → 还不确定，返回 UNKNOWN 等下次回查
            log.info("回查：订单处于创建中，等待下次回查，orderNo={}", orderNo);
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        log.info("回查：订单已存在，提交，orderNo={}", orderNo);
        return RocketMQLocalTransactionState.COMMIT;
    }
}

java

// ========== 3. 消费端：幂等 + 异常分类 + 自控重试 ==========
@Slf4j
@Component
@RocketMQMessageListener(
        topic = "STOCK_TOPIC",
        selectorExpression = "LOCK",
        consumerGroup = "stock_lock_group",
        consumeThreadMax = 32,
        maxReconsumeTimes = 3)            // 自己控制重试，不要用默认 16 次
public class StockLockConsumer implements RocketMQListener<MessageExt> {

    @Resource private StockService stockService;
    @Resource private AlarmService alarm;

    @Override
    public void onMessage(MessageExt msg) {
        String traceId = msg.getUserProperty("traceId");
        MDC.put("traceId", traceId != null ? traceId : UUID.randomUUID().toString());
        try {
            StockLockEvent event = JSON.parseObject(
                    new String(msg.getBody(), StandardCharsets.UTF_8), StockLockEvent.class);

            // 幂等：唯一索引 (order_no, sku_id) 兜底，重复直接返回
            int affected = stockService.lockStockIdempotent(event);
            log.info("库存预占完成，orderNo={}, affected={}", event.getOrderNo(), affected);

        } catch (JSONException e) {
            // 数据异常：抛出去也没用，记录后直接返回（等价于 ack）
            log.error("消息格式错误，msgId={}", msg.getMsgId(), e);
            alarm.send("库存消息格式错误", msg.getMsgId());

        } catch (BizException e) {
            // 业务异常：重试无意义
            log.warn("业务异常不重试，msgId={}, reason={}", msg.getMsgId(), e.getMessage());
            alarm.record(msg, e);

        } catch (Exception e) {
            if (msg.getReconsumeTimes() >= 3) {
                log.error("重试 3 次仍失败，告警转人工，msgId={}", msg.getMsgId(), e);
                alarm.send("库存预占失败", msg.getMsgId());
                return;   // 不再抛异常 → 不进死信，由人工处理
            }
            throw new RuntimeException(e);   // 抛异常 = RECONSUME_LATER
        } finally {
            MDC.clear();
        }
    }
}

sql

-- ========== 4. 幂等的库存扣减 SQL ==========
-- 预占记录表加唯一索引，重复消息会被 DB 拦住
CREATE TABLE t_stock_lock (
    id       BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
    order_no VARCHAR(64) NOT NULL,
    sku_id   BIGINT      NOT NULL,
    num      INT         NOT NULL,
    PRIMARY KEY (id),
    UNIQUE KEY uk_order_sku (order_no, sku_id)
) ENGINE=InnoDB;

-- 扣减库存：带条件防超卖，影响行数 0 则抛业务异常
UPDATE t_stock SET locked = locked + #{num}
WHERE sku_id = #{skuId} AND total - locked >= #{num};

边界 Case 自测清单

#	Case	期望行为
1	半消息发送成功，本地事务失败	`executeLocalTransaction` 返回 ROLLBACK，半消息被丢弃，消费者收不到
2	本地事务成功，二次确认网络中断	Broker 60 秒后触发回查，查到订单存在 → COMMIT，消息正常投递
3	本地事务执行中进程被 kill	回查时订单不存在（事务已回滚）→ ROLLBACK，无幽灵消息
4	回查时订单处于「创建中」中间态	返回 UNKNOWN，等下次回查，不做错误判断
5	回查 15 次仍返回 UNKNOWN	Broker 直接 ROLLBACK 丢弃，必须有对账任务兜住这种情况
6	生产者 Pod 重建后收到回查	回查只查数据库，不依赖内存，新实例也能正确应答
7	同一订单号重复下单	本地事务抛 `DuplicateKeyException`，按业务语义返回 COMMIT
8	消费端收到重复消息	`uk_order_sku` 唯一索引拦截，不会重复预占库存
9	库存不足	`UPDATE` 影响行数 0 → 抛 `BizException`，不重试，落异常表并告警
10	消费端下游超时	识别为瞬时异常，抛异常触发 `RECONSUME_LATER`，3 次后告警转人工

🚀 变种拓展题

补齐对账任务：扫描「订单已创建超过 10 分钟但没有对应库存预占记录」的数据，主动补发消息或回滚订单。
把事务消息方案改写成本地消息表方案，对比两者在代码量、可观测性、故障排查上的差异。
实现一个事务状态表（t_tx_record），让 checkLocalTransaction 不依赖业务表而是查这张通用表，使组件可以复用到任意业务。
给事务消息加上回查次数监控：回查次数超过 3 次的消息要告警，因为这通常意味着本地事务耗时异常或代码有 bug。
压测对比事务消息 vs 普通消息的吞吐差异，并分析半消息恢复过程对 CommitLog 写入量的放大效应。

题 4：实现延时关单（RocketMQ 延时等级版 + RabbitMQ TTL+DLX 版）

核心实现

java

// ========== 方案 A：RocketMQ 延时消息 ==========
@Slf4j
@Component
public class OrderCloseProducer {

    /** RocketMQ 4.x 的 18 个延时等级 */
    private static final int LEVEL_30_MIN = 16;   // 1s 5s 10s 30s 1m...10m 20m 30m 1h 2h

    @Resource private RocketMQTemplate mqTemplate;

    public void sendCloseMessage(String orderNo) {
        Message<String> msg = MessageBuilder
                .withPayload(JSON.toJSONString(new CloseEvent(orderNo)))
                .setHeader(RocketMQHeaders.KEYS, "close:" + orderNo)
                .build();

        // 4.x 只能用等级：syncSend(dest, msg, timeout, delayLevel)
        // ⚠️ 等级是固定档位，无法在发送侧做随机抖动。同一秒下单的订单
        // 会在同一秒集中到期，大促要削这个毛刺只有两条路：
        //   1) 升到 5.x 用 setDeliveryTimestamp（见下面的 V5 方法）；
        //   2) 消费端拿到消息后先 Thread.sleep(random) 再处理（简陋且占线程，不推荐）。
        SendResult result = mqTemplate.syncSend("ORDER_CLOSE_TOPIC", msg, 3000, LEVEL_30_MIN);
        log.info("关单延时消息已发送，orderNo={}, status={}", orderNo, result.getSendStatus());
    }
}

// ========== 方案 A+：RocketMQ 5.x 任意秒级精度，可做到期抖动，推荐 ==========
@Slf4j
@Component
public class OrderCloseProducerV5 {

    private static final long BASE_DELAY_MILLIS = 30 * 60 * 1000L;

    private ClientServiceProvider provider;
    private Producer producer;

    @PostConstruct
    public void init() throws ClientException {
        provider = ClientServiceProvider.loadService();
        ClientConfiguration config = ClientConfiguration.newBuilder()
                .setEndpoints("127.0.0.1:8081")       // 5.x 连 Proxy，不是 NameServer
                .build();
        producer = provider.newProducerBuilder()
                .setClientConfiguration(config)
                .setTopics("ORDER_CLOSE_TOPIC")
                .build();
    }

    public void sendCloseMessage(String orderNo) throws ClientException {
        // 真正有效的抖动：直接作用在到期时间戳上，把同一秒的到期量打散到 60 秒内
        long jitterMillis = ThreadLocalRandom.current().nextInt(60) * 1000L;

        org.apache.rocketmq.client.apis.message.Message msg = provider.newMessageBuilder()
                .setTopic("ORDER_CLOSE_TOPIC")
                .setKeys("close:" + orderNo)
                .setBody(JSON.toJSONString(new CloseEvent(orderNo))
                        .getBytes(StandardCharsets.UTF_8))   // 与 4.x 保持同一报文格式，消费端可复用
                .setDeliveryTimestamp(System.currentTimeMillis() + BASE_DELAY_MILLIS + jitterMillis)
                .build();
        producer.send(msg);
    }

    @PreDestroy
    public void destroy() throws IOException {
        // 5.x 的 Producer 是 Closeable，必须显式关闭，否则后台线程和长连接泄漏
        if (producer != null) {
            producer.close();
        }
    }
}

java

// ========== 关单消费者：回查状态 + 状态机互斥 ==========
@Slf4j
@Component
@RocketMQMessageListener(
        topic = "ORDER_CLOSE_TOPIC",
        consumerGroup = "order_close_group",
        maxReconsumeTimes = 3)
public class OrderCloseConsumer implements RocketMQListener<MessageExt> {

    @Resource private OrderMapper orderMapper;
    @Resource private StockService stockService;

    @Override
    public void onMessage(MessageExt msg) {
        CloseEvent event = JSON.parseObject(
                new String(msg.getBody(), StandardCharsets.UTF_8), CloseEvent.class);
        String orderNo = event.getOrderNo();

        // 关键一：延时消息只是「触发器」，必须回查订单当前状态
        Order order = orderMapper.selectByOrderNo(orderNo);
        if (order == null) {
            log.warn("订单不存在，直接忽略，orderNo={}", orderNo);
            return;
        }
        if (order.getStatus() != OrderStatus.UNPAID.code()) {
            log.info("订单状态已变更，无需关单，orderNo={}, status={}", orderNo, order.getStatus());
            return;
        }

        // 关键二：关单 SQL 必须带状态条件，靠 DB 行锁与支付流程互斥
        int affected = orderMapper.closeIfUnpaid(orderNo);
        if (affected == 0) {
            log.info("关单时被支付抢先，放弃关单，orderNo={}", orderNo);
            return;   // 幂等：影响行数 0 说明已被处理，直接 ack
        }

        // 关单成功才释放库存（同样要幂等）
        stockService.releaseStock(orderNo);
        log.info("关单成功，orderNo={}", orderNo);
    }
}

sql

-- 关单与支付的互斥完全靠这条 SQL 的状态条件
UPDATE t_order SET status = 5, close_time = NOW()
WHERE order_no = #{orderNo} AND status = 1;

java

// ========== 方案 B：RabbitMQ 队列级 TTL + DLX ==========
@Configuration
public class DelayCloseConfig {

    /** 延时队列：没有消费者，消息只会因 TTL 过期而进 DLX */
    @Bean
    public Queue delayQueue30Min() {
        return QueueBuilder.durable("order.close.delay.30m")
                .ttl(30 * 60 * 1000)                               // 队列级 TTL，不会队头阻塞
                .deadLetterExchange("order.close.dlx")
                .deadLetterRoutingKey("order.close.real")
                .lazy()
                .build();
    }

    /** 真实队列：挂关单消费者 */
    @Bean
    public Queue realCloseQueue() {
        return QueueBuilder.durable("order.close.real.queue").build();
    }

    @Bean
    public DirectExchange closeDlx() {
        return ExchangeBuilder.directExchange("order.close.dlx").durable(true).build();
    }

    @Bean
    public Binding realBinding() {
        return BindingBuilder.bind(realCloseQueue()).to(closeDlx()).with("order.close.real");
    }
}

text

RabbitMQ 延时关单的消息流转路径：

  生产者
    │ publish(routingKey = order.close.delay)
    ▼
  order.close.delay.exchange
    │
    ▼
  order.close.delay.30m 队列（无消费者，队列级 TTL = 30min）
    │  30 分钟后消息过期
    ▼
  order.close.dlx（死信交换机）
    │ routingKey = order.close.real
    ▼
  order.close.real.queue → 关单消费者消费

⚠️ 必须用「队列级 TTL」而不是「消息级 TTL」：
   队列级 → 所有消息 TTL 相同，到期顺序 = 入队顺序，不会队头阻塞
   消息级 → RabbitMQ 只检查队头，队头 30 分钟的消息会卡住后面 1 分钟的消息

java

// ========== 兜底扫表任务：MQ 是主路径，扫表是保险 ==========
@Component
public class OrderCloseCompensateJob {

    @Resource private OrderMapper orderMapper;
    @Resource private OrderCloseService closeService;

    @Scheduled(cron = "0 5 * * * ?")   // 每小时的第 5 分钟
    public void run() {
        // 扫「创建超过 35 分钟（比 30 分钟多留 5 分钟缓冲）仍未支付」的订单
        Date deadline = DateUtils.addMinutes(new Date(), -35);
        List<String> orderNos = orderMapper.selectTimeoutUnpaid(deadline, 500);
        for (String orderNo : orderNos) {
            try {
                closeService.close(orderNo);   // 内部复用同一套状态条件 SQL，天然幂等
            } catch (Exception e) {
                log.error("兜底关单失败，orderNo={}", orderNo, e);
            }
        }
    }
}

边界 Case 自测清单

#	Case	期望行为
1	用户在第 29 分钟完成支付	关单消息到期时回查状态已是「已支付」，直接 ack 不关单
2	支付和关单在同一毫秒并发	关单 SQL 带 `status = 1` 条件，靠 DB 行锁，有且只有一个成功
3	关单消息丢失	兜底扫表任务在下一个整点 5 分补偿关单
4	关单消息重复投递	第二次 `UPDATE` 影响行数 0，直接返回，库存不会被重复释放
5	订单已被用户主动取消	状态不是「待支付」，回查后直接忽略
6	大促同一秒下单 5 万笔	5.x 用 `setDeliveryTimestamp` 加 0~60 秒随机抖动把到期投递打散；4.x 等级固定无法抖动，只能靠扩消费者扛
7	RabbitMQ 用了消息级 TTL	会出现队头阻塞（TTL 小的消息被 TTL 大的卡住），必须改队列级 TTL
8	关单后释放库存失败	抛异常触发重试；`releaseStock` 必须幂等（按 orderNo 唯一索引判断是否已释放）
9	需要支持 15m/1h/24h 多档位	RocketMQ 用不同 `delayLevel`；RabbitMQ 要每个档位建一个独立队列
10	延时超过 RocketMQ 最大 2 小时	4.x 做「二次延时」（先延 2h 再重投一条延时消息）；5.x 直接 `setDeliveryTimestamp`

🚀 变种拓展题

实现 RocketMQ 4.x 的二次延时组件：支持任意时长，内部自动按 2 小时分段重投，并保证分段过程中消息不丢。
用 Redis ZSet 实现一个延时队列（score = 到期时间戳），要求支持多实例部署、避免重复扫描、支持持久化恢复。
实现基于 Netty HashedWheelTimer 的本地时间轮延时方案，并分析它为什么不能单独用于生产（提示：单机内存、无持久化）。
设计一个延时任务统一平台：对外提供「在某时刻回调某 HTTP 接口」的能力，内部用 MQ 延时消息实现，支持取消和改期。
对比「延时消息」和「定时扫表」在订单量 100 万/天和 1 亿/天两种规模下的成本与可靠性，给出选型建议。

题 5：实现消息堆积的应急处理程序（搬运工扩容 + 批量消费）

核心实现

java

// ========== 1. 搬运工：只转发不处理，把消息挪到队列数更多的新 Topic ==========
@Slf4j
@Component
@RocketMQMessageListener(
        topic = "ORDER_SYNC_TOPIC",              // 原 Topic，只有 8 个队列
        consumerGroup = "order_sync_transfer_group",
        consumeThreadMax = 64,                   // 搬运不耗 CPU，线程可以开大
        consumeMessageBatchMaxSize = 32)
public class BacklogTransferConsumer implements RocketMQListener<List<MessageExt>> {

    /** 应急开关，默认关闭，只在堆积时打开 */
    @Value("${mq.transfer.enabled:false}")
    private boolean enabled;

    @Resource private RocketMQTemplate mqTemplate;

    @Override
    public void onMessage(List<MessageExt> msgs) {
        if (!enabled) {
            throw new IllegalStateException("搬运工未启用，消息退回重试");
        }
        List<Message<byte[]>> batch = new ArrayList<>(msgs.size());
        for (MessageExt m : msgs) {
            batch.add(MessageBuilder.withPayload(m.getBody())
                    .setHeader(RocketMQHeaders.KEYS, m.getKeys())
                    // 保留原始 msgId，便于全链路追溯
                    .setHeader("originMsgId", m.getMsgId())
                    .setHeader("traceId", m.getUserProperty("traceId"))
                    .build());
        }
        // 转发到 ORDER_SYNC_TOPIC_EXPAND（64 个队列），由大量消费者并行消费
        mqTemplate.syncSend("ORDER_SYNC_TOPIC_EXPAND", batch, 3000);
        log.info("搬运 {} 条消息到扩容 Topic", batch.size());
    }
}

java

// ========== 2. 扩容 Topic 的批量消费者：攒批写库，把 RT 摊薄 ==========
@Slf4j
@Component
@RocketMQMessageListener(
        topic = "ORDER_SYNC_TOPIC_EXPAND",
        consumerGroup = "order_sync_expand_group",
        consumeThreadMax = 32,
        consumeMessageBatchMaxSize = 100)        // 一次回调给 100 条
public class BatchSyncConsumer implements RocketMQListener<List<MessageExt>> {

    @Resource private OrderSyncMapper syncMapper;
    @Resource private DeadLetterRecorder deadLetterRecorder;   // 坏消息落库，不走 MQ 重试
    @Resource private RateLimiter rateLimiter;   // Guava RateLimiter，保护下游

    @Override
    public void onMessage(List<MessageExt> msgs) {
        // 2.1 解析 + 按业务主键去重（批内去重，减少无效写）
        Map<String, OrderSyncDO> distinct = new LinkedHashMap<>();
        List<MessageExt> badMsgs = new ArrayList<>();
        for (MessageExt m : msgs) {
            try {
                OrderSyncDO dto = JSON.parseObject(
                        new String(m.getBody(), StandardCharsets.UTF_8), OrderSyncDO.class);
                // 同一订单多次变更，批内只保留最后一条（靠 version 判断）
                distinct.merge(dto.getOrderNo(), dto,
                        (old, cur) -> cur.getVersion() > old.getVersion() ? cur : old);
            } catch (Exception e) {
                badMsgs.add(m);   // 坏消息单独收集，不能让它拖垮整批
            }
        }
        if (!badMsgs.isEmpty()) {
            log.error("本批有 {} 条消息格式错误，已记录", badMsgs.size());
            badMsgs.forEach(m -> deadLetterRecorder.save(m));
        }
        if (distinct.isEmpty()) {
            return;
        }

        // 2.2 下游限流：扩容消费者时一定要保护下游，否则把 DB 压死
        rateLimiter.acquire(distinct.size());

        // 2.3 批量 upsert：1 次网络往返代替 100 次
        try {
            syncMapper.batchUpsert(new ArrayList<>(distinct.values()));
        } catch (Exception e) {
            // 批量失败时降级为单条处理，精确定位失败条目
            log.warn("批量写入失败，降级单条处理，size={}", distinct.size(), e);
            fallbackOneByOne(distinct.values());
        }
    }

    private void fallbackOneByOne(Collection<OrderSyncDO> list) {
        List<String> failed = new ArrayList<>();
        for (OrderSyncDO dto : list) {
            try {
                syncMapper.upsert(dto);
            } catch (Exception e) {
                failed.add(dto.getOrderNo());
            }
        }
        if (!failed.isEmpty()) {
            // 有失败就抛异常让整批重试，重复的部分靠 upsert 幂等兜住
            throw new RuntimeException("单条降级仍有失败：" + failed);
        }
    }
}

sql

-- ========== 3. 幂等的批量 upsert：靠唯一索引 + 版本号 ==========
INSERT INTO t_order_sync (order_no, status, amount, version, update_time)
VALUES
    (#{orderNo1}, #{status1}, #{amount1}, #{version1}, NOW()),
    (#{orderNo2}, #{status2}, #{amount2}, #{version2}, NOW())
    -- 实际用 MyBatis 的 foreach 标签拼接，单批建议 100~500 条
ON DUPLICATE KEY UPDATE
    status      = IF(VALUES(version) > version, VALUES(status), status),
    amount      = IF(VALUES(version) > version, VALUES(amount), amount),
    update_time = IF(VALUES(version) > version, NOW(), update_time),
    version     = IF(VALUES(version) > version, VALUES(version), version);
-- 关键：IF(VALUES(version) > version, ...) 让乱序到达的旧消息自动失效

bash

# ========== 4. 应急处置命令清单 ==========
# 4.1 确认堆积分布：是全部队列还是个别队列
sh mqadmin consumerProgress -n ${NS} -g order_sync_group

# 4.2 查看 Topic 的队列数（决定了并发上限）
sh mqadmin topicStatus -n ${NS} -t ORDER_SYNC_TOPIC

# 4.3 扩容队列数（注意：顺序消息不能在存量消费完前扩容）
sh mqadmin updateTopic -n ${NS} -c DefaultCluster -t ORDER_SYNC_TOPIC -r 32 -w 32

# 4.4 核武器：跳过堆积（仅限非核心业务，必须评审！）
sh mqadmin resetOffsetByTime -n ${NS} -g order_sync_group -t ORDER_SYNC_TOPIC -s now

# 4.5 回溯重放（修完 bug 后重新消费，前提是消费端幂等）
sh mqadmin resetOffsetByTime -n ${NS} -g order_sync_group -t ORDER_SYNC_TOPIC -s 1698400000000

text

堆积应急决策树：

  发现堆积告警
    │
    ├─ 部分队列堆积？
    │    ├─ 某消费者实例挂了/卡死 → 重启实例，堆积自然追平
    │    └─ 数据倾斜（某 key 消息量巨大）→ 改哈希策略或单独拆 Topic
    │
    └─ 全部队列堆积？
         ├─ 消费者实例数 < 队列数 → 【最优】水平扩容实例到队列数
         ├─ 消费者实例数 = 队列数 → 队列数是瓶颈
         │     ├─ 非顺序消息 → 【搬运工方案】转发到队列数更多的新 Topic
         │     └─ 顺序消息 → 只能优化单条耗时，不能搬运（会乱序）
         ├─ 下游是瓶颈（DB/RPC 慢） → 【改批量】攒批写入 + 限流保护下游
         └─ 消息已无业务价值 → 【评审后】resetOffsetByTime -s now 跳过

边界 Case 自测清单

#	Case	期望行为
1	搬运工开关默认关闭	抛异常让消息退回重试，不会在非应急时误消费原 Topic
2	搬运工转发失败	抛异常 → `RECONSUME_LATER`，原消息不丢，稍后重试
3	搬运过程中消息顺序被打乱	符合预期，搬运工方案不适用于顺序消息，必须提前确认
4	批量中有 1 条消息格式错误	单独收集到异常表，不影响其余 99 条正常处理
5	批量写库整批失败	降级为单条处理，精确定位失败条目后抛异常让整批重试
6	整批重试导致部分消息重复处理	`ON DUPLICATE KEY UPDATE` + 版本号比较保证幂等，重复无副作用
7	同一订单在一批内出现 3 次	批内按 `version` 合并，只保留最新一条，减少无效写入
8	乱序到达的旧版本消息	`IF(VALUES(version) > version, ...)` 使旧数据不覆盖新数据
9	扩容消费者后把数据库压死	`RateLimiter` 限流兜住，宁可慢点追也不能打挂下游
10	扩容队列数后顺序被破坏	顺序 Topic 必须等存量消息消费完才能扩容，否则同 key 消息会分散到不同队列

🚀 变种拓展题

给搬运工加上动态并发调节：根据实时堆积量自动调整 consumeThreadMax 和限流速率，堆积消除后自动降回。
实现消费延迟时间（而非条数）的监控指标：取队列最后一条已消费消息的存储时间与当前时间差，并接入告警。
设计一个堆积自愈方案：堆积超过阈值自动触发 K8s HPA 扩容消费者 Pod，追平后自动缩容，并说明有哪些风险。
实现带消费端反压的消费者：下游 RT 升高时自动降低拉取速率（减小 pullBatchSize），RT 恢复后再提速。
把批量消费改造成支持部分成功语义：成功的条目直接确认，失败的条目单独重投到重试 Topic，避免整批重试的放大效应。

⚠️ 当日踩坑实录

❌ 坑	✅ 解决方案
用 `msgId` 做幂等 key，结果重复消息还是进来了	生产端重试会生成新的 msgId，必须用业务唯一键（订单号、支付流水号，或「业务号 + 操作类型」组合）
Redis 只标记「已处理」，业务失败后重试被误判为重复，数据永久丢失	必须区分 `PROCESSING`（短过期 60s）和 `DONE`（7 天）两个状态，处理中要抛异常走重试而不是直接 ack
幂等 key 设了 1 小时过期，最后几次重试失去幂等保护	幂等 key 过期时间必须大于消息最大重试周期，RocketMQ 16 次重试跨度 2 小时多，建议至少 24 小时
RabbitMQ `default-requeue-rejected` 默认 `true`，消息 nack 后毫秒级死循环打满 CPU	必须设成 `false` 并配置死信队列，所有 `basicNack` 都传 `requeue=false`
RabbitMQ 没设 `prefetch`，一个消费者把全部消息拉到本地，其他消费者空转	必须显式设置 `prefetch`（一般 10~50），否则是无限预取
RabbitMQ 队列没设 `durable`，重启后队列和消息全没了	持久化三要素必须全开：交换机 durable + 队列 durable + 消息 `deliveryMode=2`
只开了消息持久化但队列没持久化，重启后消息照样丢	队列没了消息自然无处可依，三要素缺一不可
RabbitMQ 堆积 200 万条后内存告警，所有生产者被阻塞	预计有堆积的队列必须声明 lazy queue（`x-queue-mode=lazy`），消息直接落盘
RabbitMQ 磁盘剩余空间低于 50MB，生产者被静默阻塞	调大 `disk_free_limit`（建议 `{mem_relative, 2.0}`）并监控磁盘水位
routingKey 写错，消息被静默丢弃，查了半天找不到消息	设置 `mandatory=true` + 实现 `ReturnCallback`，或配置 `alternate-exchange` 备份交换机
用消息级 TTL 做延时，TTL 1 分钟的消息被队头 30 分钟的消息卡住	RabbitMQ 只检查队头是否过期，必须用队列级 TTL，每个延时档位一个独立队列
用 RabbitMQ 延时插件，节点重启后延时消息全丢	延时插件的消息存在节点内存里且不支持镜像队列，可靠性要求高的场景不能用
RabbitMQ 镜像队列设成 `ha-mode=all`，队列一多集群就被同步流量打死	改成 `ha-mode=exactly, ha-params=2`（或 3），只做 2~3 副本
镜像队列在网络分区后出现数据丢失	镜像队列的 GM 协议只保证最终一致，新版本应该直接用 Quorum Queue（Raft，强一致）
RocketMQ 消费者实例数加到 20 个，但 Topic 只有 8 个队列，12 个实例完全空转	队列数才是并发上限，扩容前先 `updateTopic` 加队列数，或用搬运工方案
RocketMQ 让消息跑满默认 16 次重试，跨度 2小时产生 800 万次无效消费把 Broker 打爆	用 `maxReconsumeTimes` 自己控制重试次数（3~5 次），超过主动落异常表 + 告警
无脑 `catch (Exception e) { return RECONSUME_LATER; }`，反序列化失败的消息被无限重试	必须按异常类型分支：瞬时异常重试、业务异常落表 ack、数据异常直接进死信
RocketMQ 死信 Topic 读不出消息	死信 Topic `%DLQ%消费者组` 默认 `perm=2`（只写），要 `mqadmin updateTopicPerm -p 6` 改成读写
死信队列有消息但没人发现，一周后才被客诉暴露	死信队列只要有消息就必须告警，死信代表确定性的业务数据损失
死信消费者里调了业务接口，自己也失败了，消息彻底无处可去	死信消费者只做「落异常表 + 告警」这类绝不会失败的操作
新消费者组上线后收不到上线前的消息	`consumeFromWhere` 默认 `CONSUME_FROM_LAST_OFFSET`，需要补历史数据要显式改成 `FIRST_OFFSET` 或 `TIMESTAMP`
新组设了 `CONSUME_FROM_FIRST_OFFSET`，上线瞬间消费 3 天历史消息把下游打挂	用 `CONSUME_FROM_TIMESTAMP` 精确指定起点，并提前做好限流
`resetOffsetByTime -s now` 跳过堆积后，才发现那是核心业务的消息	位点重置是高危操作，必须评审 + 确认业务可接受，且重置前要先停消费者
运行中重置位点，被消费者的位点上报覆盖，重置无效	重置前先停掉消费者进程，重置完成后再启动
顺序消息中一条消息反复重试，整个队列阻塞 40 分钟，1/32 的订单停止更新	顺序消费必须设重试上限，超过就跳过（落异常表）；更优的是改成乱序 + 状态机版本校验
顺序消息扩容队列数后，同一订单的新旧消息分散到不同队列导致乱序	`hash % queueCount` 的结果会变，存量消息消费完之前不能扩容队列
以为用了 `MessageQueueSelector` 就有序了，结果还是乱	乱序有三个并发点：发送并发、队列分散、消费并发，三个都要解决（同步发送 + 哈希选队列 + Orderly 监听器）
事务消息的 `checkLocalTransaction` 依赖内存变量，生产者重启后回查全部返回 UNKNOWN	回查只能查数据库，必须能在生产者重启后依然给出确定答案
事务消息回查 15 次仍 UNKNOWN，Broker 直接回滚，消息悄悄没了	回查逻辑必须在 15 分钟内能定论；同时必须有对账任务兜底
以为事务消息能保证下游一定消费成功	事务消息只保证「本地事务成功 ⇒ 消息能被投递」，下游可靠性要靠重试 + 死信 + 幂等
用事务消息做顺序业务，发现消息顺序完全乱了	半消息恢复到真实 Topic 的顺序与发送顺序无关，事务消息不支持顺序
本地消息表的定时任务多实例部署，同一条消息被多个实例同时重投	定时任务必须加分布式锁（Redisson / 调度框架的单机执行模式）
本地消息表涨到 5000 万行，扫描 SQL 变慢拖垮业务库	建 `idx_scan(status, next_retry_at)` 索引 + 定期归档已确认记录（保留 7 天）
消息体塞了 2MB 的导入文件，把 PageCache 挤爆，同 Broker 上十几个 Topic 延迟全部飙升	消息只放「事件 + 引用」，大数据走 OSS 只传 URL；单条消息体建议控制在 4KB 以内
批量发送总大小超过 4MB，整批发送失败	发送前用 `ListSplitter` 按 4MB 分片
单条消息超过 `maxMessageSize`（4MB），直接报 `MESSAGE_ILLEGAL`	要么拆分消息，要么走 OSS 传引用，不要盲目调大 Broker 参数
核心交易 Topic 和离线导入 Topic 混在同一个 Broker，互相影响	混合存储下所有 Topic 共用 CommitLog 和 PageCache，核心业务要独立集群物理隔离
同步刷盘上线后吞吐掉了一半，接口超时	同步刷盘只给资金类业务用；普通业务用异步刷盘 + 同步复制折中（消息至少在两台机器内存里）
Broker 磁盘用到 90%，突然开始拒绝写入	`diskMaxUsedSpaceRatio` 默认 88%，超过会拒写；磁盘使用率 75% 就要告警并扩容/缩短保留时长
只监控了堆积条数，一个低 QPS Topic 堆积 1 万条（延迟 16 分钟）没被发现	消费延迟时间才是和业务影响强相关的指标，告警应该以时间为主、条数为辅
消费线程池复用后 traceId 串号，排查时看到的日志链路完全是错的	消费逻辑必须在 `finally` 里 `MDC.clear()`
全量 Topic 开启消息轨迹，Broker 磁盘 IO 翻倍	轨迹会产生约一倍的额外消息量，只给核心 Topic 开启，并配置独立的轨迹 Topic 和保留时长
`consumeMessageBatchMaxSize` 调大后，一条失败导致整批 100 条全部重试	批量消费要么保证整批幂等，要么实现「部分成功」语义（失败条目单独重投）
大促时关单消息集中在同一秒到期，产生投递毛刺	用 5.x 的 `setDeliveryTimestamp` 给到期时间加 0~60 秒随机抖动打散峰值；4.x 延时等级是固定档位抖不了，别在 payload 里塞个 jitter 字段自欺欺人
延时关单消息直接执行关单，把用户刚支付的订单给关了	延时消息只是触发器，消费时必须回查当前状态，并用带状态条件的 SQL 保证互斥
关单消息丢了，超时订单永远不关单，库存一直被占	MQ 是主路径，必须配兜底扫表任务（每小时扫「超时未支付」订单补偿关单）
扩容消费者后把数据库直接压死，故障从「延迟」升级成「不可用」	扩容消费者时必须给下游加限流（RateLimiter），宁可慢点追也不能打挂下游
消息发送用了 `sendOneway`，以为很快，结果丢消息排查不出来	`sendOneway` 不关心结果、无法感知失败，只能用于日志这类可丢场景
异步发送没实现 `SendCallback` 的 `onException`，发送失败完全没人知道	异步发送必须实现失败回调并落本地消息表或告警
生产者在本地事务提交之前发了 MQ，结果事务回滚产生了幽灵消息	发消息必须在事务提交之后（`TransactionSynchronization#afterCommit`），或用事务消息/本地消息表
MQ 挂了导致下单接口直接报错，核心链路被非核心依赖拖死	发送失败要降级写本地消息表，MQ 恢复后补发；核心链路绝不能因 MQ 不可用而失败
RocketMQ NameServer 只配了一个地址，这台挂了客户端全部失联	NameServer 要配多个地址（分号分隔），它们之间无状态、互不通信，配几个就有几重保障
以为 NameServer 是集群会自动同步，结果新建的 Topic 在另一台上查不到	NameServer 节点之间完全不通信，Broker 会向所有 NameServer 注册，少配一个就会出现元数据不一致的错觉
Broker 主从用了异步复制，主宕机后切到从库丢了最后一段消息	资金类业务用 `SYNC_MASTER`；或升级到 DLedger（Raft）模式实现自动选主 + 强一致
传统主从模式下主节点宕机无法自动切换，从节点只能读不能写	用 DLedger 模式（至少 3 节点）实现自动故障转移
消费者 `consumeThreadMax` 设成 200，结果把下游连接池耗尽	消费线程数要和下游承载能力匹配，不是越大越好，一般 20~64
以为 MQ 能做到 Exactly Once	做不到。重试、ACK 丢失、Rebalance 都会产生重复，系统只能做到「不丢 + 可能重复」，所以幂等是必选项不是可选项
对账任务被当成可选项没做，MQ 边界情况导致的不一致全部变成客诉	关键业务必须有定时对账：扫出不一致数据主动补偿。任何 MQ 配置都存在边界情况，对账是最后一道防线

📌 当日复盘任务

一、MQ 基础与选型（能脱口而出）

[ ] 说清 MQ 的三大核心价值：异步、解耦、削峰，各举一个真实业务例子
[ ] 说清引入 MQ 的四个代价：可用性降低、复杂度上升、一致性问题、排查成本上升
[ ] 能给出「该不该用 MQ」的判断标准：这个操作是不是必须同步拿到结果
[ ] 默写 RabbitMQ / RocketMQ / Kafka / Pulsar 的四维对比表（吞吐、延迟、功能、适用场景）
[ ] 能说清 Kafka 为什么分区多了性能下降，而 RocketMQ 为什么队列到万级也不衰减
[ ] 能说清自己项目为什么选了当前这个 MQ，以及如果重新选会怎么选
[ ] 知道 Kafka 的 KRaft 模式是什么、从哪个版本开始生产可用

二、RabbitMQ 核心机制

[ ] 画出 AMQP 模型：Producer → Exchange → Binding → Queue → Consumer
[ ] 默写四种 Exchange 的路由规则，并说清 * 和 # 的区别
[ ] 说清 Connection 和 Channel 的关系，以及为什么Channel 不能跨线程共享
[ ] 默写持久化三要素（交换机 durable + 队列 durable + 消息 deliveryMode=2）
[ ] 说清 publisher confirm 和 return callback 分别解决什么问题
[ ] 说清三种 ACK 模式的差异，以及 prefetch 不设置的后果
[ ] 说清 DLX 的三个触发条件，并解释为什么 TTL + DLX 能当延时队列用
[ ] 能解释 default-requeue-rejected=true 为什么会造成毫秒级死循环
[ ] 说清 lazy queue 的作用和适用场景，以及内存/磁盘告警会阻塞生产者
[ ] 说清 RabbitMQ 高可用三代方案：普通集群 → 镜像队列 → Quorum Queue（Raft）

三、RocketMQ 核心机制

[ ] 画出四大组件架构图：NameServer / Broker / Producer / Consumer
[ ] 说清 NameServer 节点之间完全不通信，以及这个设计带来的利弊
[ ] 画出存储模型：CommitLog + ConsumeQueue + IndexFile，并说清各自作用
[ ] 说清 ConsumeQueue 为什么是固定 20 字节，以及它能被完整重建的原因
[ ] 说清四种刷盘/复制组合的取舍，并说出为什么异步刷盘 + 同步复制是最优折中
[ ] 说清 Topic / Queue / 消费者数量的关系，以及「队列数就是并发上限」
[ ] 默写五种消息类型（普通/顺序/延时/事务/批量）和三种发送方式（同步/异步/单向）
[ ] 说清 Push 消费者底层其实是 Pull + 长轮询，以及长轮询的实现原理
[ ] 说清集群消费和广播消费的区别，以及各自的位点存储位置
[ ] 说清 Rebalance 的触发时机、四种分配策略，以及 Rebalance 期间的重复消费问题
[ ] 默写延时等级的 18 个值，以及超过 2 小时该怎么办
[ ] 说清死信 Topic 的命名规则和「默认只写不可读」的坑

四、可靠性三件套（面试必考）

[ ] 默写消息丢失的三个环节及每个环节的解决方案
[ ] 能写出 RabbitMQ 和 RocketMQ 各自的「不丢消息」完整配置清单
[ ] 说清为什么 Exactly Once 做不到（三个原因：生产重试、ACK 丢失、Rebalance）
[ ] 默写幂等的三层防线：布隆过滤器 / Redis 两态 SETNX / 数据库唯一索引
[ ] 能解释为什么不能用 msgId 做幂等 key
[ ] 能解释为什么 Redis 幂等必须区分 PROCESSING 和 DONE 两个状态
[ ] 说清「让业务操作天然幂等」为什么是最优解，并写出带状态条件的 UPDATE 语句
[ ] 默写异常分类处理表：瞬时异常重试、业务异常落表、数据异常进死信
[ ] 说清为什么要自己控制重试次数而不是用默认 16 次
[ ] 能说清对账兜底为什么是必选项，并设计一个对账任务

五、顺序、事务与延时（三大高频难点）

[ ] 默写乱序的三个并发点及对应解法
[ ] 手写顺序消息的完整代码（MessageQueueSelector + MessageListenerOrderly）
[ ] 说清顺序消费的三个坑：队列阻塞、扩容破坏顺序、Rebalance 短暂阻塞
[ ] 能提出「乱序 + 状态机版本校验」这个比顺序消息更优的方案
[ ] 默写事务消息的六步完整流程，重点说清事务回查的作用
[ ] 手写 executeLocalTransaction 和 checkLocalTransaction，并强调回查只查数据库
[ ] 说清本地消息表 / 事务消息 / Seata AT / TCC 四种方案的适用场景
[ ] 手写本地消息表的表结构、扫描 SQL（含索引）和分布式锁定时任务
[ ] 默写延时消息的四种实现及各自缺陷，重点说清消息级 TTL 的队头阻塞
[ ] 手写延时关单的完整三件套：MQ 主路径 + 扫表兜底 + 状态机互斥

六、性能、堆积与运维

[ ] 说清 mmap 和 sendfile 的本质差异，以及 RocketMQ 为什么选 mmap
[ ] 说清 PageCache 的作用、刷盘抖动的原因，以及文件预热（写 0 + mlock）
[ ] 说清 Broker 端 TAG 过滤和 SQL92 过滤的区别与性能影响
[ ] 默写堆积排查三步法和应急决策树（五种原因对应五种处置）
[ ] 能解释为什么「加消费者实例」有时完全无效
[ ] 手写搬运工方案的代码，并说清它为什么不能用于顺序消息
[ ] 说清批量消费的收益（RT 摊薄）和风险（整批重试放大）
[ ] 默写位点重置的两个用途（跳过堆积 / 回溯重放）和各自风险
[ ] 说清 consumeFromWhere 三个取值的行为差异和踩坑点
[ ] 默写监控告警指标清单（资源层 / 链路层 / 业务层），说清为什么「消费延迟时间」比「堆积条数」重要
[ ] 说清为什么死信队列只要有消息就必须告警
[ ] 会用 mqadmin 的核心命令：consumerProgress / topicStatus / queryMsgById / resetOffsetByTime / updateTopic / updateTopicPerm

七、输出与复盘（当天必须产出）

[ ] 用 ASCII 图画出自己项目的完整 MQ 链路（生产 → Broker → 消费 → 下游 → 对账）
[ ] 整理一份自己项目的 MQ 生产配置清单，逐项说明为什么这么设
[ ] 写出自己项目中至少两个真实的 MQ 故障案例：现象 → 定位过程 → 根因 → 修复 → 长效改进
[ ] 准备好「保证消息不丢失」这道题的 90 秒口述版，并实际计时练习 3 遍
[ ] 准备好「怎么保证幂等」这道题的 90 秒口述版，重点讲清两态 SETNX 的必要性
[ ] 准备好「堆积几百万怎么办」这道题，要求能当场画出决策树
[ ] 准备好「事务消息完整流程」这道题，要求能当场画出六步时序
[ ] 为自己项目补齐缺失的那一环（大概率是对账任务或死信告警），并写成改进方案
[ ] 把本文的 5 道手撕题在 IDE 里真实跑通一遍，不要只看不写
[ ] 复盘：今天哪些知识点是「以为自己会，实际讲不清」的，单独列出来明天重背

Day18 消息队列 MQ：RabbitMQ / RocketMQ，异步解耦、削峰、事务消息、死信队列 ​

📖 基础知识讲解【基础原理 + 高级深挖知识点】 ​

基础原理 ​

1. MQ 到底解决了什么问题：异步、解耦、削峰 ​

2. 引入 MQ 的代价：必须能回答的四个问题 ​

3. 四大主流 MQ 选型对比 ​

4. AMQP 协议与 RabbitMQ 的核心模型 ​

5. Connection 与 Channel：为什么不是一个线程一个连接 ​

6. 四种 Exchange 的路由规则与选型 ​

7. RabbitMQ 的持久化三要素：少一个都会丢 ​

8. 发送端可靠性：publisher confirm 与 return callback ​

9. 消费端可靠性：三种 ACK 模式与 prefetch ​

10. RocketMQ 的四大组件与整体架构 ​

11. RocketMQ 的存储模型：CommitLog、ConsumeQueue、IndexFile ​

12. 刷盘机制与主从复制：两组配置决定可靠性等级 ​

13. Topic、Queue 与消费者数量的关系 ​

14. RocketMQ 的五种消息类型 ​

15. Push 还是 Pull：长轮询的真相 ​

16. 消费模式：集群消费与广播消费 ​

17. Rebalance：消费者负载均衡机制 ​

18. 消息重试与延时等级 ​

19. 死信队列：进入条件与处理规范 ​

20. 消息可靠性全链路：三处可能丢失 ​

21. 消费幂等：为什么 MQ 做不到 Exactly Once ​

✨ 高级深挖知识点 ​

1. 消息乱序的根因：三个并发点 ​

2. 顺序消息在消费端的三个坑 ​

3. RocketMQ 事务消息：半消息 + 事务回查 ​

4. 分布式事务三种方案的对比：本地消息表、事务消息、Seata ​

5. 零拷贝：mmap 与 sendfile 的本质差异 ​

6. PageCache 与刷盘抖动：RocketMQ 的性能毛刺从哪来 ​

7. Broker 端消息过滤：TAG 与 SQL92 ​

8. 消息堆积：排查顺序与应急扩容方案 ​

9. 消费者端限流与反压 ​

10. RabbitMQ 高可用的三代方案 ​

11. RabbitMQ 的惰性队列 Lazy Queue ​

12. 消息去重的工程实现：三层防线 ​

13. 重试与死信的边界：哪些异常该重试 ​

14. 消费位点管理与位点重置 ​

15. 延时消息的三种实现与 RocketMQ 5.0 的任意精度延时 ​

16. 消息轨迹与可观测性 ​

17. 大消息与消息压缩 ​

18. Kafka 与 RocketMQ 的核心差异 ​

19. 生产配置清单（Spring Boot 落地） ​

20. 监控告警指标清单 ​

❓ 高频面试题【基础简答 + 连环追问链 + 精简口述答案 + 项目关联话术】 ​

题 1：为什么要用消息队列？引入 MQ 会带来哪些问题？ ​

题 2：怎么保证消息不丢失？ ​

题 3：怎么保证消息不被重复消费（幂等）？ ​

题 4：死信队列是什么？哪些场景会进死信？ ​

题 5：RocketMQ 事务消息的完整流程是什么？ ​

题 6：怎么保证消息的顺序性？ ​

题 7：线上堆积了几百万条消息怎么办？ ​

题 8：延时消息怎么实现？延时关单你是怎么做的？ ​

题 9：RabbitMQ 有哪几种 Exchange？分别什么场景用？ ​

题 10：RocketMQ 为什么用 CommitLog 混合存储？和 Kafka 的分区存储比有什么取舍？ ​

题 11：RabbitMQ、RocketMQ、Kafka 怎么选型？ ​

题 12：MQ 怎么实现分布式事务的最终一致性？ ​

✍️ 手撕代码【核心实现 + 边界 Case 自测清单 + 变种拓展题】 ​

题 1：实现一个可靠的 RabbitMQ 消费者（手动 ACK + 幂等 + 死信） ​

题 2：实现一个本地消息表的最终一致性组件 ​

题 3：实现 RocketMQ 事务消息的完整收发链路 ​

题 4：实现延时关单（RocketMQ 延时等级版 + RabbitMQ TTL+DLX 版） ​

题 5：实现消息堆积的应急处理程序（搬运工扩容 + 批量消费） ​

⚠️ 当日踩坑实录 ​

📌 当日复盘任务 ​

Day18 消息队列 MQ：RabbitMQ / RocketMQ，异步解耦、削峰、事务消息、死信队列

📖 基础知识讲解【基础原理 + 高级深挖知识点】

基础原理

1. MQ 到底解决了什么问题：异步、解耦、削峰

2. 引入 MQ 的代价：必须能回答的四个问题

3. 四大主流 MQ 选型对比

4. AMQP 协议与 RabbitMQ 的核心模型

5. Connection 与 Channel：为什么不是一个线程一个连接

6. 四种 Exchange 的路由规则与选型

7. RabbitMQ 的持久化三要素：少一个都会丢

8. 发送端可靠性：publisher confirm 与 return callback

9. 消费端可靠性：三种 ACK 模式与 prefetch

10. RocketMQ 的四大组件与整体架构

11. RocketMQ 的存储模型：CommitLog、ConsumeQueue、IndexFile

12. 刷盘机制与主从复制：两组配置决定可靠性等级

13. Topic、Queue 与消费者数量的关系

14. RocketMQ 的五种消息类型

15. Push 还是 Pull：长轮询的真相

16. 消费模式：集群消费与广播消费

17. Rebalance：消费者负载均衡机制

18. 消息重试与延时等级

19. 死信队列：进入条件与处理规范

20. 消息可靠性全链路：三处可能丢失

21. 消费幂等：为什么 MQ 做不到 Exactly Once

✨ 高级深挖知识点

1. 消息乱序的根因：三个并发点

2. 顺序消息在消费端的三个坑

3. RocketMQ 事务消息：半消息 + 事务回查

4. 分布式事务三种方案的对比：本地消息表、事务消息、Seata

5. 零拷贝：mmap 与 sendfile 的本质差异

6. PageCache 与刷盘抖动：RocketMQ 的性能毛刺从哪来

7. Broker 端消息过滤：TAG 与 SQL92

8. 消息堆积：排查顺序与应急扩容方案

9. 消费者端限流与反压

10. RabbitMQ 高可用的三代方案

11. RabbitMQ 的惰性队列 Lazy Queue

12. 消息去重的工程实现：三层防线

13. 重试与死信的边界：哪些异常该重试

14. 消费位点管理与位点重置

15. 延时消息的三种实现与 RocketMQ 5.0 的任意精度延时

16. 消息轨迹与可观测性

17. 大消息与消息压缩

18. Kafka 与 RocketMQ 的核心差异

19. 生产配置清单（Spring Boot 落地）

20. 监控告警指标清单

❓ 高频面试题【基础简答 + 连环追问链 + 精简口述答案 + 项目关联话术】

题 1：为什么要用消息队列？引入 MQ 会带来哪些问题？

题 2：怎么保证消息不丢失？

题 3：怎么保证消息不被重复消费（幂等）？

题 4：死信队列是什么？哪些场景会进死信？

题 5：RocketMQ 事务消息的完整流程是什么？

题 6：怎么保证消息的顺序性？

题 7：线上堆积了几百万条消息怎么办？

题 8：延时消息怎么实现？延时关单你是怎么做的？

题 9：RabbitMQ 有哪几种 Exchange？分别什么场景用？

题 10：RocketMQ 为什么用 CommitLog 混合存储？和 Kafka 的分区存储比有什么取舍？

题 11：RabbitMQ、RocketMQ、Kafka 怎么选型？

题 12：MQ 怎么实现分布式事务的最终一致性？

✍️ 手撕代码【核心实现 + 边界 Case 自测清单 + 变种拓展题】

题 1：实现一个可靠的 RabbitMQ 消费者（手动 ACK + 幂等 + 死信）

题 2：实现一个本地消息表的最终一致性组件

题 3：实现 RocketMQ 事务消息的完整收发链路

题 4：实现延时关单（RocketMQ 延时等级版 + RabbitMQ TTL+DLX 版）

题 5：实现消息堆积的应急处理程序（搬运工扩容 + 批量消费）

⚠️ 当日踩坑实录

📌 当日复盘任务