从零到一学习中间件之 RocketMQ

本文系统梳理 Apache RocketMQ 的核心概念、架构设计、存储引擎、消息类型、高可用机制与实战用法，力求让从未接触过消息中间件的读者也能建立完整认知体系。

一、为什么需要消息中间件

在分布式系统中，服务之间最直接的通信方式是同步 RPC 调用——服务 A 调用服务 B，等待 B 返回结果后再继续。这种模式在小规模、低流量的场景下工作良好，但随着业务规模增长，三个问题逐渐暴露：

1. 同步阻塞导致级联故障：当下游服务 B 响应变慢或宕机时，上游服务 A 的请求线程被阻塞，A 自己也开始积压请求，最终雪崩效应将整个调用链拖垮。
2. 流量突增压垮下游：秒杀、大促等场景下，瞬间涌入的请求直接打到数据库或下游服务，超出其处理能力导致服务不可用。
3. 系统间强耦合：每新增一个下游消费者，上游都需要修改代码增加调用逻辑，发布协调成本高，牵一发而动全身。

消息中间件通过引入一个"中间代理人"角色来缓解这些问题。上游系统不再直接调用下游，而是把请求封装成一条消息发送给中间件，由中间件负责将消息可靠地投递给下游。这种异步通信模式带来了三重收益：

• 解耦：上游只管发消息，不关心谁消费、何时消费。新增消费者无需修改上游代码。
• 削峰填谷：中间件作为大容量缓冲池，吸收瞬时流量洪峰，下游按自身能力匀速消费。
• 异步提升响应速度：上游发出消息即可返回，不必等待下游处理完成，用户体验更流畅。

RocketMQ 正是一款诞生于阿里巴巴超大规模电商场景、专为金融级可靠业务消息设计的分布式消息中间件。

二、RocketMQ 简介与发展历程

2.1 诞生背景

2012 年，阿里巴巴内部面临双 11 等极端高并发场景，当时使用的 ActiveMQ 无法满足万亿级消息量的吞吐和可靠性要求。阿里巴巴中间件团队基于 MetaQ（早期的内部消息系统）重新设计并开源，这就是 RocketMQ 的起源。2016 年底，RocketMQ 成为 Apache 顶级项目，是国内首个非 Hadoop 体系的 Apache 顶级项目。

2.2 版本演进路线

RocketMQ 5.x 的架构演进围绕三个目标展开：消息基础设施的云原生化、集成效率的痛点升级（全新 gRPC SDK）、以及面向 AI 场景的异步通信能力。

2.3 适用场景

• 业务消息与微服务解耦：订单处理、支付通知、库存扣减等异步链路
• 金融级事务消息：分布式事务的最终一致性保障
• 事件流处理：与 Flink/Spark Streams 集成做实时计算
• 延迟任务调度：订单超时取消、定时提醒等
• IoT 设备消息：MQTT 协议支持百万级设备连接
• AI Agent 通信（5.5+）：多智能体异步编排与协调

三、核心概念与领域模型

理解 RocketMQ 的关键在于掌握它的领域模型。RocketMQ 采用异步通信模型和发布/订阅（Pub/Sub）消息传输模型，一条消息的完整生命周期经历三个阶段：生产 → 存储 → 消费。

3.1 领域模型总览

3.2 核心概念逐项解析

Topic（主题）

Topic 是消息的分类容器，类似于数据库中的"表"。生产者将消息发送到指定 Topic，消费者订阅 Topic 来消费消息。一个 Topic 由多个 MessageQueue 组成，分布在不同的 Broker 上以实现水平扩展。

MessageQueue（消息队列）

MessageQueue 是 Topic 的物理分区，概念上等同于 Kafka 的 Partition。它是并发和顺序的基本单元——同一队列内的消息严格有序，不同队列间不保证顺序。每个 Topic 默认创建 4 个队列（可配置），可根据吞吐需求调整。

Message（消息）

消息是数据传输的最小单位，包含以下核心字段：

Producer（生产者）

生产者是消息的发送方，轻量且无状态。同一个生产者组内的生产者实例可以分担消息发送负载。RocketMQ 支持三种发送方式：同步发送、异步发送和单向发送（Oneway）。

ConsumerGroup（消费者组）

消费者组是一组共享相同消费逻辑和配置的消费者实例集合。同一组内的消费者共同消费订阅的消息——在集群模式下，每条消息只被组内一个消费者处理；在广播模式下，每条消息会被组内所有消费者各处理一次。

Consumer（消费者）

消费者是消息的接收方，必须隶属于某个消费者组。RocketMQ 提供两种消费模型：Push 模型（服务端推送，实为长轮询拉取）和 Pull/Simple 模型（消费者主动拉取）。

Subscription（订阅关系）

订阅关系是消费者组与 Topic 之间的绑定契约，定义了消息过滤规则、重试策略和消费进度等配置。订阅关系持久化在 Broker 端，不受消费者重启或断连影响。

3.3 两种消息传输模型

消息中间件有两种经典传输模型：

• 点对点模型（Queue）：消费者匿名，每条消息只被一个消费者消费。结构简单但扩展性有限。
• 发布/订阅模型（Pub/Sub）：消费者以组身份订阅，同一条消息可被多个独立的消费者组各自完整消费。扩展性强，支持一对多通信。

RocketMQ 采用发布/订阅模型，天然支持多消费者组独立消费同一 Topic 的全量消息。

四、整体架构设计

RocketMQ 的架构经历了从 4.x 经典架构到 5.x 云原生架构的演进。理解这两个版本的设计差异，有助于在实际选型和部署时做出正确决策。

4.1 四大核心组件

无论哪个版本，RocketMQ 都由四个核心角色组成：

NameServer 是一个轻量级的服务注册与发现中心，维护集群中所有 Broker 的元数据信息（Topic 路由、Broker 地址等）。Broker 启动时向所有 NameServer 注册自身信息并定期发送心跳。NameServer 采用 AP 模式（Available & Partition Tolerant），各节点间互不通信、数据独立——这意味着某一台 NameServer 宕机不影响其他节点服务。Producer 和 Consumer 启动时从任一 NameServer 获取路由信息，并在本地缓存，定期刷新。

RocketMQ 没有选择 ZooKeeper/etcd 这类强一致协调组件作为注册中心，而是自研 NameServer，主要原因是：在消息系统中，路由信息的短暂不一致（几秒级别）是可以容忍的，而 AP 模式带来的更高可用性更加关键。

Broker 是消息系统的核心，承担消息的接收、存储、投递和状态维护。每个 Broker 管理一组 Topic 的队列数据，多个 Broker 组成集群共同服务海量 Topic。Broker 内部包含消息存储引擎、网络通信模块、消费管理模块等核心子系统。

4.2 4.x 经典架构

在 4.x 版本中，NameServer、Broker、Producer、Consumer 四个角色直接通信：

4.x 架构的特点是 Broker 同时承担计算和存储职责，客户端直接通过自研的 Remoting 协议（基于 Netty 的私有 TCP 协议）与 Broker 通信。这种架构简单直接，但存在两个局限：一是计算和存储耦合，无法独立扩缩容；二是私有协议限制了多语言客户端的生态。

4.3 5.x 云原生架构

5.x 引入了存算分离架构，新增了 Proxy 层和 Controller 组件，形成了四层架构：

5.x 架构的核心改进包括：

• 存算分离：Proxy 层（无状态）负责协议转换、鉴权、负载均衡等计算逻辑；Store 层（有状态）专注消息存储。两者可独立扩缩容——IoT 场景下大量设备连接时，可单独扩容 Proxy 而不影响存储节点。
• gRPC 多语言 SDK：基于 gRPC 重新设计了 Java、Go、C++、Rust、Python、Node.js 等多语言客户端，API 统一且集成成本低。同时还支持 CloudEvents SDK（云原生事件标准）和 MQTT SDK（IoT 场景）。
• Controller 组件：实现 Broker 主节点的自动切换，当 Master 宕机时自动将 Slave 提升为 Master，无需人工干预。4.x 时代主从切换需要借助 DLedger 或手动操作。
• 消息粒度负载均衡：5.0 引入了消息级别的负载均衡机制，打破了传统队列级别分配在消费者数量与队列数量不整除时的不均衡问题。
• 部署灵活性：Proxy 和 Broker 可以合并部署（小规模场景）也可以分离部署（大规模场景）。

五、消息存储引擎深度解析

存储引擎是 RocketMQ 的核心竞争力之一。它在保证高吞吐写入的同时，还能支持数万级 Topic 并存——这是 Kafka 在大规模 Topic 场景下难以做到的。

5.1 存储文件体系

RocketMQ 的消息存储由三类文件协作完成：

5.2 CommitLog —— 消息的物理存储

CommitLog 是所有消息的物理存储文件。无论消息属于哪个 Topic，所有消息都按到达顺序追加写入同一个 CommitLog 文件目录下的文件中。每个 CommitLog 文件固定大小 1GB，文件名为起始偏移量（20 位数字补零），写满后创建新文件。

这种"所有 Topic 共享一个 CommitLog"的设计是 RocketMQ 与 Kafka 的关键差异。Kafka 为每个 Partition 维护独立的日志文件，当 Topic 数量多时，写入变成随机 I/O，性能急剧下降。RocketMQ 的设计保证了无论多少个 Topic，写入始终是纯粹的顺序 I/O，单机可支持数万 Topic 的高吞吐写入。

每条消息在 CommitLog 中的存储格式包含以下核心字段：

写入 CommitLog 时支持两种刷盘策略：

• 同步刷盘（SYNC_FLUSH）：消息写入内存后立即刷盘，返回成功前等待磁盘写入完成。可靠性最高但性能有损耗，适用于金融级场景。
• 异步刷盘（ASYNC_FLUSH）：消息写入内存页（PageCache）即返回成功，由后台线程定期刷盘。性能高但宕机时可能丢失少量未刷盘消息，适用于吞吐优先的场景。

5.3 ConsumeQueue —— 消费索引

ConsumeQueue 是 CommitLog 的索引层，为每个 Topic 的每个 MessageQueue 维护一个独立的 ConsumeQueue 文件。它的作用是让消费者能快速定位到指定位置的消息，而不需要遍历整个 CommitLog。

每个 ConsumeQueue 条目固定 20 字节，包含三个字段：

消费者拉取消息时，先根据消费进度（逻辑 Offset）计算出 ConsumeQueue 中的条目位置，读取条目获取 CommitLog 物理偏移量和消息大小，再到 CommitLog 中读取完整消息。这个过程实现了从逻辑消费位点到物理消息的快速定位。

ConsumeQueue 文件本身也被设计为定长条目的数组结构，可以像数组一样通过下标随机访问，查找复杂度为 O(1)。

5.4 IndexFile —— 消息 Hash 索引

IndexFile 是基于 Hash 索引的消息查询文件，支持通过消息的 Key（业务唯一标识）快速查询消息。这是 RocketMQ 提供消息轨迹查询和消息回溯能力的基础。

一个 IndexFile 文件包含：

• Index Header（索引头）：40 字节，记录开始/结束时间戳、开始/结束 phyOffset、Hash 槽数量、索引条目数量等。
• Hash Slot Table（Hash 槽表）：默认 500 万个槽，每个槽 4 字节，存储该槽最后一个索引条目的序号。
• Index Linked List（索引链表）：默认 2000 万个条目，每个条目 20 字节，包含 Key Hash、phyOffset、timeDiff、prevIndex（用于 Hash 冲突时形成链表）。

通过消息 Key 查询时，先计算 Key 的 Hash 值定位到槽位，再通过链表遍历找到所有匹配的索引条目，最终从 CommitLog 读取消息内容。

5.5 消息存储文件目录结构

${storePathRootDir}/
├── commitlog/                 # CommitLog 文件目录
│   ├── 00000000000000000000   # 1GB 文件
│   ├── 00000000001073741824
│   └── ...
├── consumequeue/              # ConsumeQueue 文件目录
│   ├── TopicA/
│   │   ├── queue0/
│   │   │   └── 00000000000000000000
│   │   ├── queue1/
│   │   └── queue2/
│   └── TopicB/
│       └── queue0/
├── index/                     # IndexFile 文件目录
│   ├── 20260730000000
│   └── 20260730120000
├── config/                    # 配置元数据
│   ├── topics.json            # Topic 配置信息
│   ├── subscriptionGroup.json # 消费组配置信息
│   └── consumerOffset.json    # 消费进度信息
├── abort                      #  Broker 异常退出标记文件
└── checkpoint                 # 刷盘检查点文件

六、消息的生命周期

理解消息从生产到消费的完整流转过程，是掌握 RocketMQ 工作原理的关键。

6.1 消息生产流程

生产者启动时会从 NameServer 获取目标 Topic 的路由信息（包含哪些 Broker 上有哪些队列），并在本地缓存。发送消息时，默认采用轮询策略选择一个队列，将消息发送给该队列所属的 Broker。

三种发送方式的区别：

6.2 消息存储流程

消息到达 Broker 后的存储处理链路：

1. 接收与校验：Broker 网络层接收消息，校验 Topic 权限、消息体大小等。
2. 写入 CommitLog：消息追加到 CommitLog 文件末尾，根据刷盘策略决定同步或异步持久化。
3. 主从同步：如果配置了主从架构，Master 将消息同步到 Slave，根据同步策略决定是同步还是异步复制。
4. 构建 ConsumeQueue：ReputMessageService 后台线程扫描 CommitLog 新增的消息，为每条消息在对应的 ConsumeQueue 中追加索引条目。
5. 构建 IndexFile：如果消息带有 Key，异步构建 Hash 索引条目。

步骤 4 和 5 是异步进行的，与消息写入解耦，不影响写入性能。

6.3 消息消费流程

RocketMQ 的 Push 模型本质上仍然是 Pull——消费者后台线程以长轮询方式持续向 Broker 拉取消息。所谓"长轮询"，是指当 Broker 没有新消息时不会立即返回空结果，而是挂起请求等待新消息到达（默认 15 秒超时），这样既保证了实时性又避免了频繁空轮询。

七、消息类型详解

RocketMQ 支持五种消息类型，覆盖了绝大多数业务场景需求。每种消息类型都对应特定的使用场景和实现机制。

7.1 普通消息

最常见的消息类型，无特殊语义约束，生产者发送后消费者按消费进度顺序消费。适用于大多数不需要严格顺序或事务保证的场景，如日志记录、通知推送、数据同步等。

// 同步发送普通消息
Message msg = new Message("TopicTest", 
    "TagA",                          // 标签
    "OrderID_001",                   // 业务 Key
    ("Hello RocketMQ").getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg);

7.2 顺序消息

顺序消息保证同一业务 Key 的消息按发送顺序被消费。这在订单状态流转等场景中至关重要——"创建订单 → 支付 → 发货 → 确认收货"必须严格按序执行，否则业务逻辑会错乱。

RocketMQ 的顺序消息分为两个层次：

• 分区顺序（Partitioned FIFO）：同一队列内的消息严格有序。生产者通过 MessageQueueSelector 将同一业务 Key 的消息路由到同一队列，消费者单线程消费该队列。
• 全局顺序（Global FIFO）：整个 Topic 只有一个队列，所有消息严格有序。性能牺牲大，仅适用于极少数对顺序要求极高的场景。

graph LR
    subgraph 生产端 - 按业务Key路由
        M1[订单1001 创建]
        M2[订单1001 支付]
        M3[订单1001 发货]
        M4[订单1002 创建]
        M5[订单1002 支付]
        M6[订单1003 创建]
    end

    subgraph 路由策略
        SEL[MessageQueueSelector<br/>hash(orderId) % queueSize]
    end

    subgraph 队列分布
        Q0[Queue 0]
        Q1[Queue 1]
        Q2[Queue 2]
    end

    M1 --> SEL
    M2 --> SEL
    M3 --> SEL
    M4 --> SEL
    M5 --> SEL
    M6 --> SEL

    SEL -->|订单1001| Q0
    SEL -->|订单1002| Q1
    SEL -->|订单1003| Q2

    Q0 --> C0[Consumer 0<br/>单线程顺序消费<br/>创建→支付→发货]
    Q1 --> C1[Consumer 1<br/>单线程顺序消费<br/>创建→支付]
    Q2 --> C2[Consumer 2<br/>单线程顺序消费<br/>创建]

实现顺序消费需要生产端和消费端配合：生产端用 MessageQueueSelector 按业务 Key 做 Hash 路由，保证同一 Key 的消息进入同一队列；消费端使用 MessageListenerOrderly 接口，同一队列由同一消费线程串行处理。

7.3 延迟消息

延迟消息允许消息发送后延迟一段指定时间再被消费者消费。典型场景包括：下单 30 分钟未支付自动取消订单、任务延迟调度、延迟重试等。

RocketMQ 4.x 提供了 18 个固定延迟级别（1s/5s/10s/30s/1m/2m/3m/4m/5m/6m/7m/8m/9m/10m/20m/30m/1h/2h），5.x 支持任意精度的延迟时间（指定具体的延迟时间戳）。

延迟消息的实现原理：

延迟消息在 Broker 内部先被存储到一个特殊的内部 Topic（SCHEDULE_TOPIC_XXXX），对原始 Topic 的消费者不可见。Broker 的 ScheduleMessageService 按每个延迟级别定时扫描到期消息，将其恢复为原始 Topic 的消息重新写入，此时消费者才能拉取到。

7.4 事务消息

事务消息是 RocketMQ 的标志性特性之一，用于解决分布式系统中本地事务与消息发送的原子性问题。核心思想是通过两阶段提交加回查机制，保证"本地事务执行"和"消息发送"要么同时成功，要么同时失败。

核心场景：用户下单扣减库存后，需要发送消息通知积分服务发放积分。如果先扣库存再发消息，发消息失败则积分未发放；如果先发消息再扣库存，扣库存失败则多发了积分。事务消息解决了这个两难。

两阶段提交流程：

事务消息的关键设计在于半消息机制。生产者先发送一条"半消息"到 Broker，这条消息存储在内部 Topic 中，对消费者完全不可见。生产者收到 Broker 的确认后，开始执行本地事务。本地事务执行完毕后，生产者根据结果向 Broker 发送提交（COMMIT）或回滚（ROLLBACK）指令。

回查机制解决了一个边界问题：如果生产者执行完本地事务后、发送提交/回滚指令前宕机或网络中断，Broker 收不到确认指令。此时 Broker 会主动向生产者发起事务回查，询问该半消息对应的本地事务最终状态。生产者需要实现 TransactionListener 接口的 checkLocalTransaction 方法，通过查询本地数据库来判断事务的最终状态。

// 事务消息生产者示例
TransactionMQProducer producer = new TransactionMQProducer("tx_producer_group");
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务，返回事务状态
        try {
            doBusinessTransaction(); // 扣减库存等本地操作
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 事务回查：查询本地事务最终状态
        boolean committed = checkTransactionStatus(msg.getKeys());
        return committed ? LocalTransactionState.COMMIT_MESSAGE 
                         : LocalTransactionState.ROLLBACK_MESSAGE;
    }
});

// 发送事务消息
Message msg = new Message("TopicTx", "TagA", "OrderID_001", 
    "事务消息内容".getBytes());
TransactionSendResult result = producer.sendMessageInTransaction(msg, null);

注意：事务消息需要配合本地事务的幂等性设计。回查可能多次发生，checkLocalTransaction 方法必须能正确判断事务最终状态。同时消费者也需保证幂等消费，因为极端情况下消息可能被重复投递。

7.5 批量消息

批量消息允许在一次网络请求中发送多条消息，减少网络往返开销，显著提升吞吐量。适用于日志批量上报、数据同步等场景。

批量消息要求同一批消息必须具有相同的 Topic，且不支持延迟消息和事务消息的批量发送。消息总大小不能超过 4MB（默认配置）。

List<Message> messages = new ArrayList<>();
messages.add(new Message("TopicBatch", "TagA", "msg1".getBytes()));
messages.add(new Message("TopicBatch", "TagA", "msg2".getBytes()));
messages.add(new Message("TopicBatch", "TagA", "msg3".getBytes()));

// 批量发送
SendResult result = producer.send(messages);

八、消费者模型

8.1 集群消费与广播消费

RocketMQ 的消费者组有两种消费模式：

集群消费是默认模式。在集群模式下，同一消费者组的多个实例分担消息消费任务，消费进度由 Broker 统一管理，实现消费能力的水平扩展。当某个消费者实例宕机，其负责的队列会被重新分配给组内其他实例。

广播模式下，每个消费者实例独立消费 Topic 的全量消息，消费进度保存在本地。适合所有节点都需要接收同一份消息的场景，如本地缓存更新。

8.2 负载均衡机制

RocketMQ 5.x 提供两种负载均衡策略：

队列级负载均衡（传统模式）

消费者组内的消费者实例按照一致的分配算法（如一致性 Hash、平均分配等）将 Topic 的队列分配给各个实例。每个实例只消费分配到自己的队列。

问题在于：当消费者实例数与队列数不整除时，负载不均衡。例如 3 个队列分配给 2 个消费者，其中一个消费者要承担 2/3 的负载。更极端的是，当消费者数量超过队列数量时，多余的消费者将完全空闲。

消息粒度负载均衡（5.0 新增）

打破队列绑定关系，消费者不再固定消费某个队列，而是从所有队列中随机拉取消息。消费进度以消息为粒度记录，而非队列级别。这种方式实现了真正的负载均衡——无论消费者数量多少，消息都能均匀分配到每个实例。同时，消费者数量与 Broker 的队列数量完全解耦，非常适合 Serverless 架构下消费者实例频繁弹性伸缩的场景。

graph TB
    subgraph 队列级负载均衡（传统）
        Q1[Queue 0] --> CA[Consumer A]
        Q2[Queue 1] --> CA
        Q3[Queue 2] --> CB[Consumer B]
        Q4[Queue 3] --> CB
        CC1[Consumer C<br/>空闲!]
        Note: 消费者数 > 队列数时，多余消费者空闲
    end

    subgraph 消息粒度负载均衡（5.0+）
        Q5[Queue 0] --> MIX
        Q6[Queue 1] --> MIX
        Q7[Queue 2] --> MIX
        Q8[Queue 3] --> MIX
        MIX[消息随机分配]
        MIX --> CC2[Consumer A<br/>均衡分配]
        MIX --> CC3[Consumer B<br/>均衡分配]
        MIX --> CC4[Consumer C<br/>均衡分配]
        Note: 消费者数量与队列数量解耦
    end

8.3 消费进度管理

消费进度（Consumer Offset）记录了消费者组在每个队列上已消费到的位置，是实现"至少一次"（At-Least-Once）语义的基础。

集群模式下，消费进度存储在 Broker 端的 consumerOffset.json 文件中。消费者每次成功消费消息后，异步向 Broker 提交新的 Offset。Broker 以队列为粒度记录每个消费者组的消费位点。

消费进度支持消息回溯——可以将消费位点重置到历史某个位置，重新消费之前的消息。这在以下场景中很有用：消费逻辑有 Bug 导致错误处理了一批消息，修复后需要重新消费。

8.4 Push 与 Simple 消费模型

RocketMQ 5.x 提供两种消费 API 模型：

• PushConsumer：面向大多数业务场景。内部通过长轮询机制自动拉取消息并回调消费监听器，开发者只需实现 MessageListener 接口。使用简单，自带限流和重试机制。
• SimpleConsumer：面向需要精细控制的场景。开发者主动调用 receive() 拉取消息，处理完成后调用 ack() 确认。灵活性高，可自行控制拉取速率和并发度，适合与 LLM Batch API 等批量推理场景配合。

九、高级特性

9.1 消息过滤

RocketMQ 支持在服务端对消息进行过滤，只将消费者感兴趣的消息投递过去，减少网络传输和客户端处理压力。过滤方式有两种：

Tag 过滤

生产者在发送消息时设置 Tag，消费者订阅时指定感兴趣的 Tag 列表。Broker 端通过比对 ConsumeQueue 中的 Tags Hash Code 进行快速过滤。

// 生产者：设置 Tag
Message msg = new Message("TopicTest", "TagA", "body".getBytes());

// 消费者：订阅指定 Tag
consumer.subscribe("TopicTest", "TagA || TagB");

SQL92 表达式过滤

支持使用 SQL92 语法子集对消息属性进行复杂条件过滤，比 Tag 过滤更灵活。需要在消息 Properties 中设置自定义属性，在订阅时编写 SQL 表达式。

// 生产者：设置自定义属性
Message msg = new Message("TopicTest", "TagA", "body".getBytes());
msg.putUserProperty("age", "25");
msg.putUserProperty("region", "hangzhou");

// 消费者：SQL92 表达式过滤
consumer.subscribe("TopicTest", 
    MessageSelector.bySql("age > 18 AND region = 'hangzhou'"));

9.2 重试与死信队列

当消息消费失败时，RocketMQ 会自动将消息放入重试队列进行重试，而不是直接丢弃。重试机制为瞬时故障（如下游服务短暂不可用）提供了自动恢复能力。

重试策略：

• 消费失败后，消息被投递到重试队列（%RETRY%消费者组名）
• 重试采用递增延迟策略（如 10s、30s、1m、2m、3m…），最多重试 16 次
• 每次重试的延迟时间逐步增长，避免短时间内反复失败

死信队列（DLQ）：

当消息重试达到最大次数（默认 16 次）仍然失败，消息会被转移到死信队列（%DLQ%消费者组名）。死信队列中的消息不会再自动投递，需要人工介入处理——排查故障原因后，可通过控制台或命令行工具将死信消息重新发送到原 Topic。

正常消费 → 失败 → 重试队列（递增延迟，最多16次）→ 仍然失败 → 死信队列（人工处理）

9.3 消息轨迹

RocketMQ 内置了全链路消息轨迹追踪能力。开启后，每条消息从生产、存储到消费的完整路径都会被记录到专门的轨迹 Topic（RMQ_SYS_TRACE_TOPIC）中，包含：

• 消息发送时间、发送者 IP、发送耗时
• 消息存储时间、存储 Broker、是否成功
• 消息消费时间、消费者 IP、消费耗时、消费结果

消息轨迹对于问题排查（消息为什么没消费？为什么消费慢？）和消息审计非常有价值。Kafka 不提供内置的轨迹能力，需要借助第三方插件。

9.4 消息重试与幂等性

RocketMQ 保证消息至少被消费一次（At-Least-Once），这意味着在网络抖动或消费者宕机重启的场景下，消息可能被重复投递。因此，消费者必须实现幂等性处理——即多次消费同一条消息的结果与消费一次相同。

常见的幂等性实现方案：

• 业务唯一键去重：利用消息的 Key 或业务唯一标识，在数据库或 Redis 中记录已处理的 Key，消费前检查是否已处理。
• 乐观锁/版本号：在数据库更新操作中带上版本号，重复消费时版本号不匹配则跳过。
• 状态机校验：业务流程有状态流转时，消费前校验当前状态是否允许该操作。

十、高可用与容灾

10.1 主从复制

RocketMQ 的 Broker 支持主从架构，每个 Broker 组包含一个 Master 和一个或多个 Slave。Master 负责读写，Slave 负责只读和数据备份。

主从同步有两种模式：

在 4.x 中，当 Master 宕机时，Slave 可以继续提供读服务（消费消息），但无法提供写服务（生产消息）。要恢复写服务需要人工干预或借助 DLedger。

10.2 DLedger —— 基于 Raft 的高可用方案

DLedger 是 RocketMQ 社区开发的基于 Raft 协议的共识组件，为 Broker 提供自动 Leader 选举和日志复制能力。引入 DLedger 后，当 Master 节点宕机时，集群能自动选举新的 Master，实现写服务的自动恢复。

DLedger 的工作原理：

• 至少需要 3 个节点组成 DLedger Group（Raft 需要多数派）
• 消息写入 Leader 后，同步复制到过半数的 Follower 才算成功
• Leader 宕机后，剩余节点通过 Raft 选举协议选出新 Leader，整个过程自动完成，通常在数秒内恢复

10.3 Controller —— 5.x 自动主从切换

RocketMQ 5.x 引入了 Controller 组件，作为更轻量的主从切换方案。与 DLedger 不同，Controller 不修改数据复制方式（仍然使用原有的主从同步），而是在 Master 宕机时，由 Controller 主动将 Slave 提升为 Master。

Controller 自身也是一个集群（基于 Raft 或第三方协调服务），保证自身高可用。它的工作流程：

1. Controller 持续监控所有 Broker 组的 Master 状态
2. 当检测到 Master 不可用时，选择一个数据最完整的 Slave
3. 将该 Slave 的角色提升为 Master，更新 NameServer 中的路由信息
4. 客户端从 NameServer 获取最新路由，继续向新 Master 读写

相比 DLedger，Controller 方案的优势在于不需要 3 个节点（2 个即可，Slave 平时只做备份），成本更低。缺点是在切换瞬间可能有短暂不可用。

10.4 高可用场景总结

graph TB
    subgraph NameServer 集群（AP模式）
        NS1[NS 1]
        NS2[NS 2]
        NS3[NS 3]
    end

    subgraph Broker Group A
        BA_M[Master A<br/>读写]
        BA_S[Slave A<br/>只读/备份]
    end

    subgraph Broker Group B
        BB_M[Master B<br/>读写]
        BB_S[Slave B<br/>只读/备份]
    end

    BA_M -->|注册| NS1
    BB_M -->|注册| NS2

    subgraph 故障处理
        F1[NameServer 宕机] --> R1[其他 NS 节点正常服务<br/>客户端自动切换]
        F2[Master 宕机] --> R2{切换方案}
        R2 -->|4.x + DLedger| R2A[Raft 自动选举新 Master]
        R2 -->|5.x + Controller| R2B[Controller 提升Slave为Master]
        R2 -->|4.x 无DLedger| R2C[Slave 继续提供读服务<br/>写服务需人工恢复]
        F3[Slave 宕机] --> R3[Master 正常服务<br/>数据安全性降低]
    end

十一、快速上手实战

11.1 环境准备与安装

以 RocketMQ 5.x 为例，最快捷的本地启动方式是使用 Docker：

# 创建 docker-compose.yml
cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
  namesrv:
    image: apache/rocketmq:5.3.1
    container_name: rmqnamesrv
    ports:
      - 9876:9876
    command: sh mqnamesrv

  broker:
    image: apache/rocketmq:5.3.1
    container_name: rmqbroker
    ports:
      - 10909:10909
      - 10911:10911
    depends_on:
      - namesrv
    environment:
      NAMESRV_ADDR: "namesrv:9876"
    command: sh mqbroker -c ../conf/broker.conf
EOF

# 启动集群
docker-compose up -d

启动后，可以通过 Dashboard 控制台查看集群状态：

# 启动 RocketMQ Dashboard（Web 管理界面）
docker run -d --name rmqdashboard \
  -p 8080:8080 \
  -e "rocketmq.config.namesrvAddr=host.docker.internal:9876" \
  apacherocketmq/rocketmq-dashboard:latest

11.2 命令行快速验证

# 进入 Broker 容器
docker exec -it rmqbroker bash

# 创建 Topic
sh mqadmin updateTopic -n localhost:9876 \
  -c DefaultCluster -t TopicTest

# 发送消息
sh mqadmin sendMessage -n localhost:9876 \
  -t TopicTest -p "Hello RocketMQ"

# 消费消息
sh mqadmin consumeMessage -n localhost:9876 \
  -t TopicTest

11.3 Java 生产者示例

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        // 1. 创建生产者实例，指定消费者组名
        DefaultMQProducer producer = new DefaultMQProducer("producer_group_demo");
        // 2. 指定 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 3. 启动生产者
        producer.start();

        for (int i = 0; i < 10; i++) {
            // 4. 创建消息，指定 Topic、Tag、Key 和消息体
            Message msg = new Message(
                "TopicTest",                           // Topic
                "TagA",                                 // Tag
                "OrderKey_" + i,                        // Key
                ("Hello RocketMQ " + i).getBytes()      // Body
            );
            // 5. 同步发送消息
            SendResult result = producer.send(msg);
            System.out.printf("发送结果: %s%n", result);
        }

        // 6. 关闭生产者
        producer.shutdown();
    }
}

11.4 Java 消费者示例

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

public class PushConsumer {
    public static void main(String[] args) throws Exception {
        // 1. 创建消费者实例，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group_demo");
        // 2. 指定 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 3. 设置从最早的消息开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 4. 订阅 Topic，过滤 Tag（"*" 表示接收所有 Tag）
        consumer.subscribe("TopicTest", "*");
        // 5. 注册消费监听器
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                System.out.printf("收到消息: Topic=%s, Tag=%s, Key=%s, Body=%s%n",
                    msg.getTopic(),
                    msg.getTags(),
                    msg.getKeys(),
                    new String(msg.getBody())
                );
            }
            // 返回消费状态：SUCCESS 表示消费成功，RECONSUME_LATER 表示消费失败需重试
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 6. 启动消费者
        consumer.start();
        System.out.println("消费者已启动");
    }
}

11.5 Spring Boot 集成

在实际项目中，通常使用 Spring Boot starter 简化集成：

<!-- pom.xml 依赖 -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.3.1</version>
</dependency>

# application.yml 配置
rocketmq:
  name-server: localhost:9876
  producer:
    group: spring_boot_producer_group
    send-message-timeout: 3000

// 生产者：使用注解方式
@Service
public class OrderMessageService {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    public void sendOrderMessage(Order order) {
        // 同步发送
        rocketMQTemplate.syncSend("order-topic:tagA", order);

        // 异步发送
        rocketMQTemplate.asyncSend("order-topic", order, new SendCallback() {
            @Override
            public void onSuccess(SendResult result) { /* 成功处理 */ }
            @Override
            public void onException(Throwable e) { /* 失败处理 */ }
        });

        // 发送顺序消息（按 orderId 哈希路由到同一队列）
        rocketMQTemplate.syncSendOrderly("order-topic", order, 
            String.valueOf(order.getId()));
    }
}

// 消费者：使用注解方式
@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "order_consumer_group",
    selectorExpression = "tagA || tagB"
)
@Component
public class OrderMessageConsumer implements RocketMQListener<Order> {
    @Override
    public void onMessage(Order order) {
        System.out.println("处理订单消息: " + order);
        // 业务处理逻辑
    }
}

十二、与 Kafka、RabbitMQ 对比选型

Kafka、RabbitMQ 和 RocketMQ 是目前最主流的三款消息中间件，它们在设计哲学和适用场景上有明显差异。

12.1 核心设计哲学对比

12.2 功能特性对比

12.3 性能特征对比

12.4 选型建议

三者的选择本质上是对不同设计取舍的权衡：

• 选 RocketMQ：需要事务消息、延迟消息、消息过滤等丰富业务特性；金融级可靠性要求高；Topic 数量大（万级）；需要内置消息轨迹和死信队列等运维能力。
• 选 Kafka：追求极致吞吐量（日志流、大数据管道）；与大数据生态（Flink、Spark、Connect）深度集成；Topic 数量可控（百级以内）。
• 选 RabbitMQ：需要复杂的消息路由（Topic Exchange、Header Exchange）；延迟要求极低（微秒级）；消息量不大但路由逻辑复杂；已有 AMQP 生态积累。

十三、最佳实践与避坑指南

13.1 Topic 与队列规划

• Topic 数量：RocketMQ 单机可支持数万 Topic，但仍建议按业务域合理划分，避免无序膨胀。一般一个业务场景对应一个 Topic，用 Tag 做二级分类。
• 队列数量：默认 4 个队列。队列数决定了该 Topic 的最大消费并行度——集群消费模式下，消费者实例数不应超过队列数（传统负载均衡模式）。需要更高并发时可增加队列数，但不宜过多（每个队列都会占用 Broker 内存和存储资源）。
• 队列扩容：新增队列只对新消息生效，已有消息仍在原队列中。不要在生产高峰期做队列扩缩容操作。

13.2 消息发送最佳实践

• 发送超时设置：根据业务 SLA 合理设置 sendMsgTimeout（默认 3000ms），过长会拖垮调用方，过短可能导致正常消息发送失败。
• 重试策略：同步发送默认重试 2 次（共 3 次）。对于非幂等业务，需关闭重试避免重复发送。
• 消息大小：单条消息建议不超过 4MB。大消息应考虑拆分或使用 OSS 存储后只传引用。
• 异步发送异常处理：异步发送的 SendCallback 中必须处理异常，否则失败会静默丢失。

13.3 消费端最佳实践

• 幂等性：务必实现消费幂等。推荐使用消息 Key + Redis/数据库 做去重，或利用业务状态机保证幂等。
• 消费耗时控制：单条消息消费时间不宜过长（建议不超过 30 秒），否则会触发消费超时导致消息重复投递。耗时操作应异步化处理。
• 批量拉取优化：通过 pullBatchSize 参数控制单次拉取消息数，在吞吐和延迟间取得平衡。
• 消费失败处理：对于预期内失败（如下游服务维护），可以返回 RECONSUME_LATER 让消息自动重试。对于永久性失败（如消息格式错误），建议记录日志并返回成功，避免消息堆积在重试队列。

13.4 生产环境部署建议

• NameServer 部署：至少 2 个节点，分布在不同物理机/机房。NameServer 是无状态的，可以随时增减。
• Broker 部署：生产环境建议至少 2 组 Broker（每组一主一从），分布在不同机房实现跨机房容灾。
• 刷盘策略：金融场景使用同步刷盘 + 同步复制；普通业务场景使用异步刷盘 + 异步复制以提升吞吐。
• 监控告警：重点监控 Broker 堆积量、消费 TPS、消费延迟、死信队列消息数等指标。RocketMQ Dashboard 提供了 Web 可视化界面。
• 资源规划：CommitLog 默认保留 72 小时（fileReservedTime），根据日均消息量预估磁盘容量。建议磁盘使用率告警阈值设为 75%。

13.5 常见问题排查

总结

RocketMQ 从诞生之初就面向超大规模电商和金融场景，在可靠性、功能丰富度和运维友好性方面做了大量工程化设计。它的核心价值在于：

1. 金融级可靠性：事务消息、同步刷盘、主从同步、DLedger/Controller 自动切换等多重保障
2. 丰富的消息语义：顺序、延迟、事务、批量消息一站式覆盖
3. 优秀的 Topic 扩展性：CommitLog 共享存储设计让单机支持万级 Topic
4. 5.x 云原生演进：存算分离、gRPC 多语言 SDK、消息粒度负载均衡，适应云原生和 AI 时代需求

从零开始学习 RocketMQ，建议的路径是：先理解领域模型（Topic/Queue/Producer/Consumer）→ 跑通本地 Demo → 逐个实践消息类型 → 深入存储引擎原理 → 研究高可用机制 → 在实际项目中应用最佳实践。每一步都配合代码实验和 Dashboard 观察，理论和实践交替验证，才能真正建立完整的认知体系。

Apache RocketMQ 官方文档是学习和参考的最佳来源，地址：https://rocketmq.apache.org/docs/