学习笔记 消息队列 分布式 消息中间件 RocketMQ

RocketMQ入门

围绕消息模型、架构设计与存储机制理解 RocketMQ

Posted by Ekko on August 20, 2020

这篇笔记的主线是把 RocketMQ 本身的核心知识点和原理机制串起来，包括消息模型、架构角色、消费语义、顺序消息、事务消息和存储设计等几个层面。

Kafka 对比只作为学习过程中的辅助参照，用来澄清 Topic、Queue、顺序语义和存储组织这些容易混淆的概念，而不是本文唯一的关注点。文章重心仍然放在 RocketMQ 自身的知识体系与设计思路上。

参考资料：

Apache RocketMQ Documentation

JavaGuide - RocketMQ

Kafka 入门笔记

三太子敖丙 - RocketMQ

[TOC]

消息队列扫盲

回顾下消息队列能做什么：异步、解耦、削峰

《消息队列》章有提到异步、解耦、削峰

异步：

首先看下 同步通信： 比如有一个购票系统，需求是用户在购买完之后能接收到购买完成的短信

RocketMQ章节同步通信.png

省略中间的网络通信时间消耗，假如购票系统处理需要 150ms ，短信系统处理需要 200ms ，那么整个处理流程的时间消耗就是 150ms + 200ms = 350ms

用户购票在购票系统的时候其实就已经完成了购买，而现在通过 同步调用 非要让整个请求拉长时间，而短信系统又不是很有必要，它仅仅是一个辅助功能增强用户体验感而已

现在整个调用流程就有点 头重脚轻 的感觉了，购票是一个不太耗时的流程，而现在因为同步调用，非要等待发送短信这个比较耗时的操作才返回结果。那如果再加一个发送邮件呢？

RocketMQ章节同步通信2.png

这样整个系统的调用链又变长了，整个时间就变成了 550ms

为了解决这一问题，在系统中加入消息中间件（消息队列），将上面的模型进行改造后：

RocketMQ章节异步通信.png

这样，在将消息存入消息队列之后就可以直接返回了，所以整个耗时只是 150ms + 10ms = 160ms

用户体验上时间变短了，但是整个系统上的调用链的时间并没有受影响，比如发送短信，依然是 150 + 200 （大致）

解耦：

回到上面的同步调用过程，写个伪代码简单概括一下：

1
2
3
4
5
6
7
8

public void purchaseTicket(Request request){
    // 校验
    validate(request);
    // 购票
    Result result = purchase(request);
    // 发送短信
    sendMessage(result);
}

接下来，系统中又添加了一个发送邮件，这个时候我们又必须去重新修改代码同步调用；如果购买完后又需要加积分，那么还得再次修改代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

public void purchaseTicket(Request request){
    // 校验
    validate(request);
    // 购票
    Result result = purchase(request);
    // 发送短信
    sendMessage(result);
    // 发送邮件
    sendEmail(result);
    // 添加积分
    addPoint(result);
    // 后面可能还会有其他需求
    ...
    ...
}

如果新需求要求不要再发送邮件，那么这个时候又得改代码

此时我们就用一个消息队列在中间进行解耦。需要注意的是，后面的发送短信、发送邮件、添加积分等一些操作都依赖于上面的 result ，这东西抽象出来就是购票的处理结果 ，比如订单号，用户账号等等，也就是说我们后面的一系列服务都是需要同样的消息来进行处理。既然这样，我们是不是可以通过 “广播消息” 来实现

我上面所讲的“广播”并不是真正的广播，而是接下来的系统作为消费者去 订阅 特定的主题。比如这里的主题就可以叫做 订票 ，我们购买系统作为一个生产者去生产这条消息放入消息队列，然后消费者订阅了这个主题，会从消息队列中拉取消息并消费。就比如我们刚刚画的那张图，会发现在生产者这边只需要关注 生产消息到指定主题中 ，而 消费者只需要关注从指定主题中拉取消息 就行了

Rocket章节之解耦.png

如果没有消息队列，每当一个新的业务接入，我们都要在主系统调用新接口、或者当我们取消某些业务，我们也得在主系统删除某些接口调用。有了消息队列，我们只需要关心消息是否送达了队列，至于谁希望订阅，接下来收到消息如何处理，是下游的事情，无疑极大地减少了开发和联调的工作量

削峰：

回到一开始我们使用同步调用系统的情况，并且思考一下，如果此时有大量用户请求购票整个系统会变成什么样？

Rocket章节之削峰.png

如果，此时有一万的请求进入购票系统，我们知道运行我们主业务的服务器配置一般会比较好，所以这里我们假设购票系统能承受这一万的用户请求，那么也就意味着我们同时也会出现一万调用发短信服务的请求。而对于短信系统来说并不是我们的主要业务，所以我们配备的硬件资源并不会太高，那么你觉得现在这个短信系统能承受这一万的峰值么，且不说能不能承受，系统会不会 直接崩溃 了？

短信业务又不是我们的主业务，我们能不能 折中处理 呢？如果我们把购买完成的信息发送到消息队列中，而短信系统 尽自己所能地去消息队列中取消息和消费消息 ，即使处理速度慢一点也无所谓，只要我们的系统没有崩溃就行了

消息队列的副作用：

  1. 降低了系统的可用性
  2. 系统的复杂度
  3. 重复消费消息问题
  4. 消息顺序消费问题
  5. 解决分布式事务问题
  6. 解决消息堆积的问题

这些问题后面慢慢讲到

RocketMQ简介

RocketMQ 是一个纯 Java 实现的分布式消息中间件，前身是阿里开源的 MetaQ，后来捐赠给 Apache 并发展成为顶级开源项目。它既保留了传统消息队列在业务解耦、异步处理和削峰填谷方面的优势，也围绕顺序消息、事务消息、定时消息等典型业务场景做了较强的工程化支持。

如果只从“都是消息队列”这个角度去看，RocketMQ 和 Kafka 很容易被混在一起；但如果从“它们分别如何组织消息、如何做消费并发、如何理解顺序语义”这几个角度切入，两者的设计侧重点其实并不相同。

这篇文章的主线可以概括为四层：

  • 从 Topic、Queue、Group 这些概念入手理解 RocketMQ 的消息模型
  • 从 NameServer、Broker、Producer、Consumer 理解 RocketMQ 的整体架构
  • 从顺序消息、重复消费、事务消息理解 RocketMQ 面向业务问题时的处理方式
  • 从 CommitLog、ConsumeQueue、刷盘与复制理解 RocketMQ 的底层存储机制

核心模块

  • rocketmq-broker：接受生产者发来的消息并存储（通过调用rocketmq-store），消费者从这里取得消息
  • rocketmq-client：提供发送、接受消息的客户端 API。
  • rocketmq-namesrv：NameServer，类似于 Zookeeper，这里保存着消息的TopicName，队列等运行时的元信息。
  • rocketmq-common：通用的一些类，方法，数据结构等。
  • rocketmq-remoting：基于 Netty4 的 client/server + fastjson 序列化 + 自定义二进制协议。
  • rocketmq-store：消息、索引存储等。
  • rocketmq-filtersrv：消息过滤器 Server，需要注意的是，要实现这种过滤，需要上传代码到 MQ！（一般而言，我们利用Tag足以满足大部分的过滤需求，如果更灵活更复杂的过滤需求，可以考虑 filtersrv 组件）。
  • rocketmq-tools：命令行工具。

四大核心： NameServer、Broker、Producer以及 Consumer 四部分

RocketMQ 的整体认知

如果把 RocketMQ 看成一条完整的数据链路，那么它大致可以拆成下面几个环节：

  1. Producer 根据 Topic 从 NameServer 获取路由信息
  2. Producer 按照负载均衡策略把消息发送到目标 Broker 的某个 Queue
  3. Broker 先把消息顺序写入 CommitLog
  4. Broker 再为 Topic / Queue 维护 ConsumeQueue 和索引信息
  5. Consumer 根据 Topic 路由找到对应 Broker，并从自己负责的 Queue 拉取消息
  6. Consumer Group 维护各自的消费位点，持续推进消费进度

也就是说，RocketMQ 的核心并不只是“把消息从生产者送到消费者”，而是围绕下面几件事建立了一整套机制：

  • 如何做路由发现
  • 如何承载高并发写入
  • 如何组织多个 Topic 和 Queue
  • 如何保证消费并发与顺序语义
  • 如何兼顾可靠性、性能和可扩展性

如果先抓住这个整体工作链路，再分别去看 NameServer、消息模型、顺序消费和存储结构，很多分散的概念就能自动串起来。

和 Kafka 的对应关系（辅助理解）

从 Kafka 切换到 RocketMQ 时，最容易模糊的地方通常不是定义本身，而是“哪些概念是一一对应的，哪些只是看起来像”。先把这张对应关系表建立起来，后面的内容会清晰很多：

维度 Kafka RocketMQ 应该怎样理解
逻辑主题 Topic Topic 两者都是消息分类的逻辑概念
物理拆分单元 Partition Queue / Message Queue Kafka 用 Partition 做物理分片；RocketMQ 在 Topic 下配置多个 Queue 来承载并发
消费并发基础 Partition 数量 Queue 数量 两者都依赖“一个主题拆成多个子单元”来提高并发能力
顺序保证边界 单个 Partition 内有序 单个 Queue 内有序 两者都不天然保证 Topic 级别全局顺序
元数据管理 ZooKeeper / KRaft NameServer 都承担路由和元数据发现职责，但实现方式不同
存储组织 每个 Partition 本身就是一条独立追加日志 多个 Topic / Queue 共享 CommitLog，再用 ConsumeQueue 建立逻辑索引 Kafka 更像“分区日志天然可消费”，RocketMQ 更像“共享主日志 + 队列索引”

这一层的核心结论可以概括为：

  • Kafka：Topic 下面拆的是 Partition
  • RocketMQ：Topic 下面拆的是 Queue
  • 两者都只保证拆分单元内部有序，不保证整个 Topic 全局有序

不过，这个“Queue 类似 Kafka 的 Partition”只适合帮助入门建立第一层映射，不能简单理解成它们完全一样。原因在于：

  • Kafka 更强调“Partition 就是核心存储和消费单位”
  • RocketMQ 更强调“消息先顺序写入 CommitLog，再通过 ConsumeQueue 组织成可消费的逻辑队列”

也就是说，在概念层面可以把 Queue 暂时类比成 Kafka 的 Partition，但到了存储层面，两者的内部实现已经明显分叉。

存储组织差异为什么会影响使用场景

如果把这一点再往前推进一步，就会发现 Kafka 和 RocketMQ 在“高吞吐日志场景”上的气质确实不完全一样。

在 Kafka 中，每个 Partition 本身就是一条独立的追加日志：

  • Producer 把消息写入某个 Partition
  • Broker 只需要持续向该 Partition 对应的日志段末尾追加数据
  • Consumer 也直接围绕 Partition 顺序拉取和推进 offset

这种设计使 Kafka 的“存储单元”“消费单元”“顺序保证单元”三者高度重合，因此它天然贴合事件流、埋点日志、监控日志、CDC 这类大吞吐、长时间保留、强调回放和顺序扫描的场景。

而在 RocketMQ 中，Queue 在消费语义上虽然类似 Partition，但消息主体并不是分别写进各自独立的 Queue 日志文件，而是先统一进入共享 CommitLog，再通过 ConsumeQueue 把消息映射回 Topic / Queue 视角进行消费。

这意味着：

  • RocketMQ 的写路径同样是顺序写，也同样能获得很高吞吐
  • 但它的读写组织方式更偏“共享主日志 + 逻辑索引”的工程设计
  • 它更强调业务消息系统常见的能力组合，例如事务消息、Tag 过滤、定时消息、顺序消息等

因此，更准确的说法不是“RocketMQ 吞吐就不高”，而是：

  • Kafka 更原生地适合做高吞吐日志流和事件流平台
  • RocketMQ 也有很高吞吐，但整体设计更偏向业务消息中间件

前者更强调“分区日志本身就是核心抽象”，后者更强调“在高性能前提下，把业务消息投递、索引、过滤和事务机制组合起来”。

队列模型和主题模型

消息队列为什么要叫消息队列？

早期的消息中间件是通过 队列 这一模型来实现的，所以都习惯把消息中间件称为消息队列

但是，如今例如 RocketMQ 、Kafka 这些优秀的消息中间件不仅仅是通过一个 队列 来实现消息存储的

队列模型：

就像我们理解队列一样，消息中间件的队列模型就真的只是一个队列

消息队列之队列模型.png

在一开始提到了一个 “广播” 的概念，也就是说如果此时我们需要将一个消息发送给多个消费者(比如需要将信息发送给短信系统和邮件系统)，这个时候单个队列即不能满足需求了

当然也可以让 Producer 生产消息放入多个队列中，然后每个队列去对应每一个消费者。问题是可以解决，创建多个队列并且复制多份消息是会很影响资源和性能的。而且，这样子就会导致生产者需要知道具体消费者个数然后去复制对应数量的消息队列，这就违背我们消息中间件的 解耦 这一原则

主题模型：

主题模型可以解决上面 队列模型 的问题（多个队列，达不到解耦目的）

主题模型 也可以称为 发布订阅模型

在主题模型中，消息的生产者称为 发布者(Publisher) ，消息的消费者称为 订阅者(Subscriber)，存放消息的容器称为 主题(Topic)

其中，发布者将消息发送到指定主题中，订阅者需要 提前订阅主题 才能接受特定主题的消息

消息队列之主题模型.png

RocketMQ中的消息模型

RocketMQ 中的消息模型本质上也是 主题模型 的一种实现。

不过，不同消息中间件虽然都声称自己支持“发布/订阅”，底层承载这个模型的结构并不一样。例如：

  • Kafka 中的 分区
  • RocketMQ 中的 队列
  • RabbitMQ 中的 Exchange

可以把 主题模型 / 发布订阅模型 理解成一层抽象接口，而不同消息中间件则用各自的数据结构把它落到物理实现上。

RocketMQ 对这层模型的实现方式是：一个 Topic 下面维护多个 Queue，生产者把消息写入某个 Queue，消费者组再并行消费这些 Queue。 根据下面这张图，可以先建立整体印象。

RocketMQ主题模型.png

可以看到在整个图中有 Producer Group、Topic、Consumer Group 三个关键角色。

  • Producer Group 生产者组： 代表某一类的生产者，比如我们有多个秒杀系统作为生产者，这多个合在一起就是一个 Producer Group 生产者组，它们一般生产相同的消息
  • Consumer Group 消费者组： 代表某一类的消费者，比如我们有多个短信系统作为消费者，这多个合在一起就是一个 Consumer Group 消费者组，它们一般消费相同的消息
  • Topic 主题： 代表一类消息，比如订单消息、物流消息等

图中最关键的地方在于：生产者并不是把消息写进一个抽象的 Topic 就结束了，而是最终要落到 Topic 下的某个具体 Queue 中。

每个 Topic 都可以配置多个 Queue。集群消费模式下，一个消费者组中的多个消费者会共同消费这个 Topic 下的多个 Queue，但同一时刻一个 Queue 只会被组内一个消费者处理。如果某个消费者挂掉，组内其它消费者会接替它原本负责的 Queue。

这一点和 Kafka 非常接近：在 Kafka 中，一个 Partition 同一时刻只会分配给组内一个 Consumer；在 RocketMQ 中，则对应为一个 Queue 同一时刻只会分配给组内一个 Consumer。

因此，Queue 数量在 RocketMQ 中承担的角色，和 Partition 数量在 Kafka 中非常相似：

  • 决定消费并发上限
  • 决定生产端可分摊的写入压力
  • 影响消费者实例是否会空闲

当然也可以消费者个数小于队列个数，只不过不太建议。如下图：

rocket消费组个数小于队列个数.png

每个消费组为什么都要在每个队列上维护一个消费位置？

因为刚刚画的仅仅是一个消费者组，在发布订阅模式中一般会涉及到多个消费者组，而每个消费者组在每个队列中的消费位置都是不同的。如果此时有多个消费者组，那么消息被一个消费者组消费完之后是不会删除的(因为其它消费者组也需要呀)，它仅仅是为每个消费者组维护一个 消费位移(offset) ，每次消费者组消费完会返回一个成功的响应，然后队列再把维护的消费位移加一，这样就不会出现刚刚消费过的消息再一次被消费了

RocketMQ消费位移.png

为什么一个 Topic 中需要维护多个 Queue？

答案可以概括为：提高并发能力。 这和 Kafka 在一个 Topic 下配置多个 Partition 的动机是一样的。

理论上，一个 Topic 只保留一个 Queue 也能工作，因为它同样可以维护每个消费者组各自的消费位置，也同样能实现发布/订阅。如下图：

RocketMQ主题单一队列情况.png

但这样一来，生产者写入只能集中到单个 Queue，消费者组内部也很难展开并行消费，并发能力会明显受限。

总结来说，RocketMQ 通过使用在一个 Topic 中配置多个队列并且每个队列维护每个消费者组的消费位置实现了主题模式/发布订阅模式

作为辅助参照可以这样记

如果已经熟悉 Kafka，那么 RocketMQ 这一层可以直接翻译成下面这几句话：

  • RocketMQ 的 Topic 仍然只是逻辑分类，不是最终落盘的最小消费单元
  • RocketMQ 的 Queue 可以先近似理解成 Kafka 的 Partition
  • Producer 最终也是把消息写到某个具体 Queue，而不是停留在抽象 Topic 这一层
  • Consumer Group 的并行消费能力，本质上也受 Queue 数量约束
  • RocketMQ 的顺序保证边界，同样停留在单个 Queue，而不是整个 Topic

当把这几个映射关系放在脑中之后，再去看 RocketMQ 的顺序消息、消费模型和存储结构，整体就不会再和 Kafka 混淆了

RocketMQ的架构图(NameServer\Broker\Producer\Consumer)

RocketMQ 技术架构中有四大角色 NameServer 、Broker 、Producer 、Consumer

Broker：

主要负责消息的存储、投递和查询以及服务高可用保证。说白了就是消息队列服务器，生产者生产消息到 Broker ，消费者从 Broker 拉取消息并消费

Broker是具体提供业务的服务器，单个 Broker 节点与所有的 NameServer 节点保持长连接及心跳，并会定时将 Topic 信息注册到 NameServer，顺带一提底层的通信和连接都是基于 Netty 实现的

普及一下关于 Broker 、Topic 和 队列的关系。上面讲解了 Topic 和队列的关系，一个 Topic 中存在多个队列，那么这个 Topic 和队列存放在哪呢？

一个 Topic 分布在多个 Broker 上，一个 Broker 可以配置多个 Topic ，它们是多对多的关系

如果某个 Topic 消息量很大，应该给它多配置几个队列(上文中提到了提高并发能力)，并且 尽量多分布在不同 Broker 上，以减轻某个 Broker 的压力

Topic 消息量都比较均匀的情况下，如果某个 broker 上的队列越多，则该 broker 压力越大

RocketMQ中的Topic和Broker.png

NameServer：

类似 ZooKeeper 和 Spring Cloud 中的 Eureka ，它其实也是一个 注册中心 ，主要提供两个功能：

  • Broker管理
  • 路由信息管理

说白了就是 Broker 会将自己的信息注册到 NameServer 中，此时 NameServer 就存放了很多 Broker 的信息(Broker 的路由表)，消费者和生产者就从 NameServer 中获取路由表然后照着路由表的信息和对应的 Broker 进行通信(生产者和消费者定期会向 NameServer 去查询相关的 Broker 的信息)

NameServer 与 Zookeeper 相比更轻量。主要是因为每个NameServer 节点互相之间是独立的，没有任何信息交互

NameServer压力不会太大，平时主要开销是在维持心跳和提供Topic-Broker的关系数据

但有一点需要注意，Broker 向 NameServer 发心跳时， 会带上当前自己所负责的所有 Topic 信息，如果 Topic 个数太多（万级别），会导致一次心跳中，就 Topic 的数据就几十 M，网络情况差的话，网络传输失败，心跳失败，导致 NameServer 误认为 Broker 心跳失败

NameServer 被设计成几乎无状态的，可以横向扩展，节点之间相互之间无通信，通过部署多台机器来标记自己是一个伪集群。

每个 Broker 在启动的时候会到 NameServer 注册，Producer 在发送消息前会根据 Topic 到 NameServer 获取到 Broker 的路由信息，Consumer 也会定时获取 Topic 的路由信息。

所以从功能上看 NameServer 应该是和 ZooKeeper 差不多，据说 RocketMQ 的早期版本确实是使用的 ZooKeeper ，后来改为了自己实现的 NameServer

从原理上看，NameServer 更像一个轻量级路由中心，而不是一个负责强一致协调的中心节点。它不去承担复杂的选举和一致性协议，而是通过“Broker 主动上报 + Producer/Consumer 主动拉取”的方式维持路由视图。

这种设计的优点在于：

  • 结构简单，扩容容易
  • 节点之间无复杂同步开销
  • 路由发现成本低，整体实现轻量

代价则是 NameServer 保存的是一种“近实时路由视图”，它更偏向工程上的可用与简洁，而不是像强一致协调组件那样承担全局状态裁决。

Producer：

消息发布的角色，支持分布式集群方式部署。说白了就是生产者

Producer 由用户进行分布式部署，消息由 Producer 通过多种负载均衡模式发送到 Broker 集群，发送低延时，支持快速失败

RocketMQ 提供了三种方式发送消息：同步、异步和单向

  • 同步发送： 同步发送指消息发送方发出数据后会在收到接收方发回响应之后才发下一个数据包。一般用于重要通知消息，例如重要通知邮件、营销短信
  • 异步发送： 异步发送指发送方发出数据后，不等接收方发回响应，接着发送下个数据包，一般用于可能链路耗时较长而对响应时间敏感的业务场景，例如用户视频上传后通知启动转码服务
  • 单向发送： 单向发送是指只负责发送消息而不等待服务器回应且没有回调函数触发，适用于某些耗时非常短但对可靠性要求并不高的场景，例如日志收集

Consumer：

消息消费的角色，支持分布式集群方式部署。支持以 push 推，pull 拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制。说白了就是消费者

Pull： 拉取型消费者（Pull Consumer）主动从消息服务器拉取信息，只要批量拉取到消息，用户应用就会启动消费过程，所以 Pull 称为主动消费型

Push： 推送型消费者（Push Consumer）封装了消息的拉取、消费进度和其他的内部维护工作，将消息到达时执行的回调接口留给用户应用程序来实现。所以 Push 称为被动消费类型，但从实现上看还是从消息服务器中拉取消息，不同于 Pull 的是 Push 首先要注册消费监听器，当监听器处触发后才开始消费消息

RocketMQ的架构图.png

为什么需要 NameServer，为什么不直接让 Producer、Consumer 和 Broker 彼此直连？

上文提到过 Broker 是需要保证高可用的，如果整个系统仅仅靠着一个 Broker 来维持的话，那么这个 Broker 的压力会很大，所以需要使用多个 Broker 来保证 负载均衡

如果说，消费者和生产者直接和多个 Broker 相连，那么当 Broker 修改的时候必定会牵连着每个生产者和消费者，这样就会产生耦合问题，而 NameServer 注册中心就是用来解决这个问题的

和 SpringCloud 中的 Eureka 注册中心类似

当然，RocketMQ 中的技术架构肯定不止前面那么简单，因为上面图中的四个角色都是需要做集群的。官网的架构图：

RocketMQ的架构图2.png

和上面架构图的大致没有区别，不过是做了集群而已，另外还有一些细节上的差别：

  1. Broker 做了集群并且还进行了主从部署 ，由于消息分布在各个 Broker 上，一旦某个 Broker 宕机，则该 Broker 上的消息读写都会受到影响。所以 Rocketmq 提供了 master/slave 的结构， salve 定时从 master 同步数据(同步刷盘或者异步刷盘)，如果 master 宕机，则 slave 提供消费服务，但是不能写入消息 (后面还会提到)

  2. 为了保证 HA（高可用），NameServer 也做了集群部署，但是请注意它是 去中心化 的。也就意味着它没有主节点，你可以很明显地看出 NameServer 的所有节点是没有进行 Info Replicate 的，在 RocketMQ 中是通过 单个 Broker 和所有 NameServer 保持长连接 ，并且在每隔 30 秒 Broker 会向所有 Nameserver 发送心跳，心跳包含了自身的 Topic 配置信息，这个步骤就对应这上面的 Routing Info

  3. 在生产者需要向 Broker 发送消息的时候，需要先从 NameServer 获取关于 Broker 的路由信息，然后通过 轮询 的方法去向每个队列中生产数据以达到 负载均衡 的效果

  4. 消费者通过 NameServer 获取所有 Broker 的路由信息后，向 Broker 发送 Pull 请求来获取消息数据。Consumer 可以以两种模式启动—— 广播（Broadcast）和集群（Cluster）。广播模式下，一条消息会发送给 同一个消费组中的所有消费者 ，集群模式下消息只会发送给一个消费者

消息领域模型

RocketMQ消息领域模型.png

这一章可以看作是理解 RocketMQ 的“术语地图”。如果前面关注的是整体架构，那么这里关注的就是：RocketMQ 到底用哪些核心对象来描述一条消息从生产到消费的全过程。

Message

Message（消息）就是要传输的信息本身。一条消息至少要有一个 Topic，它决定了这条消息属于哪一类业务。

除了消息体之外，一条消息通常还会携带一些辅助信息，例如：

  • Tag：用于进一步细分消息类型
  • Key：用于业务查找、排障和检索
  • 其他属性：用于描述消息附加信息

因此，从消息建模的角度看，RocketMQ 并不是简单传一段数据，而是围绕“消息属于谁、该被谁消费、如何被检索、如何区分子类型”组织了一套完整元信息。

Topic

Topic（主题）可以看作消息的一级分类。比如在电商系统里，通常可以按业务域划分出：

  • 订单消息
  • 支付消息
  • 物流消息
  • 营销消息

一个 Topic 和生产者、消费者之间是松耦合关系：

  • 一个 Topic 可以被多个 Producer 写入
  • 一个 Producer 也可以同时向多个 Topic 发送消息
  • 一个 Topic 可以被多个 Consumer Group 订阅

因此，Topic 的核心职责不是承载并发，而是提供业务语义上的分类边界。

Tag

Tag（标签）可以看作 Topic 之下的一层更细粒度分类。它不是新的 Topic，而是对同一 Topic 内部消息的进一步区分。

例如都属于“订单消息”这个 Topic，但内部又可以继续细分为：

  • 订单创建
  • 订单支付
  • 订单取消
  • 订单完成

这时就可以使用同一个 Topic，再配合不同 Tag 来组织消息。这样做的好处是：

  • Topic 数量不会无限膨胀
  • 业务语义仍然足够清晰
  • 消费端可以在同一业务域下做更灵活的订阅和过滤

Group

Group（分组）分为两类：

  • Producer Group
  • Consumer Group

它们本质上代表的是“一类功能相同、协同工作的客户端实例”。

例如多个订单服务实例可以组成一个 Producer Group，多个积分服务实例可以组成一个 Consumer Group。分组的意义不只是命名，而是让 RocketMQ 能够识别：

  • 哪些 Producer 属于同一类发送方
  • 哪些 Consumer 需要协同消费同一份消息
  • 哪些实例共同维护同一个消费进度和负载分配关系

Queue

Queue 是 RocketMQ 在 Topic 之下继续拆分出来的逻辑队列，也是消费并发和顺序语义的重要承载单元。可以先把它近似理解成 Kafka 的 Partition，但要注意它在 RocketMQ 中最终会映射到 ConsumeQueue + CommitLog 的组合结构上。

每个 Queue 内部有序；而一个 Topic 下通常会配置多个 Queue 来提升并发能力。RocketMQ 里还会区分读队列和写队列，工程上通常要求两者数量保持一致，否则容易在路由和消费分配上带来额外复杂性。

Message Queue

Message Queue 可以理解为 Topic 被进一步拆分后的消息队列单元。一个 Topic 下可以配置多个 Queue，Producer 发送消息时最终会把消息路由到某个具体 Queue。

Queue 的引入有两个直接作用：

  • 让消息写入和消费具备并发扩展能力
  • 让顺序语义可以收敛在单个 Queue 内部

因此，Topic 更偏逻辑分类，而 Message Queue 更偏实际承载消息流转的执行单元。

Offset

在 RocketMQ 中，Queue 可以近似看成一个长度理论上无限的逻辑数组，而 Offset 就是这个数组中的位置下标。

消费端正是依靠 Offset 来判断“自己已经消费到哪里”。也正因为此，消费位点推进、消息重试、回溯消费等能力，最终都要落到 Offset 的维护上。

消息消费模式

RocketMQ 的消息消费模式主要有两种：

  • Clustering：集群消费
  • Broadcasting：广播消费

这两种模式的核心区别，不在于“消费者怎么拉消息”，而在于同一条消息在同一个 Consumer Group 内部究竟应该被消费几次。

集群消费

集群消费是 RocketMQ 的默认模式，也是最常见的生产环境模式。

在这种模式下：

  • 一个 Consumer Group 会共同消费某个 Topic 下的多个 Queue
  • 同一时刻，一个 Queue 只会分配给组内一个 Consumer
  • 如果组内某个 Consumer 宕机，其他 Consumer 会接管它原本负责的 Queue

这意味着，集群消费更强调负载均衡和横向扩展。它适合的场景通常是：

  • 订单处理
  • 库存扣减
  • 物流状态更新
  • 积分发放

这些场景的共同点是：一条消息通常只需要被同一个消费组中的某个实例处理一次。

广播消费

广播消费的语义则完全不同。

在广播模式下，同一个 Consumer Group 内的每个 Consumer 都会收到这条消息。也就是说，这条消息不是在组内“分摊”，而是在组内“全量广播”。

广播消费适合的场景通常包括：

  • 配置刷新通知
  • 本地缓存失效通知
  • 轻量级监控事件分发
  • 每个实例都必须感知一次的控制类消息

它的核心特点不是提高吞吐，而是保证同组内每个实例都能看到同一份消息。

两种消费模式应该如何选择

如果一条消息只需要被一个消费者实例处理，那么应该选择集群消费；如果同一条消息需要在同组内的每个消费者实例上都执行一次，那么才应该选择广播消费。

换句话说：

  • 集群消费关注的是“如何分工”
  • 广播消费关注的是“如何全员感知”

消息顺序

RocketMQ 中的消息顺序通常分为两类：

  • Orderly：顺序消费
  • Concurrently：并行消费

顺序消费

顺序消费表示：消费者处理消息的顺序，应当与生产者向同一 Message Queue 发送消息的顺序保持一致。

这里最关键的限定条件是：顺序只在单个 Queue 内成立。

这意味着：

  • 如果同一业务主键的消息都进入同一个 Queue，就有机会保证业务顺序
  • 如果消息被打散到多个 Queue，就不存在 Topic 级别的天然全局顺序

如果业务要求的是绝对全局顺序，那么通常只能把消息收敛到极少数 Queue，甚至单 Queue；代价则是并发能力明显下降。

并行消费

并行消费不再承诺严格顺序，它追求的是更高的消费吞吐。消费端可以借助线程池并发处理消息，因此更适合大多数“顺序要求不强，但吞吐要求较高”的业务场景。

它适合的典型场景包括：

  • 埋点上报处理
  • 通知发送
  • 非关键异步任务
  • 允许短暂乱序但要求高处理效率的业务链路

因此，在 RocketMQ 中，顺序与吞吐通常是一组需要权衡的指标：越强调严格顺序，并发空间通常越小；越强调吞吐，越需要接受顺序约束的收缩。

Topic、Queue 和 Kafka Partition 的关系

把 RocketMQ 和 Kafka 最容易混淆的几个词放在一起时，可以按下面这个层级来理解：

1
2
3
4
5
6
7
8
9
10
11

Kafka:
  Topic
    ├── Partition-0
    ├── Partition-1
    └── Partition-2

RocketMQ:
  Topic
    ├── Queue-0
    ├── Queue-1
    └── Queue-2

这两个结构在“认知入口”上非常像，但阅读时要注意三个差别：

  1. 在 Kafka 中，Partition 同时承担了存储分片、消费并发和顺序保证这几层职责。
  2. 在 RocketMQ 中，Queue 在消费模型上和 Kafka Partition 很接近，但在底层存储上并不是“一队列一份独立日志文件”那样的直观结构。
  3. RocketMQ 的消息首先顺序写入 CommitLog，再由 ConsumeQueue 为每个 Topic / Queue 建立索引，因此它更像“共享主日志 + 逻辑队列索引”的模型。

因此，当看到 RocketMQ 的 Queue 时，第一反应可以先把它当成 Kafka 的 Partition 来理解消费语义；但看到 RocketMQ 的 CommitLog 和 ConsumeQueue 时，就要切换到它自己的存储模型中来理解了。

消息可靠性机制

在真正把 RocketMQ 用到业务里时，单纯理解 Topic、Queue 和 Consumer Group 还不够。更关键的问题往往是：消息会不会丢？发送成功到底意味着什么？消费失败之后会怎样？一条始终处理不了的消息最终会被放到哪里？

这一组问题可以统一放到“消息可靠性机制”下面来看。RocketMQ 并不是靠某一个单点特性来保证可靠性，而是通过发送确认、刷盘策略、复制策略、消费确认、重试机制、死信队列等多层机制共同完成。

生产消息全链路时序图

如果把一条消息从发送到最终消费完成，或者进入重试 / 死信队列的过程放到同一张图里，大致可以抽象成下面这条链路：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

Producer
   |
   | 1. 根据 Topic 向 NameServer 查询路由
   v
NameServer
   |
   | 2. 返回 Topic -> Broker / Queue 路由信息
   v
Producer
   |
   | 3. 选择目标 Broker / Queue，发送消息
   v
Broker
   |
   | 4. 先顺序写入 CommitLog
   | 5. 再构建对应 Topic / Queue 的 ConsumeQueue 索引
   |
   | 6. 按发送方式 + 刷盘策略 + 复制策略返回 ACK
   v
Producer

-------------------- 消费阶段 --------------------

Consumer
   |
   | 7. 根据 Topic 路由找到 Broker，拉取自己负责的 Queue
   v
Broker
   |
   | 8. 根据 ConsumeQueue 定位 CommitLog 中的消息体
   v
Consumer
   |
   | 9. 执行业务逻辑
   |
   |-- 成功 -> 返回 CONSUME_SUCCESS -> Broker 推进消费位点
   |
   `-- 失败 -> 返回 RECONSUME_LATER
               |
               v
            Retry Topic / Retry Queue
               |
               | 多次重试后仍失败
               v
            DLQ（Dead Letter Queue）

这张图里最关键的不是箭头本身，而是它把 RocketMQ 的几层机制放到了同一条时间线上：

  • NameServer 负责路由发现，而不直接承载消息
  • Broker 负责真正接收、存储和转发消息
  • CommitLog 负责顺序写入消息主体
  • ConsumeQueue 负责让消息按 Topic / Queue 维度被高效消费
  • ACK 负责把发送成功和消费成功这两层确认区分开
  • Retry / DLQ 负责把“临时失败”和“长期失败”分层治理

也正因为这几个环节是串联的，所以 RocketMQ 的可靠性从来不是单点能力，而是一整条链路协同工作的结果。

消息可能在哪些阶段丢失

如果按一条消息的完整生命周期来看，消息丢失风险大致出现在 3 个阶段：

  1. Producer 到 Broker 之间 Producer 发送请求后，如果网络异常、Broker 不可达、请求超时，消息可能根本没有成功到达 Broker。

  2. Broker 落盘与复制阶段 即使 Broker 已经收到消息，如果只写入内存映射区但尚未刷盘，或者主从复制尚未完成，此时 Broker 异常宕机，仍然可能造成消息丢失。

  3. Broker 到 Consumer 之间 Consumer 拉取到消息后，如果业务处理已经完成，但消费成功响应没有回到 Broker，就可能触发重复投递；反过来，如果业务尚未真正完成却错误地返回消费成功，则会出现逻辑上的“消息丢失”。

也就是说，所谓“消息丢失问题”并不只发生在 Broker 端，它贯穿了发送、存储、复制、消费确认的整个过程。

RocketMQ 的 ACK 语义应该怎样理解

RocketMQ 没有像 Kafka 那样用一个单独的 acks=0/1/all 参数来概括所有确认语义，但它同样存在多层 ACK 机制。只是这些确认分散在发送方式、刷盘策略、复制策略和消费返回值之中。

可以把 RocketMQ 的 ACK 理解成两层：

  • 生产端 ACK：Broker 是否确认收到了这条消息
  • 消费端 ACK：Consumer 是否确认自己已经成功处理了这条消息

生产端 ACK

在生产端，最直接的确认来自发送结果：

  • 同步发送：Producer 会等待 Broker 返回发送结果，这是最典型的“收到 ACK 再返回”
  • 异步发送：Producer 不阻塞当前线程，但会通过回调拿到发送结果
  • 单向发送：Producer 只发不等回应，不关心 Broker 是否返回结果，可靠性最低

因此，如果业务对消息可靠性敏感，通常不应该使用单向发送，因为它天然放弃了生产端确认。

不过，RocketMQ 里的“发送成功”还要继续追问一句：Broker 返回成功时，这条消息到底持久化到什么程度了？

这就进入刷盘和复制层的确认语义了：

  • 如果是同步刷盘，Broker 通常会等消息刷盘成功后再返回，可靠性更高
  • 如果是异步刷盘，Broker 可能在消息进入内存映射区后就返回，性能更高但宕机窗口更大
  • 如果是同步复制，主从都确认后再返回，消息更安全
  • 如果是异步复制，主节点先返回成功，从节点稍后再追赶，可靠性窗口更宽

所以在 RocketMQ 中，生产端 ACK 实际上是一个组合语义：

发送方式决定“Producer 是否等待返回”，刷盘和复制策略决定“Broker 返回成功时，这条消息到底安全到了什么程度”。

消费端 ACK

消费端的 ACK 则体现在 Consumer 对消费结果的返回上。

在并发消费模型下，Consumer 一般会向 Broker 返回两类结果：

  • CONSUME_SUCCESS：表示这条消息已经成功处理，可以推进消费位点
  • RECONSUME_LATER：表示当前处理失败，希望稍后重新投递

在顺序消费模型下，语义略有不同，但本质仍然是一样的：消费成功才推进位点，消费失败则进入稍后重试或挂起当前 Queue 的逻辑。

因此，消费端 ACK 的核心不是“收到了消息”，而是“业务处理是否完成并明确向 Broker 返回成功状态”。

消息丢失通常如何规避

如果把上面的机制转成工程实践，RocketMQ 常见的可靠性增强手段包括：

  • 生产端尽量使用同步发送或可观测的异步发送
  • Broker 侧根据业务重要性选择同步刷盘
  • 主从复制要求更高时选择同步复制或更高可用方案
  • 消费端必须在业务真正处理完成后再返回成功
  • 关键业务的消费者实现幂等，接受可能的重复投递

因此，RocketMQ 的可靠性从来不是“完全不丢消息”的绝对承诺，而是“通过一组机制把丢失概率压到业务可接受范围内”。

重试机制

消息重试是 RocketMQ 可靠性设计中的另一条主线。它的作用不是保证消息永不失败，而是给临时失败一个自动恢复的机会。

RocketMQ 的重试可以从两端来看：

  • 生产端重试
  • 消费端重试

生产端重试

生产端重试发生在消息还没有被 Broker 确认成功之前。比如：

  • 网络超时
  • Broker 短暂不可用
  • 路由命中的 Broker 临时故障

这类情况下，Producer 可以根据客户端配置进行重试，以提高发送成功率。

不过生产端重试有一个天然副作用：如果第一次实际上已经成功写入 Broker，但返回结果在网络中丢失，那么重试就可能带来重复消息。因此，生产端重试提升的是送达概率，但它并不自动保证“绝不重复”。

消费端重试

消费端重试更常见，也更贴近业务。

当 Consumer 拿到一条消息后，如果业务处理失败、数据库暂时不可用、下游 RPC 超时，通常不会立刻把消息判定为永久失败，而是返回稍后重试。此时 RocketMQ 会把消息重新投递给该消费组，等待后续再次消费。

这里最关键的一点是：消费重试本质上是为了处理“暂时失败”，不是为了掩盖“永久失败”。

因此，消费端重试通常适合：

  • 下游系统短暂抖动
  • 数据库锁冲突或瞬时不可用
  • 外部依赖偶发超时
  • 短时间内可恢复的业务异常

如果某条消息连续多次重试仍然失败，就说明它大概率不是“暂时失败”，而是已经进入了需要人工关注或特殊补偿的状态。

重试和消费模式的关系

RocketMQ 的消费重试机制主要是在集群消费模式下发挥作用。

因为集群消费强调的是“同一条消息在同一个消费组里最终要有一个实例处理成功”，所以失败后重新投递是合理的。而广播消费更强调“每个实例都看到消息”，它在失败处理上的语义和集群消费并不完全相同，因此工程上更常把自动重试、死信治理这些能力放在集群消费路径下讨论。

死信队列

当一条消息经过多次重试仍然无法成功消费时，继续无限重试通常已经没有意义了。此时 RocketMQ 会把它送入死信队列（DLQ, Dead Letter Queue）。

死信队列的作用可以概括为两点：

  • 不让一条异常消息无限阻塞正常消费链路
  • 给运维和开发一个独立排查、补偿、重放的落点

在 RocketMQ 中，死信队列通常与消费组绑定。可以把它理解成：

  • 某个 Consumer Group 下有一批消息始终消费失败
  • 当超过最大重试次数后，这些消息不再继续正常重投
  • 而是进入该消费组对应的死信主题，例如 %DLQ%<ConsumerGroup>

这样做之后，正常消息仍然可以继续流转，而问题消息则被隔离出来，等待后续处理。

死信队列适合怎么用

死信队列不应该被理解成“消息彻底没用了”，它更像一个异常消息缓冲区。常见处理方式包括：

  • 人工排查消息体和业务上下文
  • 修复代码或依赖问题后重新投递
  • 做一次性补偿处理
  • 直接标记废弃，避免继续污染主链路

因此，死信队列的价值不在于“兜底存档”，而在于把“自动重试仍然失败的消息”从主消费链路中剥离出来，交给更明确的治理流程。

一条完整的可靠性链路

如果把 ACK、重试、死信队列串起来，RocketMQ 的可靠性链路可以概括成下面这条主线：

1
2
3
4
5
6
7
8
9
10
11

Producer 发送消息
   ->
Broker 接收并按刷盘/复制策略确认
   ->
Consumer 拉取并处理消息
   ->
处理成功则返回消费成功，推进位点
   ->
处理失败则进入重试
   ->
超过最大重试次数仍失败，则进入死信队列

这条链路说明了一点：RocketMQ 的可靠性不是靠某一个参数或某一个组件单独保证的，而是靠发送确认 + 持久化策略 + 消费确认 + 重试机制 + 死信治理共同完成。

如何解决 顺序消费、重复消费

这些问题不仅仅是 RocketMQ ，而是应该每个消息中间件都需要去解决的

其实 Kafka 的架构基本和 RocketMQ 类似，只是它注册中心使用了 Zookeeper 、它的 分区 就相当于 RocketMQ 中的 队列 。还有一些小细节不同会在后面提到

顺序消费：

RocketMQ 在 主题上是无序的、它只有在 队列层面才是保证有序 的

这句话如果放到 Kafka 的语境中，其实几乎可以一字不差地改写为：Kafka 在 Topic 层面不保证全局有序，它只保证单个 Partition 内有序。 两者在顺序保证边界上是非常相似的。

这又扯到两个概念：普通顺序 和 严格顺序

普通顺序： 消费者通过 同一个消费队列收到的消息是有顺序的 ，不同消息队列收到的消息则可能是无顺序的。普通顺序消息在 Broker 重启情况下不会保证消息顺序性 (短暂时间)

严格顺序： 消费者 收到的所有消息均是有顺序的。严格顺序消息 即使在异常情况下也会保证消息的顺序性

严格顺序看起来虽好，实现它可会付出巨大的代价。如果你使用严格顺序模式，Broker 集群中只要有一台机器不可用，则整个集群都不可用。现在主要场景也就在 binlog 同步

一般而言，我们的 MQ 都是能容忍短暂的乱序，所以推荐使用普通顺序模式

那么，我们现在使用了普通顺序模式。上文已经提到，Producer 在默认情况下会按照负载均衡策略向同一 Topic 下的不同 Queue 发送消息。问题也就出在这里：如果“同一个订单”的创建、支付、发货被轮询到不同 Queue，那么就无法再利用单个 Queue 内部有序这个特性来保证业务顺序了。

这一点和 Kafka 完全同构：如果同一个 orderId 的事件被打散到不同 Partition，Kafka 也同样无法保证这个订单链路的顺序。

Rocket普通消费顺序不同队列情况.png

解决思路也和 Kafka 非常一致：让同一业务主键始终进入同一个 Queue。 例如以 orderId 作为路由键，通过 Hash 取模等方式，确保同一订单始终落到同一个 Queue 中，这样队列级别的有序性才有机会转化为业务上的顺序性。

情况和 kafka 的分区类似，解决顺序消费的问题，不过只能保证生产者放入队列的顺序，客户端的消费顺序需要其他手段来保证

重复消费：

重复消费的问题无非就是 幂等， 在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。比如说，这个时候我们有一个订单的处理积分的系统，每当来一个消息的时候它就负责为创建这个订单的用户的积分加上相应的数值。可是有一次，消息队列发送给订单系统 FrancisQ 的订单信息，其要求是给 FrancisQ 的积分加上 500。但是积分系统在收到 FrancisQ 的订单信息处理完成之后返回给消息队列处理成功的信息的时候出现了网络波动(当然还有很多种情况，比如 Broker 意外重启等等)，这条回应没有发送成功

那么，消息队列没收到积分系统的回应会不会尝试重发这个消息？问题就来了，我再发这个消息，万一它又给 FrancisQ 的账户加上 500 积分怎么办呢？

所以我们需要给我们的消费者实现幂等，也就是对同一个消息的处理结果，执行多少次都不变

那么如何给业务实现幂等呢？这个还是需要结合具体的业务的。你可以使用 写入 Redis 来保证，因为 Redis 的 key 和 value 就是天然支持幂等的。当然还有使用 数据库插入法 ，基于数据库的唯一键来保证重复数据不会被插入多条

不过最主要的还是需要 根据特定场景使用特定的解决方案 ，你要知道你的消息消费是否是完全不可重复消费还是可以忍受重复消费的，然后再选择强校验和弱校验的方式。毕竟在 CS 领域还是很少有技术银弹的说法

而在整个互联网领域，幂等不仅仅适用于消息队列的重复消费问题，这些实现幂等的方法，也同样适用于，在其他场景中来解决重复请求或者重复调用的问题 。比如将 HTTP 服务设计成幂等的，解决前端或者 APP 重复提交表单数据的问题 ，也可以将一个微服务设计成幂等的，解决 RPC 框架自动重试导致的重复调用问题

分布式事务

如何解释分布式事务呢？在同一个系统中可以轻松地实现事务，但是在分布式架构中，有很多服务是部署在不同系统之间的，而不同服务之间又需要进行调用。比如此时我下订单然后增加积分，如果保证不了分布式事务的话，就会出现 A 系统下了订单，但是 B 系统增加积分失败或者 A 系统没有下订单，B 系统却增加了积分。前者对用户不友好，后者对运营商不利，这是我们都不愿意见到的

如今比较常见的分布式事务实现有 2PC、TCC 和事务消息( half 半消息机制)。每一种实现都有其特定的使用场景，但是也有各自的问题，都不是完美的解决方案

在 RocketMQ 中使用的是 事务消息加上事务反查机制 来解决分布式事务问题的

如果从设计目标上看，RocketMQ 事务消息真正要解决的不是“让两个系统共享同一个数据库事务”，而是解决下面这个更现实的问题：

  • 本地业务事务已经成功，但消息没有成功投递，下游永远收不到事件
  • 消息已经投递成功，但本地业务事务失败，下游却基于错误前提继续执行

也就是说，事务消息的关键不在于做一个“大一统事务”，而在于让本地事务状态和消息最终是否对消费者可见这两件事尽可能保持一致。

RocketMQ事务反查机制.png

在第一步发送的 half 消息 ，它的意思是 在事务提交之前，对于消费者来说，这个消息是不可见的

那么，如何做到写入消息但是对用户不可见呢？RocketMQ 事务消息的做法是：如果消息是 half 消息，将备份原消息的主题与消息消费队列，然后 改变主题 为 RMQ_SYS_TRANS_HALF_TOPIC。由于消费组未订阅该主题，故消费端无法消费 half类型 的消息，然后 RocketMQ 会开启一个定时任务，从 Topic 为RMQ_SYS_TRANS_HALF_TOPIC 中拉取消息进行消费，根据生产者组获取一个服务提供者发送回查事务状态请求，根据事务状态来决定是提交或回滚消息

可以试想一下，如果没有从第5步开始的 事务反查机制 ，如果出现网路波动第4步没有发送成功，这样就会产生 MQ 不知道是不是需要给消费者消费的问题，他就像一个无头苍蝇一样。在 RocketMQ 中就是使用的上述的事务反查来解决的，而在 Kafka 中通常是直接抛出一个异常让用户来自行解决

还需要注意的是，在 MQ Server 指向系统 B 的操作已经和系统 A 不相关了，也就是说在消息队列中的分布式事务是——本地事务和存储消息到消息队列才是同一个事务。这样也就产生了事务的最终一致性，因为整个过程是异步的，每个系统只要保证它自己那一部分的事务就行了

事务消息的时序可以拆成 6 步

从工作流程上看，RocketMQ 事务消息大致可以拆成下面这个时序：

1
2
3
4
5
6

1. Producer 先发送一条 Half Message 到 Broker
2. Broker 持久化 Half Message，但此时对 Consumer 不可见
3. Producer 开始执行本地事务
4. Producer 根据本地事务结果，向 Broker 发送 Commit 或 Rollback
5. 如果 Broker 长时间没有收到明确结果，就会发起事务回查
6. Producer 返回事务最终状态，Broker 决定让消息可见或直接丢弃

这 6 步背后的核心思想是：先把消息以“待确认”状态安全地放进 Broker，再让业务系统去执行本地事务，最后由事务结果决定消息是否真正投递给下游。

换句话说，Half Message 并不是“半条消息”，而是一条已经写入 Broker、但尚未进入可消费状态的预备消息。

为什么一定要有事务回查

事务回查机制是 RocketMQ 事务消息里最关键的一环。因为在真实网络环境下，下面这些情况都可能发生：

  • Producer 本地事务已经成功，但 Commit 指令发丢了
  • Producer 本地事务失败，但 Rollback 指令没有到达 Broker
  • Producer 在执行完本地事务后宕机，Broker 长时间拿不到最终状态

如果没有回查机制，Broker 就会卡在一种非常尴尬的状态中：消息已经收到，但既不知道该投递，也不知道该丢弃。

事务回查的作用就是让 Broker 在“不确定”时主动发问。它会回过头去询问消息对应的 Producer：“这条消息对应的本地事务到底成功了还是失败了？” 然后再根据返回结果把 Half Message 转成真正可消费的消息，或者直接回滚删除。

因此，RocketMQ 事务消息并不是一次性交互，而是一套发送预备消息 -> 执行业务事务 -> 明确结果 -> 不确定时反查的闭环机制。

事务消息解决的是哪一层一致性

RocketMQ 事务消息解决的，本质上是本地事务与消息投递之间的一致性问题，而不是让整个分布式调用链进入强一致事务。

以“下单成功后发送积分消息”为例，它保证的是：

  • 订单本地事务失败时，积分消息最终不会对下游可见
  • 订单本地事务成功时，积分消息最终会被提交给下游消费

但它并不保证“积分服务一定立刻执行成功”。因为积分系统本身仍然是异步消费消息，它有自己的重试、幂等、失败补偿逻辑。

所以事务消息所提供的，更准确地说是一种基于消息驱动的最终一致性机制：

  • 上游保证“事务成功才发出最终可见消息”
  • 下游保证“拿到消息后按自己的方式完成本地处理”
  • 整个系统通过异步解耦换取更高的可扩展性，而不是追求单次调用里的强一致

一句话总结事务消息

可以把 RocketMQ 事务消息理解成一句话：

先把消息安全地以待确认状态写入 Broker，再由本地事务结果决定这条消息是否真正对消费者可见；如果中间状态不明确，就通过事务回查补齐最终决策。

消息堆积问题

在上面我们提到了消息队列一个很重要的功能——削峰 。那么如果这个峰值太大了导致消息堆积在队列中怎么办呢？

其实这个问题可以将它广义化，因为产生消息堆积的根源其实就只有两个——生产者生产太快或者消费者消费太慢

我们可以从多个角度去思考解决这个问题，当流量到峰值的时候是因为生产者生产太快，我们可以使用一些 限流降级 的方法，当然你也可以 增加多个消费者实例去水平扩展增加消费能力 来匹配生产的激增。如果消费者消费过慢的话，我们可以先检查是否是消费者出现了大量的消费错误 ，或者打印一下日志查看是否是哪一个线程卡死，出现了锁资源不释放等等的问题

当然，最快速解决消息堆积问题的方法还是增加消费者实例，不过同时你还需要增加每个主题的队列数量

别忘了在 RocketMQ 中，一个队列只会被一个消费者消费，如果你仅仅是增加消费者实例就会出现最原始的架构图的那种情况（消费者空闲）

RocketMQ消费者空闲.png

回溯消费

回溯消费是指 Consumer 已经消费成功的消息，由于业务上需求需要重新消费，在 RocketMQ 中， Broker 在向 Consumer 投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度，例如由于 Consumer 系统故障，恢复后需要重新消费 1 小时前的数据，那么 Broker 要提供一种机制，可以按照时间维度来回退消费进度。RocketMQ 支持按照时间回溯消费，时间维度精确到毫秒

RocketMQ 的刷盘机制

在 Topic 中的 队列是以什么样的形式存在的？

队列中的消息又是如何进行存储持久化的呢？

上文中提到的 同步刷盘 和 异步刷盘 又是什么呢？

同步刷盘和异步刷盘：

RocketMQ同步刷盘和异步刷盘.png

如上图所示，在同步刷盘中需要等待一个刷盘成功的 ACK ，同步刷盘对 MQ 消息可靠性来说是一种不错的保障，但是 性能上会有较大影响 ，一般地适用于金融等特定业务场景

而异步刷盘往往是开启一个线程去异步地执行刷盘操作。消息刷盘采用后台异步线程提交的方式进行，降低了读写延迟，提高了 MQ 的性能和吞吐量，一般适用于如发验证码等对于消息保证要求不太高的业务场景

一般地，异步刷盘只有在 Broker 意外宕机的时候会丢失部分数据，可以设置 Broker 的参数 FlushDiskType 来调整你的刷盘策略(ASYNC_FLUSH 或者 SYNC_FLUSH)

Broker 在消息的存取时直接操作的是内存（内存映射文件），这可以提供系统的吞吐量，但是无法避免机器掉电时数据丢失，所以需要持久化到磁盘中

刷盘的最终实现都是使用 NIO 中的 MappedByteBuffer.force() 将映射区的数据写入到磁盘，如果是 同步刷盘 的话，在 Broker 把消息写到 CommitLog 映射区后，就会等待写入完成

异步而言 只是唤醒对应的线程，不保证执行的时机，流程如图所示

RocketMQ异步刷盘.png

存储机制

队列是以什么样的形式存在的？队列中的消息又是如何进行存储持久化的呢？

这里涉及到了 RocketMQ 是如何设计它的存储结构。首先介绍 RocketMQ 消息存储架构中的三大角色 —— CommitLog 、ConsumeQueue 和 IndexFile

如果把 RocketMQ 的存储过程拆成“写路径”和“读路径”两部分，会更容易理解：

  • 写路径： Producer 把消息发送到 Broker 后，Broker 先把消息主体顺序追加到 CommitLog
  • 索引构建： 消息写入 CommitLog 之后，再按 Topic 和 Queue 维度为其建立 ConsumeQueue 条目，并在需要时补充 IndexFile
  • 读路径： Consumer 并不是直接遍历 CommitLog 找消息，而是先根据 Topic、Queue 和消费位点定位 ConsumeQueue，再由 ConsumeQueue 映射到 CommitLog 中的物理位置读取消息

因此，RocketMQ 的存储设计可以概括成一句话：用 CommitLog 保证高效顺序写，用 ConsumeQueue 保证按 Topic / Queue 维度高效读。

RocketMQ 的存储结构可以先看成两层

如果只抓主干，RocketMQ 的存储结构可以先简化为两层：

1
2

第一层：消息主体顺序写入 CommitLog
第二层：按 Topic / Queue 维度建立 ConsumeQueue 索引

也可以画成下面这种关系：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

Producer
   |
   v
Broker
   |
   |-- CommitLog
   |     |- 顺序追加写入所有消息主体
   |
   |-- ConsumeQueue(topicA, queue0)
   |     |- 记录消息在 CommitLog 中的物理偏移
   |
   |-- ConsumeQueue(topicA, queue1)
   |     |- 记录消息在 CommitLog 中的物理偏移
   |
   |-- ConsumeQueue(topicB, queue0)
   |     |- 记录消息在 CommitLog 中的物理偏移
   |
   `-- IndexFile
         |- 支持按 key / 时间范围检索消息

这个结构最值得记住的一点是：CommitLog 负责“真正存消息”，ConsumeQueue 负责“让消息按 Topic / Queue 可被快速消费”。

  • CommitLog： 消息主体以及元数据的存储主体，存储 Producer 端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G ，文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件
  • ConsumeQueue： 消息消费队列，引入的目的主要是提高消息消费的性能(我们再前面也讲了)，由于RocketMQ 是基于主题 Topic 的订阅模式，消息消费是针对主题进行的，如果要遍历 commitlog 文件中根据 Topic 检索消息是非常低效的。Consumer 即可根据 ConsumeQueue 来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定 Topic 下的队列消息在 CommitLog 中的起始物理偏移量 offset ，消息大小 size 和消息 Tag 的 HashCode 值。consumequeue 文件可以看成是基于 topic 的 commitlog 索引文件，故 consumequeue 文件夹的组织方式如下：topic/queue/file三层组织结构，具体存储路径为：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样 consumequeue 文件采取定长设计，每一个条目共20个字节，分别为8字节的 commitlog 物理偏移量、4字节的消息长度、8字节tag hashcode，单个文件由30W个条目组成，可以像数组一样随机访问每一个条目，每个 ConsumeQueue文件大小约5.72M
  • IndexFile： IndexFile（索引文件）提供了一种可以通过 key 或时间区间来查询消息的方法。这里只做科普不做详细介绍

整个消息存储的结构，最主要的就是 CommitLoq 和 ConsumeQueue 。而 ConsumeQueue 可以大概理解为 Topic 中的队列

RocketMQ之commitLog.png

RocketMQ 采用的是混合型的存储结构，也就是 Broker 单个实例下的多个 Topic / Queue 共享 CommitLog 来存储消息主体。这种设计的核心目的，是尽量把写入组织成连续的顺序追加，从而提升写入吞吐。

这样做的直接收益在于：不必为每个 Topic、每个 Queue 都维护一套独立的大日志文件，Broker 更容易把消息聚合成连续写入，提高刷盘和顺序 I/O 的效率。但它同时也带来一个问题：如果只有 CommitLog，而没有额外索引，那么按 Topic / Queue 读取消息的代价会很高。

所以 RocketMQ 又引入了 ConsumeQueue 作为逻辑消费队列索引，专门解决“如何从共享主日志中按 Topic / Queue 高效读取消息”这个问题。消费者可以先基于 Queue 和消费位点找到 ConsumeQueue 条目，再根据条目里记录的 CommitLog 物理偏移去定位消息主体。

一条消息在 Broker 内部是怎么流转的

把抽象结构落到一条具体消息上，会更容易看懂。

假设 Producer 发送了一条消息：

1
2
3

Topic   = order-events
QueueId = 1
Body    = OrderPaid(orderId=1001)

那么 Broker 内部大致会发生下面几件事：

  1. 先把这条消息的完整内容顺序写入 CommitLog
  2. 记录这条消息在 CommitLog 中的物理偏移量和消息长度
  3. 根据 Topic=order-events 和 QueueId=1，找到对应的 ConsumeQueue
  4. 在这个 ConsumeQueue 中追加一条固定长度索引项，指向刚才那条 CommitLog 记录
  5. 如果消息带有业务 Key，再根据需要写入 IndexFile，便于后续按 Key 检索

这样一来，RocketMQ 就同时满足了两件事：

  • 写入时只需要围绕 CommitLog 做高效顺序追加
  • 读取时又可以站在 Topic / Queue 的视角快速定位目标消息

Consumer 读取消息时为什么不直接扫 CommitLog

因为 CommitLog 是所有消息共享的主日志，它天然更适合写，而不适合按 Topic / Queue 粒度直接消费。

如果 Consumer 每次拉消息都直接扫 CommitLog，就会遇到几个问题：

  • 需要不断跳过大量不属于当前 Topic / Queue 的消息
  • 读取代价会随着日志增长而持续变大
  • 很难高效按消费位点推进

ConsumeQueue 的意义就在这里。它相当于给每个 Topic / Queue 准备了一份轻量、定长、可顺序访问的消费索引。Consumer 拿着自己的消费位点，只需要在对应 ConsumeQueue 中向后读取即可，不必每次都去碰整个 CommitLog。

这也是 RocketMQ 存储设计最精妙的地方之一：读写分工清晰。

  • CommitLog 专注写入吞吐
  • ConsumeQueue 专注消费定位
  • IndexFile 专注按 Key / 时间检索

为什么这种设计既快又适合扩展

RocketMQ 采用“共享 CommitLog + 多份逻辑索引”的方式，本质上是在做两种优化：

  1. 优化写入路径 多个 Topic / Queue 的消息最终汇聚到 CommitLog，Broker 更容易把写入组织成顺序 I/O，提高吞吐。

  2. 优化读取路径 不让 Consumer 直接面对整个主日志，而是通过 ConsumeQueue 先完成 Topic / Queue 维度的裁剪，再回到 CommitLog 取消息体。

因此，这套结构非常适合 RocketMQ 的典型场景：Topic 多、Queue 多、写入并发高、消费按队列推进、同时还要支持按业务 Key 排查消息。

和 Kafka 的日志组织方式对比来看

把 RocketMQ 和 Kafka 放在一起看，这里的差别会更直观。

Kafka 更像下面这种组织方式：

1
2
3
4

Topic
  ├── Partition-0 -> 独立日志文件 / 日志段
  ├── Partition-1 -> 独立日志文件 / 日志段
  └── Partition-2 -> 独立日志文件 / 日志段

也就是说，Kafka 的每个 Partition 从存储上看就是一条天然独立的追加日志。生产者写哪个 Partition，消费者就直接读哪个 Partition，顺序语义、存储组织和消费推进几乎是天然对齐的。

RocketMQ 则更像这样：

1
2
3
4
5
6
7
8

多个 Topic / Queue
        |
        v
   共享 CommitLog
        |
        +--> ConsumeQueue(topicA, queue0)
        +--> ConsumeQueue(topicA, queue1)
        +--> ConsumeQueue(topicB, queue0)

因此，你的理解方向是对的：Kafka 的确更强调“Partition 各自独立顺序写入”，而 RocketMQ 的存储组织不太一样。也正因为这种差异，Kafka 往往更贴近“分布式提交日志 / 事件流平台”的定位；而 RocketMQ 虽然同样依赖顺序写获得高性能，但它在整体能力设计上更偏业务消息系统。

不过这里还要补一个重要的限定：

  • Kafka 更适合高吞吐日志场景，并不意味着 RocketMQ 吞吐低
  • 两者都大量利用了顺序写、页缓存、批量处理等手段
  • 真正的区别在于“核心抽象和系统定位”不同，而不只是单点性能高低

所以，把这件事记成下面这句话会更准确：

Kafka 更像以 Partition 日志为核心的事件流平台；RocketMQ 更像以业务消息投递能力为核心、再辅以高性能存储设计的消息中间件。

一句话总结存储机制

可以把 RocketMQ 的存储机制概括成一句话：

所有消息先统一顺序写入 CommitLog，再通过 ConsumeQueue 和 IndexFile 把“如何按 Topic / Queue 消费”与“如何按 Key 检索”这两件事补齐。

RockeMQ存储架构.png

首先，在最上面的那一块就是我刚刚讲的你现在可以直接 把 ConsumerQueue 理解为 Queue。

在图中最左边说明了 红色方块代表被写入的消息，虚线方块代表等待被写入的。左边的生产者发送消息会指定 Topic 、QueueId 和具体消息内容，而在 Broker 中管你是哪门子消息，他直接 全部顺序存储到了 CommitLog 。而根据生产者指定的 Topic 和 QueueId 将这条消息本身在 CommitLog 的偏移(offset)，消息本身大小，和 tag 的 hash 值存入对应的 ConsumeQueue 索引文件中。而在每个队列中都保存了 ConsumeOffset 即每个消费者组的消费位置，而消费者拉取消息进行消费的时候只需要根据 ConsumeOffset 获取下一个未被消费的消息就行了

CommitLog 文件要设计成固定大小的长度呢？内存映射机制

同步复制和异步复制

上面的同步刷盘和异步刷盘是在单个结点层面的，而同步复制和异步复制主要是指的 Borker 主从模式下，主节点返回消息给客户端的时候是否需要同步从节点

  • 同步复制： 也叫 “同步双写”，也就是说，只有消息同步双写到主从结点上时才返回写入成功
  • 异步复制： 消息写入主节点之后就直接返回写入成功

然而，很多事情是没有完美的方案的，就比如我们进行消息写入的节点越多就更能保证消息的可靠性，但是随之的性能也会下降，所以需要程序员根据特定业务场景去选择适应的主从复制方案

那么，异步复制会不会也像异步刷盘那样影响消息的可靠性呢？

答案是不会的，因为两者就是不同的概念，对于消息可靠性是通过不同的刷盘策略保证的，而像异步同步复制策略仅仅是影响到了 可用性 。为什么呢？其主要原因是 RocketMQ 是不支持自动主从切换的，当主节点挂掉之后，生产者就不能再给这个主节点生产消息了

比如这个时候采用异步复制的方式，在主节点还未发送完需要同步的消息的时候主节点挂掉了，这个时候从节点就少了一部分消息。但是此时生产者无法再给主节点生产消息了，消费者可以自动切换到从节点进行消费(仅仅是消费)，所以在主节点挂掉的时间只会产生主从结点短暂的消息不一致的情况，降低了可用性，而当主节点重启之后，从节点那部分未来得及复制的消息还会继续复制

在单主从架构中，如果一个主节点挂掉了，那么也就意味着整个系统不能再生产了。那么这个可用性的问题能否解决呢？一个主从不行那就多个主从的呗，别忘了在我们最初的架构图中，每个 Topic 是分布在不同 Broker 中的

但是这种复制方式同样也会带来一个问题，那就是无法保证 严格顺序 。在上文中我们提到了如何保证的消息顺序性是通过将一个语义的消息发送在同一个队列中，使用 Topic 下的队列来保证顺序性的。如果此时我们主节点A负责的是订单A的一系列语义消息，然后它挂了，这样其他节点是无法代替主节点A的，如果我们任意节点都可以存入任何消息，那就没有顺序性可言了

而在 RocketMQ 中采用了 Dledger 解决这个问题。他要求在写入消息的时候，要求至少消息复制到半数以上的节点之后，才给客⼾端返回写⼊成功，并且它是⽀持通过选举来动态切换主节点的。这里我就不展开说明了，读者可以自己去了解

也不是说 Dledger 是个完美的方案，至少在 Dledger 选举过程中是无法提供服务的，而且他必须要使用三个节点或以上，如果多数节点同时挂掉他也是无法保证可用性的，而且要求消息复制板书以上节点的效率和直接异步复制还是有一定的差距的

定时消息

定时消息是指消息发到 Broker 后，不能立刻被 Consumer 消费，要到特定的时间点或者等待特定的时间后才能被消费。

如果要支持任意的时间精度，在 Broker 层面，必须要做消息排序，如果再涉及到持久化，那么消息排序要不可避免的产生巨大性能开销。

RocketMQ 支持定时消息，但是不支持任意时间精度，支持特定的 level，例如定时 5s，10s，1m 等

CATALOG
    FEATURED TAGS
    工作 学习笔记 Java 分布式 JAVA 数据库 面试 AI Redis 可观测性 基础 微服务 消息中间件 消息队列 缓存中间件 Prometheus Spring 高并发 JVM Metrics RAG 协议 实战 工具 检索 源码分析 缓存 Dubbo Grafana LLM Linux MySQL RPC RocketMQ Spring Boot
    FRIENDS