Java 学习网站

主题

RocketMQ ​

Java 高级 · 消息中间件

RocketMQ 是阿里开源、经双十一历练的分布式消息引擎。它在"业务友好"特性上尤为突出：事务消息、延迟队列、死信队列都是电商/金融场景的刚需。本文聚焦其存储结构（CommitLog + ConsumeQueue）、刷盘与复制策略、事务消息半消息机制，讲清它为何既快又可靠。

一、核心角色与消息模型 ​

  • NameServer：轻量路由注册中心，无状态、彼此不通信（与 ZooKeeper 不同，避免强一致开销）。Broker 定时注册心跳，Producer/Consumer 从 NameServer 拉路由表。
  • Broker：存储与投递消息，主从架构（Master/Slave）。
  • Producer / Consumer：生产/消费。

消息模型：Topic → MessageQueue（队列，类比 Kafka 分区，但消费以队列为最小单位）→ Consumer Group（集群模式下组内消费者均摊队列）。

二、存储结构：CommitLog 顺序写 + ConsumeQueue 索引 ​

RocketMQ 性能的命根是统一顺序写：所有 Topic 的所有消息都追加写进同一个 CommitLog 文件（1G 一个，顺序写盘），再异步构建按 Topic/Queue 维度的 ConsumeQueue 索引和 IndexFile（按 key 查）供消费/查询。

  • CommitLog：存消息正文，顺序写（机械/SSD 都友好），零拷贝读（mmap/sendfile）。
  • ConsumeQueue：每个 Entry 固定 20 字节（commitLogOffset 8 + size 4 + tagsCode 8），Consumer 拉消息时先读 ConsumeQueue 定位，再去 CommitLog 取正文。因为定长 + 顺序，消费检索极快。
  • IndexFile：支持按 message key 反查消息（排查问题用）。

TIP

为什么 RocketMQ 不像 Kafka 那样"每分区一个日志文件"？因为 Kafka 多分区多文件，磁盘寻道分散；RocketMQ 把所有消息收敛到一个 CommitLog 顺序写，用 ConsumeQueue 做逻辑索引。这换来了极致的写入吞吐，代价是消费时需"索引→CommitLog"两次读（但 ConsumeQueue 很小，基本在页缓存）。这是两种设计哲学的取舍。

三、刷盘与复制：可靠性两维 ​

可靠性由两个独立维度控制：

  1. 刷盘策略（flushDiskType）：
    • SYNC_FLUSH（同步刷盘）：消息落盘（fsync）才返回，最可靠，吞吐低。金融核心用。
    • ASYNC_FLUSH（异步刷盘）：写页缓存即返回，后台线程刷盘，吞吐高，宕机可能丢少量未刷盘消息。
  2. 主从复制（brokerRole）：
    • ASYNC_MASTER：主写完即返回，异步同步给从。主宕机可能丢未同步消息。
    • SYNC_MASTER：主+从都写成功才返回，强可靠（性能略降）。

生产建议：SYNC_MASTER + SYNC_FLUSH 用于关键业务（如交易），ASYNC 组合用于日志等可丢场景。

四、事务消息：半消息 + 本地事务 + 回查 ​

这是 RocketMQ 最亮眼的能力，解决"本地事务与消息发送的一致性"——典型场景"下单后发券"：订单库落了，但发券消息没发出去，就数据不一致。

  • Half Message（半消息）：先发一条"对消费者不可见"的消息，只落 CommitLog（特殊 tag 标记），确保消息一定能先存下来。
  • 本地事务执行后再发 Commit/Rollback：Commit 后 Broker 把半消息转为正常消息（可被消费）；Rollback 则丢弃。
  • 回查（checkback）：若 Producer 在发 Commit 前宕机，Broker 不知道结果，会定时回调 Producer 的 checkLocalTransaction 问"那笔本地事务到底成没成"，根据返回决定提交或回滚。这是"最终一致"的兜底。
java
TransactionMQProducer producer = new TransactionMQProducer("order_group");
producer.setTransactionListener(new TransactionListener() {
    public LocalTransactionState executeLocalTransaction(Message m, Object arg) {
        return orderSvc.create(arg) ? COMMIT_MESSAGE : ROLLBACK_MESSAGE;
    }
    public LocalTransactionState checkLocalTransaction(MessageExt m) {
        return orderSvc.exist(m.getTransactionId()) ? COMMIT_MESSAGE : UNKNOW;
    }
});
producer.sendMessageInTransaction(new Message("order_topic", body), order);

五、延迟消息与死信队列 ​

  • 延迟消息：Broker 把延迟消息先存到内部 SCHEDULE_TOPIC_XXXX，按延迟等级（1s/5s/10s/30s/1m... 共 18 级）由定时线程到点后投递到真实 Topic。注意：RocketMQ 延迟是固定等级而非任意时间（任意时间需自己二次调度）。
  • 重试与死信：消费者返回 RECONSUME_LATER，消息进入 %RETRY% 队列按退避策略（10s→30s→1m... 最多 16 次）重试；超过最大重试次数进入 %DLQ%（死信队列），不再自动投递，需人工/定时任务补偿。

六、生产落地与误区 ​

WARNING

  • 把 RocketMQ 当"强一致"存储：事务消息只保证"本地事务与消息发送"最终一致，消费者仍需幂等（可能重复投递）。
  • 消费逻辑抛异常不返回 RECONSUME_LATER：直接抛会触发无限重试直至进死信，应明确返回状态并做好重试上限处理。
  • Broker 主从不一致就切只读：异步复制下主宕机可能丢未同步消息，关键场景用 SYNC_FLUSH + SYNC_MASTER。
  • Topic 队列数随意：队列数影响并发消费能力，需与消费实例数匹配，避免消费者空闲。
  • 多 Master 多 Slave + 同步复制/刷盘（关键业务），NameServer ≥ 2 节点。
  • 消费端实现幂等（唯一键去重表），应对重试。
  • 死信队列定期巡检 + 补偿/告警流程，避免消息"石沉大海"。
  • 监控堆积量（DiffTotal）与消费 TPS，设阈值告警。

📚 官方文档与参考资料 ​

✏️ 练习与自测 ​

  1. 画出事务消息"半消息 + 回查"的完整时序，并说明回查解决了什么问题（Producer 宕机的兜底）。
  2. 设计"订单创建后发券"场景，如何用事务消息保证订单与消息的一致性？给出 Commit/Rollback/回查三者逻辑。
  3. 消费失败达到重试上限进入死信后，应如何设计补偿流程（死信巡检 + 幂等重投 + 告警）？
  4. 比较 RocketMQ 的 CommitLog+ConsumeQueue 与 Kafka 的分区日志在存储设计上的取舍。

本节小结

RocketMQ 以 NameServer 轻路由 + CommitLog 顺序写 + ConsumeQueue 索引支撑高吞吐；事务消息用"半消息 + 本地事务 + 回查"打通业务与消息一致性；延迟/死信提供业务兜底。要点：事务消息非强一致（消费必幂等）、刷盘复制策略按业务分级、死信要兜底巡检。它是电商/金融消息场景的优选。