Appearance
RocketMQ 详解：分布式消息队列
RocketMQ 是阿里巴巴开源的分布式消息中间件，经过历年双十一的考验，具备低延迟、高吞吐、高可靠、海量消息堆积等特性。在 Spring Cloud Alibaba 生态中，RocketMQ 承担着异步解耦、流量削峰、最终一致性等关键职责。
一、为什么需要消息队列？
在单体架构中，所有业务逻辑都在一个进程内完成，方法调用是同步的、即时的。但当系统演进到分布式架构后，服务之间需要通过网络通信，同步调用会带来严重的耦合问题和性能瓶颈。消息队列正是为了解决这些问题而生的。
1.1 三大核心价值
| 价值 | 核心思路 | 一句话 |
|---|---|---|
| 异步解耦 | 非关键流程异步化，不阻塞主流程 | "不重要的先记下来，回头慢慢做" |
| 流量削峰 | 瞬时高并发请求先入队，后端平滑消费 | "洪水来了先蓄水，下游慢慢放" |
| 最终一致性 | 事务消息保证分布式事务的最终一致性 | "先记个账，办成了通知你，失败了回滚" |
1.2 场景一：异步处理
场景：用户下单后，需要发送短信通知、推送 App 消息、增加积分。如果同步处理，整个流程耗时 = 下单 + 发短信 + 推送 + 加积分，用户体验极差。
没有消息队列时（同步调用）：
用户下单
│
├── 1. 创建订单（100ms）
├── 2. 发送短信通知（200ms） ← 用户干等着
├── 3. 推送 App 消息（150ms） ← 用户干等着
└── 4. 增加积分（100ms） ← 用户干等着
─────
总耗时：550ms，用户体验差有了消息队列后（异步解耦）：
用户下单
│
└── 1. 创建订单（100ms）→ 发送消息到 MQ → 返回"下单成功"（100ms）
│
├── 短信服务消费消息 → 发送短信（200ms）
├── 推送服务消费消息 → 推送消息（150ms）
└── 积分服务消费消息 → 增加积分（100ms）
用户感知耗时：100ms，其余操作异步完成，不阻塞主流程1.3 场景二：流量削峰
场景：秒杀活动，瞬时 10 万 QPS，后端数据库只能扛 3000 QPS。如果不做削峰，数据库瞬间崩溃，所有请求都失败。
没有消息队列时：
秒杀开始
100000 QPS ──────────────► 数据库（3000 QPS）
│
└── 崩溃！所有请求失败有了消息队列后：
秒杀开始
100000 QPS ──► [ RocketMQ 队列 ] ──► 消费者（3000 QPS）
│ │
│ 消息堆积在队列中， │ 消费者按自己的节奏消费
│ 不丢失，慢慢消费 │ 数据库压力可控
│ │
└── 70,000 条消息 └── 每秒处理 3000 条
排队等待消费 约 23 秒处理完1.4 场景三：系统解耦
场景：订单服务创建订单后，需要通知库存服务扣减库存。如果通过 RPC 同步调用，订单服务必须知道库存服务的地址，且库存服务挂了会直接影响下单。
没有消息队列时（强耦合）：
订单服务 ──RPC 调用──► 库存服务
│ │
│ 库存服务挂了？ │ 需要硬编码库存服务地址
│ 扣库存失败？ │ 新增下游服务需要改订单服务代码
│ 全部影响下单！ │有了消息队列后（异步解耦）：
订单服务 ──发送消息──► [ RocketMQ Topic: ORDER_CREATED ]
│
├── 库存服务消费 → 扣减库存
├── 物流服务消费 → 创建运单
└── 风控服务消费 → 风险评估
订单服务只管发消息，不需要知道谁在消费
新增下游消费者，只需订阅 Topic，订单服务无需任何改动二、核心概念
理解 RocketMQ 的核心概念是正确使用它的前提。下面逐一介绍每个核心角色。
2.1 核心角色一览
| 角色 | 英文 | 职责 | 类比 |
|---|---|---|---|
| 生产者 | Producer | 发送消息 | 快递发件人 |
| 消费者 | Consumer | 接收并处理消息 | 快递收件人 |
| 消息代理 | Broker | 存储和转发消息 | 快递中转站 |
| 路由注册 | NameServer | 维护 Broker 路由信息 | 快递网点查询系统 |
| 主题 | Topic | 消息的逻辑分类 | 快递类型（文件/包裹/冷链） |
| 标签 | Tag | Topic 下的二级分类 | 快递子类型（加急/普通） |
| 队列 | Queue | 消息的物理分区 | 快递分拣传送带 |
| 消费者组 | ConsumerGroup | 逻辑上相同的消费者集合 | 同一个收件部门的成员 |
2.2 各概念详解
Producer（生产者）
生产者负责生产消息，将消息发送到 Broker。生产者不需要知道消费者是谁，只管把消息发到指定的 Topic。RocketMQ 支持三种发送方式：
- 同步发送：发送后等待 Broker 确认，可靠性最高，但有延迟
- 异步发送：发送后立即返回，通过回调处理结果，吞吐量高
- 单向发送：发送后不关心结果，性能最高，但可能丢失消息
Consumer（消费者）
消费者从 Broker 拉取消息并进行业务处理。RocketMQ 支持两种消费模式（详见第五章）：
- 集群消费：同一条消息只被 ConsumerGroup 中的一个消费者消费
- 广播消费：同一条消息被 ConsumerGroup 中的每一个消费者消费
Broker（消息代理）
Broker 是 RocketMQ 的核心，负责消息的存储、转发和投递。一个 Broker 可以包含多个 Topic，每个 Topic 又分为多个 Queue。Broker 通常以主从模式部署，主节点负责读写，从节点负责备份。
NameServer（路由注册）
NameServer 是 RocketMQ 的路由注册中心，维护着 Broker 的地址和 Topic 路由信息。NameServer 之间互不通信，每个 NameServer 节点都独立维护完整的路由信息。Producer 和 Consumer 从 NameServer 获取路由信息后，直接与 Broker 通信。
Topic（主题）
Topic 是消息的逻辑分类，类似于数据库中的"表"。生产者和消费者通过 Topic 进行关联：生产者往某个 Topic 发消息，消费者从某个 Topic 收消息。一个 Topic 可以跨越多个 Broker。
Tag（标签）
Tag 是 Topic 下的二级分类，用于消息过滤。比如 Topic 是"订单"，Tag 可以是"订单创建"、"订单支付"、"订单取消"。消费者可以订阅 Topic 下的某些 Tag，只消费关心的消息。
Queue（队列）
Queue 是消息的物理分区，RocketMQ 中一个 Topic 对应多个 Queue，Queue 是消息存储和消费的最小单元。Queue 数量在创建 Topic 时指定，一般建议与消费者数量匹配。每个 Queue 中的消息严格有序。
ConsumerGroup（消费者组）
ConsumerGroup 是一组逻辑上相同的消费者，它们共同消费同一个 Topic。在集群消费模式下，同一 ConsumerGroup 内的消费者分摊 Queue：一个 Queue 只能被组内一个消费者消费，但一个消费者可以消费多个 Queue。
2.3 概念关系图
┌─────────────────────────────────────────────────────────────────────┐
│ NameServer 集群 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ NS-1 │ │ NS-2 │ │ NS-3 │ │
│ │ (独立) │ │ (独立) │ │ (独立) │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ 注册路由信息 注册路由信息 注册路由信息 │
└──────────────┼───────────────┼───────────────┼───────────────────────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────────┐
│ Broker 集群 │
│ │
│ ┌──────────────────────┐ ┌──────────────────────┐ │
│ │ Broker-A (Master) │ ◄──► │ Broker-A (Slave) │ │
│ │ │ 主从复制 │ │ │
│ │ Topic: Order │ │ Topic: Order │ │
│ │ ┌────┐ ┌────┐ ┌────┐ │ │ (只读备份) │ │
│ │ │ Q0 │ │ Q1 │ │ Q2 │ │ │ │ │
│ │ └────┘ └────┘ └────┘ │ │ │ │
│ └──────────────────────┘ └──────────────────────┘ │
│ │
│ ┌──────────────────────┐ ┌──────────────────────┐ │
│ │ Broker-B (Master) │ ◄──► │ Broker-B (Slave) │ │
│ │ Topic: Order │ 主从复制 │ Topic: Order │ │
│ │ ┌────┐ ┌────┐ ┌────┐ │ │ (只读备份) │ │
│ │ │ Q3 │ │ Q4 │ │ Q5 │ │ │ │ │
│ │ └────┘ └────┘ └────┘ │ │ │ │
│ └──────────────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
▲ ▲
│ 发送消息 │ 拉取消息
│ │
┌──────────────┴───────────┐ ┌───────────────┴──────────┐
│ Producer 集群 │ │ ConsumerGroup 集群 │
│ │ │ │
│ ┌────────┐ ┌────────┐ │ │ ┌────────┐ ┌────────┐ │
│ │ P-1 │ │ P-2 │ │ │ │ C-1 │ │ C-2 │ │
│ └────────┘ └────────┘ │ │ │ (Q0,Q1)│ │ (Q2,Q3)│ │
│ │ │ └────────┘ └────────┘ │
│ 发送到 Topic: Order │ │ 消费 Topic: Order │
│ 可按 Tag: pay/create │ │ Tag: pay (只消费支付消息) │
└────────────────────────────┘ └──────────────────────────┘三、核心架构
3.1 整体架构
┌──────────────────────────────────────────────────────────────────────┐
│ NameServer 集群 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ NS-1 │ │ NS-2 │ │ NS-3 │ ← 无状态，互不通信 │
│ │ │ │ │ │ │ │
│ │ RouteInfo│ │ RouteInfo│ │ RouteInfo│ ← 每台独立维护全量路由 │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ │ Broker 向所有 NameServer 注册（每 30s 心跳） │
│ │ │ │ │
└────────┼───────────────┼───────────────┼──────────────────────────────┘
│ │ │
▼ ▼ ▼
┌──────────────────────────────────────────────────────────────────────┐
│ Broker 集群 │
│ │
│ ┌──────────────────────┐ ┌──────────────────────┐ │
│ │ Broker-Master-A │ │ Broker-Master-B │ │
│ │ │ │ │ │
│ │ Topic: Order │ │ Topic: Order │ │
│ │ Queue: 0, 1, 2, 3 │ │ Queue: 4, 5, 6, 7 │ │
│ │ (读写) │ │ (读写) │ │
│ └──────────┬───────────┘ └──────────┬───────────┘ │
│ │ 主从同步 │ 主从同步 │
│ ▼ ▼ │
│ ┌──────────────────────┐ ┌──────────────────────┐ │
│ │ Broker-Slave-A │ │ Broker-Slave-B │ │
│ │ (只读备份) │ │ (只读备份) │ │
│ └──────────────────────┘ └──────────────────────┘ │
└──────────────────────────────────────────────────────────────────────┘
▲ ▲
│ 发送消息 │ 拉取消息
│ │
┌────────┴────────┐ ┌────────┴────────┐
│ Producer 集群 │ │ Consumer 集群 │
│ │ │ │
│ ┌────┐ ┌────┐ │ │ ┌────┐ ┌────┐ │
│ │ P1 │ │ P2 │ │ │ │ C1 │ │ C2 │ │
│ └────┘ └────┘ │ │ └────┘ └────┘ │
└──────────────────┘ └──────────────────┘3.2 消息路由流程
Producer 发送消息的完整流程：
1. Producer 启动，从 NameServer 拉取 Topic 路由信息
┌──────────┐ ① 获取路由信息 ┌──────────┐
│ Producer │ ──────────────────────────► │NameServer│
└──────────┘ ◄────────────────────────── │ │
返回：Broker 列表 + Queue 列表 └──────────┘
2. Producer 根据路由信息，选择 Queue 发送消息
┌──────────┐ ② 直接发送消息 ┌──────────┐
│ Producer │ ──────────────────────────► │ Broker │
│ │ │ (Master)│
│ 轮询选择 │ │ Queue-0 │
│ Queue │ │ Queue-1 │
└──────────┘ └──────────┘
3. 路由信息定期更新（每 30s），感知 Broker 上下线
┌──────────┐ ③ 定时更新路由 ┌──────────┐
│ Producer │ ──────────────────────────► │NameServer│
└──────────┘ └──────────┘
Consumer 消费消息的流程：
1. Consumer 启动，从 NameServer 获取路由信息
2. Consumer 根据负载均衡策略，分配 Queue
3. Consumer 从分配的 Queue 拉取消息（Pull 模式）
4. 消息处理完成后返回消费结果
5. 路由信息定期更新（每 30s），感知 Queue 变化3.3 为什么 NameServer 不用 Zookeeper？
很多分布式系统选择 Zookeeper 做注册中心，RocketMQ 却自己实现了 NameServer。这不是重复造轮子，而是有明确的技术考量。
| 对比维度 | Zookeeper | NameServer |
|---|---|---|
| CAP 选择 | CP（强一致性） | AP（高可用性） |
| 一致性协议 | ZAB（Paxos 变种） | 无，各节点独立 |
| 节点通信 | 节点间心跳 + 选举 | 互不通信，完全对等 |
| 故障影响 | Leader 选举期间不可用（数十秒） | 单节点故障无影响，其他节点继续服务 |
| 运维复杂度 | 需要关注 Leader 选举、脑裂、数据一致性 | 极其简单，无状态，随意扩缩容 |
| 数据存储 | 持久化到磁盘 | 全内存，Broker 重启后重新注册 |
| 适用场景 | 需要强一致性的配置管理 | 轻量级路由发现，可容忍短暂不一致 |
RocketMQ 选择 NameServer 的核心原因：
- 轻量级：NameServer 只做路由发现，不涉及复杂的选举和一致性协议，代码量少，维护成本低
- 高可用（AP）：对于消息队列来说，NameServer 不可用比路由信息不一致更致命。NameServer 集群中任意节点挂了，其他节点继续服务，不会像 Zookeeper 那样在 Leader 选举期间不可用
- 最终一致性可接受：Broker 上下线不是高频事件，30s 内路由信息不一致是完全可以接受的。即使路由信息暂时不一致，Producer 也有重试机制兜底
- 运维简单：不需要维护 Zookeeper 集群，不需要关注选举、脑裂、数据一致性等复杂问题
四、消息类型
RocketMQ 支持四种消息类型，覆盖了从普通异步通知到分布式事务的完整场景需求。
4.1 普通消息
普通消息是最基础的消息类型，支持三种发送方式，适用于绝大部分异步解耦场景。
4.1.1 同步发送
发送方发送消息后，同步等待 Broker 返回结果。可靠性最高，但发送方会阻塞等待。
java
public class SyncProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
Message msg = new Message(
"TopicTest", // Topic
"TagA", // Tag
"Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET)
);
SendResult result = producer.send(msg);
System.out.printf("发送结果: %s%n", result);
producer.shutdown();
}
}4.1.2 异步发送
发送方发送消息后立即返回，通过回调处理发送结果。吞吐量高，适用于对延迟敏感的场景。
java
public class AsyncProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
producer.setRetryTimesWhenSendAsyncFailed(0);
Message msg = new Message("TopicTest", "TagA",
"Hello Async RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
producer.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.printf("发送成功: %s%n", sendResult);
}
@Override
public void onException(Throwable e) {
System.out.printf("发送失败: %s%n", e.getMessage());
}
});
// 异步发送，需要等待回调执行
Thread.sleep(3000);
producer.shutdown();
}
}4.1.3 单向发送
发送方发送消息后不关心发送结果，性能最高，但不保证消息到达。适用于日志采集等可容忍丢失的场景。
java
public class OnewayProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
Message msg = new Message("TopicTest", "TagA",
"Hello Oneway RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
producer.sendOneway(msg); // 不关心结果，不等待，不重试
producer.shutdown();
}
}三种发送方式对比：
| 发送方式 | 可靠性 | 吞吐量 | 延迟 | 适用场景 |
|---|---|---|---|---|
| 同步发送 | 高 | 中 | 高（等待 Broker 确认） | 重要业务通知，如订单状态变更 |
| 异步发送 | 中 | 高 | 低（立即返回） | 对延迟敏感、可接受少量失败 |
| 单向发送 | 低 | 最高 | 最低 | 日志采集、埋点数据 |
4.2 顺序消息
顺序消息保证消息按照发送顺序被消费。RocketMQ 支持两种顺序级别：
全局顺序 vs 分区顺序：
| 维度 | 全局顺序 | 分区顺序 |
|---|---|---|
| 实现方式 | 一个 Topic 只有一个 Queue | 一个 Topic 有多个 Queue，同类消息进同一 Queue |
| 并发能力 | 低（串行消费） | 高（不同 Queue 可并行消费） |
| 实现复杂度 | 低 | 中（需要 MessageQueueSelector） |
| 适用场景 | 对全局顺序有强要求的场景 | 只需要局部有序的场景（如单个订单的操作有序） |
分区顺序消息示例：
java
// ========== 生产者：按订单 ID 路由到同一 Queue ==========
public class OrderedProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 100; i++) {
int orderId = i % 10; // 10 个订单
Message msg = new Message(
"TopicOrdered",
tags[i % tags.length],
"KEY" + i,
("Ordered Message " + i).getBytes()
);
SendResult result = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
// 同一个 orderId 的消息总是路由到同一个 Queue
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);
System.out.printf("发送结果: %s%n", result);
}
producer.shutdown();
}
}java
// ========== 消费者：顺序消费 ==========
public class OrderedConsumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.subscribe("TopicOrdered", "TagA || TagB || TagC");
// 注册顺序消费监听器
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeOrderlyContext context) {
// 自动上锁，保证同一 Queue 的消息在同一个线程中顺序消费
context.setAutoCommit(true);
for (MessageExt msg : msgs) {
System.out.printf("线程: %s, Queue: %s, 内容: %s%n",
Thread.currentThread().getName(),
msg.getQueueId(),
new String(msg.getBody()));
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.println("顺序消费者启动...");
}
}顺序消费的关键点：
MessageListenerOrderly会为每个 ConsumerQueue 加锁，确保同一时刻只有一个线程消费该 Queue- 生产端使用
MessageQueueSelector将需要顺序的消息路由到同一个 Queue - 顺序消费会牺牲一定的并发性能，非必要不开启
4.3 延迟消息
延迟消息是指消息发送后，消费者不会立即收到，而是等待指定的延迟时间后才会被消费。适用于超时关单、定时提醒等场景。
RocketMQ 支持的 18 个延迟级别：
| 级别 | 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 |
注意：RocketMQ 不支持任意精度的延迟时间，只能从这 18 个级别中选择。如果需要更灵活的延迟，可以在业务层自行实现（如使用 Redis + 定时任务）。
延迟消息示例：
java
// ========== 生产者：发送延迟消息 ==========
public class DelayProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
for (int i = 0; i < 5; i++) {
Message msg = new Message(
"TopicDelay",
"TagDelay",
("Delay Message " + i).getBytes()
);
// 设置延迟级别：level 3 = 10 秒
msg.setDelayTimeLevel(3);
SendResult result = producer.send(msg);
System.out.printf("发送延迟消息: %s, 发送时间: %s%n",
result.getMsgId(),
System.currentTimeMillis());
}
producer.shutdown();
}
}java
// ========== 消费者：消费延迟消息 ==========
public class DelayConsumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.subscribe("TopicDelay", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
System.out.printf("收到延迟消息: %s, 存储时间: %s, 接收时间: %s%n",
new String(msg.getBody()),
msg.getStoreTimestamp(), // 消息存储时间
System.currentTimeMillis()); // 实际消费时间
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("延迟消息消费者启动...");
}
}延迟消息的实现原理：
延迟消息在 Broker 端的处理流程：
Producer 发送消息（delayLevel=3，即 10s）
│
▼
Broker 收到消息
│
├── 将消息的原始 Topic 改为 SCHEDULE_TOPIC_XXXX
├── 将消息存储到 CommitLog
├── 根据 delayLevel 投递到对应的 ConsumeQueue
│
▼
定时任务（Timer）扫描 SCHEDULE_TOPIC_XXXX
│
├── 检查消息是否达到延迟时间
│ ├── 未到达 → 继续等待
│ └── 已到达 → 恢复原始 Topic，投递到真实 ConsumeQueue
│
▼
Consumer 消费到消息（10s 后）4.4 事务消息
事务消息是 RocketMQ 实现分布式事务最终一致性的核心机制。它通过"半消息 + 事务状态回查"的方式，保证本地事务和消息发送的原子性。
事务消息的执行流程：
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Producer │ │ Broker │ │ Consumer │
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
│ ① 发送半消息（Half Msg） │ │
│──────────────────────────────►│ │
│ │ 半消息对 Consumer 不可见 │
│ ② 返回半消息发送结果 │ │
│◄──────────────────────────────│ │
│ │ │
│ ③ 执行本地事务 │ │
│ ┌─────────────────────┐ │ │
│ │ 扣减库存 / 更新订单 │ │ │
│ └──────────┬──────────┘ │ │
│ │ │ │
│ ④ 根据本地事务结果提交/回滚 │ │
│─────── COMMIT / ROLLBACK ─────►│ │
│ │ │ │
│ │ COMMIT ─────────┼──► 消息对 Consumer 可见 │
│ │ │──────────────────────────────►│
│ │ │ │
│ │ ROLLBACK ───────┼──► 消息被删除，不会投递 │
│ │ │ │
│ ⑤ 如果 Broker 未收到提交/回滚 │ │
│ 或者 Producer 宕机 │ │
│ │ │ │
│ │ ⑥ 事务状态回查 │ │
│ │◄─────────────────│ │
│ │ │ │
│ ⑦ 返回本地事务最终状态 │ │
│ │─────────────────►│ │
│ │ │ │事务消息代码示例：
java
public class TransactionProducer {
public static void main(String[] args) throws Exception {
// 事务消息生产者
TransactionMQProducer producer = new TransactionMQProducer("transaction_group");
producer.setNamesrvAddr("127.0.0.1:9876");
// 自定义线程池，用于执行事务回查
ExecutorService executorService = new ThreadPoolExecutor(
2, 5, 100, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2000),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("transaction-check-thread");
return thread;
}
}
);
producer.setExecutorService(executorService);
// 设置事务监听器
producer.setTransactionListener(new TransactionListener() {
/**
* 执行本地事务
* 在发送半消息成功后调用
*/
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
// 模拟执行本地事务
String businessKey = msg.getKeys();
System.out.println("执行本地事务, businessKey: " + businessKey);
try {
// 这里是你的业务逻辑：扣减库存、更新订单状态等
boolean success = doBusinessLogic(businessKey);
if (success) {
// 本地事务执行成功，提交事务消息
System.out.println("本地事务成功，提交消息");
return LocalTransactionState.COMMIT_MESSAGE;
} else {
// 本地事务执行失败，回滚事务消息
System.out.println("本地事务失败，回滚消息");
return LocalTransactionState.ROLLBACK_MESSAGE;
}
} catch (Exception e) {
// 异常情况，返回 UNKNOW，等待 Broker 回查
System.out.println("本地事务异常，返回 UNKNOW");
return LocalTransactionState.UNKNOW;
}
}
/**
* 事务状态回查
* 当 Broker 未收到 COMMIT/ROLLBACK，或收到 UNKNOW 时调用
* 时间间隔：默认 60s 一次，最多回查 15 次
*/
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
String businessKey = msg.getKeys();
System.out.println("事务状态回查, businessKey: " + businessKey);
// 查询本地事务的执行状态
boolean localTxStatus = checkLocalTransactionStatus(businessKey);
if (localTxStatus) {
System.out.println("回查：本地事务已提交，提交消息");
return LocalTransactionState.COMMIT_MESSAGE;
} else {
System.out.println("回查：本地事务已回滚，回滚消息");
return LocalTransactionState.ROLLBACK_MESSAGE;
}
}
});
producer.start();
// 发送事务消息
for (int i = 0; i < 5; i++) {
Message msg = new Message(
"TopicTransaction",
"TagA",
"KEY" + i,
("Transaction Message " + i).getBytes()
);
// 发送事务消息，第二个参数是传给 executeLocalTransaction 的 arg
TransactionSendResult result = producer.sendMessageInTransaction(msg, null);
System.out.printf("事务消息发送结果: %s%n", result);
}
// 防止程序退出（等待回查）
Thread.sleep(60000);
producer.shutdown();
}
private static boolean doBusinessLogic(String businessKey) {
// 模拟业务逻辑
return true;
}
private static boolean checkLocalTransactionStatus(String businessKey) {
// 查询数据库，判断本地事务是否已提交
// 比如：根据 businessKey 查订单表，看订单状态
return true;
}
}事务消息的三种状态：
| 状态 | 含义 | 后续动作 |
|---|---|---|
COMMIT_MESSAGE | 本地事务提交成功 | Broker 将半消息标记为可投递，Consumer 可以消费 |
ROLLBACK_MESSAGE | 本地事务回滚 | Broker 删除半消息，Consumer 不会收到 |
UNKNOW | 本地事务状态未知 | Broker 等待后发起事务状态回查 |
事务消息的适用场景：
- 下单扣库存：订单创建成功（本地事务） + 库存扣减消息（异步通知）
- 充值到账：支付流水写入（本地事务） + 账户余额变更消息（异步通知）
- 积分发放：订单完成（本地事务） + 积分变更消息（异步通知）
事务消息的局限性：
- Consumer 端无法保证幂等，需要业务层自行实现幂等
- 不保证强一致性，只保证最终一致性
- 事务回查有时间窗口，极端情况下可能延迟数分钟
五、消费模式
5.1 集群消费（Clustering）
默认模式。一条消息只会被 ConsumerGroup 中的一个消费者消费。消费者组内的消费者分摊 Queue，实现负载均衡和水平扩展。
集群消费模式：
Topic: Order
Queue: Q0, Q1, Q2, Q3, Q4, Q5
ConsumerGroup: order-consumer-group
┌──────────┐ ┌──────────┐ ┌──────────┐
│ C-1 │ │ C-2 │ │ C-3 │
│ │ │ │ │ │
│ Q0, Q1 │ │ Q2, Q3 │ │ Q4, Q5 │
└──────────┘ └──────────┘ └──────────┘
消息 M1 发送到 Q0 → 只有 C-1 消费
消息 M2 发送到 Q3 → 只有 C-2 消费
消息 M3 发送到 Q5 → 只有 C-3 消费
扩容：新增 C-4，自动重平衡，C-4 从各消费者分走部分 Queue
缩容：C-3 宕机，C-1 和 C-2 自动接管 Q4, Q5java
// 集群消费（默认模式）
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order-consumer-group");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.setMessageModel(MessageModel.CLUSTERING); // 默认就是集群模式
consumer.subscribe("Order", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
System.out.printf("集群消费: %s%n", new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();5.2 广播消费（Broadcasting）
一条消息会被 ConsumerGroup 中的每一个消费者都消费一次。每个消费者都消费全量消息。
广播消费模式：
Topic: Order
ConsumerGroup: order-broadcast-group
┌──────────┐ ┌──────────┐ ┌──────────┐
│ C-1 │ │ C-2 │ │ C-3 │
│ │ │ │ │ │
│ 全量消息 │ │ 全量消息 │ │ 全量消息 │
└──────────┘ └──────────┘ └──────────┘
消息 M1 → C-1、C-2、C-3 都收到
消息 M2 → C-1、C-2、C-3 都收到
消息 M3 → C-1、C-2、C-3 都收到
⚠ 注意：广播消费不支持顺序消息
⚠ 注意：广播消费的消费进度存储在 Consumer 本地，重启后从上次进度继续java
// 广播消费
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order-broadcast-group");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.setMessageModel(MessageModel.BROADCASTING); // 广播模式
consumer.subscribe("Order", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
System.out.printf("广播消费: %s%n", new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();5.3 两种模式对比
| 维度 | 集群消费（Clustering） | 广播消费（Broadcasting） |
|---|---|---|
| 消息投递 | 一条消息只投递给组内一个消费者 | 一条消息投递给组内所有消费者 |
| 负载均衡 | 支持，Queue 在组内分摊 | 不支持，每个消费者消费全量 |
| 消费进度 | 存储在 Broker 端 | 存储在 Consumer 本地 |
| 水平扩展 | 可以，新增消费者自动分担 | 新增消费者独立消费全量 |
| 顺序消息 | 支持 | 不支持 |
| 重试消息 | 支持 | 不支持 |
| 适用场景 | 订单处理、库存扣减（一条消息处理一次） | 配置刷新、缓存刷新（每个实例都需处理） |
六、消息可靠性
消息可靠性是 RocketMQ 的核心竞争力。RocketMQ 通过刷盘机制和主从复制两个维度来保证消息的可靠性，但可靠性和性能总是存在权衡。
6.1 刷盘机制
刷盘是指将消息从内存（Page Cache）写入磁盘的过程。
同步刷盘：
Producer 发送消息
│
▼
Broker 接收消息
│
├── 写入 Page Cache（内存）
├── 立即调用 fsync() 强制刷盘
│
▼
刷盘完成后返回 ACK 给 Producer
│
▼
Producer 收到 ACK，确认消息已持久化
可靠性：★★★★★ 消息不丢
性能： ★★☆☆☆ 刷盘耗时约 1-10ms，影响吞吐量异步刷盘：
Producer 发送消息
│
▼
Broker 接收消息
│
├── 写入 Page Cache（内存）
├── 立即返回 ACK 给 Producer
│
▼
后台线程定期（默认 500ms）将 Page Cache 刷入磁盘
可靠性：★★★☆☆ 宕机可能丢失最近 500ms 的消息
性能： ★★★★★ 吞吐量极高6.2 主从复制
同步复制（SYNC_MASTER）：
Producer 发送消息
│
▼
Master Broker 接收消息
│
├── 写入 Master CommitLog
├── 等待 Slave 同步完成
│
▼
Master 和 Slave 都写入成功后
│
▼
返回 ACK 给 Producer
可靠性：★★★★★ Master 宕机，Slave 有完整数据
性能： ★★★☆☆ 增加主从同步延迟异步复制（ASYNC_MASTER）：
Producer 发送消息
│
▼
Master Broker 接收消息
│
├── 写入 Master CommitLog
├── 立即返回 ACK 给 Producer
│
▼
后台异步同步给 Slave
可靠性：★★★☆☆ Master 宕机，Slave 可能缺少最新消息
性能： ★★★★★ 无主从同步延迟6.3 四种组合对比
| 组合 | 刷盘 | 复制 | 可靠性 | 性能 | 适用场景 |
|---|---|---|---|---|---|
| 同步刷盘 + 同步复制 | SYNC_FLUSH | SYNC_MASTER | 最高 | 最低 | 金融支付、对账 |
| 同步刷盘 + 异步复制 | SYNC_FLUSH | ASYNC_MASTER | 高 | 中 | 重要业务，可接受主从切换丢失少量 |
| 异步刷盘 + 同步复制 | ASYNC_FLUSH | SYNC_MASTER | 中高 | 中 | 可靠性要求较高，但需要高吞吐 |
| 异步刷盘 + 异步复制 | ASYNC_FLUSH | ASYNC_MASTER | 一般 | 最高 | 日志采集、埋点数据（可容忍丢失） |
6.4 生产者重试机制
当 Producer 发送消息失败时，RocketMQ 会自动重试，保证消息不丢失。
java
// 生产者重试配置
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
// 同步发送失败重试次数（默认 2 次）
producer.setRetryTimesWhenSendFailed(3);
// 异步发送失败重试次数（默认 2 次）
producer.setRetryTimesWhenSendAsyncFailed(3);
// 重试时尝试发送到其他 Broker（默认 false）
producer.setRetryAnotherBrokerWhenNotStoreOK(true);
// 发送超时时间（默认 3000ms）
producer.setSendMsgTimeout(5000);生产者重试流程：
发送消息失败
│
├── 第 1 次重试（立即）
│ ├── 成功 → 返回成功
│ └── 失败 → 继续
│
├── 第 2 次重试（立即）
│ ├── 成功 → 返回成功
│ └── 失败 → 继续
│
├── 第 3 次重试（立即）
│ ├── 成功 → 返回成功
│ └── 失败 → 抛出异常，业务层处理（记录日志、告警、人工介入）
│
└── setRetryAnotherBrokerWhenNotStoreOK(true)
重试时会避开上一轮失败的 Broker，选择其他 Broker6.5 消费者重试机制
当消费者消费消息失败时，RocketMQ 会将消息发送到重试队列，延迟后重新投递。
重试队列：
消费失败
│
▼
消息投递到重试队列：%RETRY%{ConsumerGroup}
│
├── 第 1 次重试：10s 后
├── 第 2 次重试：30s 后
├── 第 3 次重试：1m 后
├── 第 4 次重试：2m 后
├── ...
├── 第 16 次重试：2h 后
│
▼
仍失败 → 消息进入死信队列：%DLQ%{ConsumerGroup}重试间隔时间表：
| 重试次数 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15 | 16 | |----------|---|---|---|---|---|---|---|---|---|---|----|----|----|----|----|----|----|----| | 延迟时间 | 10s | 30s | 1m | 2m | 3m | 4m | 5m | 6m | 7m | 8m | 9m | 10m | 20m | 30m | 1h | 2h |
死信队列（DLQ - Dead Letter Queue）：
当消息重试 16 次仍然失败后，消息会被移入死信队列。死信队列中的消息需要人工处理或通过定时任务处理。
java
// 消费者重试配置
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
// 最大重试次数（默认 16 次），超过后进入死信队列
consumer.setMaxReconsumeTimes(16);
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
try {
// 业务处理
processMessage(msg);
} catch (Exception e) {
// 获取当前重试次数
int reconsumeTimes = msg.getReconsumeTimes();
System.out.printf("消息重试第 %d 次%n", reconsumeTimes);
if (reconsumeTimes >= 3) {
// 重试超过 3 次，记录到 DB 或发送告警，手动处理
saveToDeadLetterTable(msg);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
// 返回 RECONSUME_LATER，消息会进入重试队列
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});七、核心配置
7.1 Spring Cloud Stream RocketMQ YAML 配置
yaml
spring:
application:
name: order-service
cloud:
stream:
# 默认绑定器
default-binder: rocketmq
# 绑定器配置
rocketmq:
binder:
# NameServer 地址（必填）
name-server: 127.0.0.1:9876
# 可选：多个 NameServer 用分号分隔
# name-server: 127.0.0.1:9876;127.0.0.2:9876
# 生产者配置
producers:
# 生产者组名
group: order-producer-group
# 发送消息超时时间（毫秒）
send-message-timeout: 3000
# 同步发送失败重试次数
retry-times-when-send-failed: 3
# 异步发送失败重试次数
retry-times-when-send-async-failed: 3
# 消息体最大字节数
max-message-size: 4194304
# 压缩消息体阈值（大于此值自动压缩）
compress-message-body-threshold: 4096
# 失败时尝试发送到其他 Broker
retry-next-server: true
# VIP 通道（云环境可能需关闭）
vip-channel-enabled: false
# 消费者配置
consumers:
# 消费者组名
group: order-consumer-group
# 消息模型：CLUSTERING（集群）/ BROADCASTING（广播）
message-model: CLUSTERING
# 消费线程池最小线程数
consume-thread-min: 20
# 消费线程池最大线程数
consume-thread-max: 64
# 每次拉取消息数量
pull-batch-size: 32
# 批量消费时每次最大消息数
max-batch-size: 1
# 消费超时时间（分钟）
consume-timeout: 15
# 最大重试次数
max-reconsume-times: 3
# 消费失败后挂起当前队列的时间（毫秒）
suspend-current-queue-time-millis: 1000
# 消息过滤表达式类型：tag / sql92
selector-type: tag
# 消息过滤表达式
selector-expression: "*"
# 绑定定义
bindings:
# 输出通道（生产者）
order-output:
# 目标 Topic
destination: order-topic
# 内容类型
content-type: application/json
# 生产者配置
producer:
# 生产者组（可覆盖全局配置）
group: order-producer-group
# 事务消息
transactional: false
# 消息标签（用于消费者过滤）
# 也可以在这里指定 tag
# tags: order-create
# 输入通道（消费者）
order-input:
destination: order-topic
content-type: application/json
# 消费者组
group: order-consumer-group
# 消费者配置
consumer:
# 消息过滤：只消费 tag 为 order-create 或 order-pay 的消息
tags: order-create || order-pay
# 并发消费
concurrency: 5
# 最大重试次数
max-attempts: 3
# 函数式编程风格（Spring Cloud Stream 3.x+ 推荐）
function:
definition: orderConsumer;orderProducer
stream:
bindings:
orderConsumer-in-0:
destination: order-topic
group: order-consumer-group
orderProducer-out-0:
destination: order-topic7.2 原生 RocketMQ 配置（rocketmq-client 方式）
yaml
rocketmq:
# NameServer 地址
name-server: 127.0.0.1:9876
# 生产者配置
producer:
# 生产者组
group: order-producer-group
# 发送消息超时时间
send-message-timeout: 3000
# 重试次数
retry-times-when-send-failed: 3
# 异步发送重试次数
retry-times-when-send-async-failed: 3
# 消息最大大小
max-message-size: 4194304
# 压缩阈值
compress-message-body-threshold: 4096
# 失败时切换 Broker
retry-next-server: true
# 访问密钥（ACL 模式下需要）
access-key: your-access-key
# 访问密钥（ACL 模式下需要）
secret-key: your-secret-key
# 消费者配置
consumer:
# 消费者组
group: order-consumer-group
# 消息模型
message-model: CLUSTERING
# 消费线程数
consume-thread-min: 20
consume-thread-max: 64
# 批量消费最大消息数
consume-message-batch-max-size: 1
# 从哪个位置开始消费
# CONSUME_FROM_LAST_OFFSET: 从队列最后开始（默认）
# CONSUME_FROM_FIRST_OFFSET: 从队列开头开始
# CONSUME_FROM_TIMESTAMP: 从指定时间开始
consume-from-where: CONSUME_FROM_LAST_OFFSET
# 消息过滤类型
selector-type: tag
# 消息过滤表达式
selector-expression: "*"
# 最大重试次数
max-reconsume-times: 3
# 消费者监听器配置（多个消费者组）
listeners:
# 消费者组 1
- group: order-consumer-group
topic: order-topic
selector-type: tag
selector-expression: order-create || order-pay
message-model: CLUSTERING
consume-thread-max: 64
# 消费者组 2
- group: log-consumer-group
topic: log-topic
selector-type: tag
selector-expression: "*"
message-model: CLUSTERING
consume-thread-max: 20八、Spring Cloud Stream + RocketMQ 集成
8.1 依赖配置
xml
<!-- pom.xml -->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
<version>2021.0.5.0</version>
</dependency>8.2 生产者（函数式编程风格）
java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
@Service
public class OrderProducer {
@Autowired
private StreamBridge streamBridge;
/**
* 发送普通消息
*/
public void sendOrderMessage(Order order) {
Message<Order> message = MessageBuilder
.withPayload(order)
.setHeader("tag", "order-create") // 设置 Tag
.setHeader("keys", order.getOrderId()) // 设置 Key（用于消息查询）
.build();
boolean result = streamBridge.send("orderProducer-out-0", message);
if (result) {
System.out.println("消息发送成功");
} else {
System.out.println("消息发送失败");
}
}
/**
* 发送延迟消息
*/
public void sendDelayMessage(Order order) {
Message<Order> message = MessageBuilder
.withPayload(order)
.setHeader("tag", "order-delay")
.setHeader("delayLevel", 3) // 延迟级别 3 = 10 秒
.build();
streamBridge.send("orderProducer-out-0", message);
}
/**
* 发送事务消息（需要配合 TransactionListener）
*/
public void sendTransactionMessage(Order order) {
Message<Order> message = MessageBuilder
.withPayload(order)
.setHeader("tag", "order-transaction")
.setHeader("transactional", true) // 标记为事务消息
.setHeader("transactionId", order.getOrderId())
.build();
streamBridge.send("orderProducer-out-0", message);
}
}8.3 消费者（函数式编程风格）
java
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import java.util.function.Consumer;
@Configuration
public class OrderConsumer {
/**
* 普通消息消费者
* 函数名 = binding name 中 -in- 前面的部分
* 即 orderConsumer-in-0 → 函数名为 orderConsumer
*/
@Bean
public Consumer<Message<Order>> orderConsumer() {
return message -> {
Order order = message.getPayload();
String tag = (String) message.getHeaders().get("tag");
String keys = (String) message.getHeaders().get("keys");
System.out.printf("收到消息: tag=%s, keys=%s, order=%s%n",
tag, keys, order);
// 业务处理
processOrder(order);
};
}
/**
* 带条件过滤的消费者
* 只消费 tag 为 order-pay 的消息
*
* 在 application.yml 中配置：
* spring.cloud.stream.bindings.orderPayConsumer-in-0.consumer.tags: order-pay
*/
@Bean
public Consumer<Message<Order>> orderPayConsumer() {
return message -> {
Order order = message.getPayload();
System.out.println("收到支付消息: " + order.getOrderId());
processPayment(order);
};
}
private void processOrder(Order order) {
// 订单业务处理
}
private void processPayment(Order order) {
// 支付业务处理
}
}8.4 消息过滤配置
yaml
spring:
cloud:
stream:
bindings:
# 消费者 1：只消费订单创建消息
orderCreateConsumer-in-0:
destination: order-topic
group: order-create-consumer
consumer:
tags: order-create # 只消费 tag=order-create 的消息
# 消费者 2：只消费订单支付消息
orderPayConsumer-in-0:
destination: order-topic
group: order-pay-consumer
consumer:
tags: order-pay # 只消费 tag=order-pay 的消息
# 消费者 3：消费订单创建和支付消息
orderAllConsumer-in-0:
destination: order-topic
group: order-all-consumer
consumer:
tags: order-create || order-pay # 消费两种 tag 的消息
# 消费者 4：消费所有消息（不区分 tag）
orderLogConsumer-in-0:
destination: order-topic
group: order-log-consumer
consumer:
tags: "*" # 消费所有消息8.5 事务消息完整示例
java
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
@Service
public class TransactionOrderService {
@Autowired
private StreamBridge streamBridge;
@Autowired
private OrderService orderService;
/**
* 发送事务消息
*/
public void sendOrderTransaction(Order order) {
Message<Order> message = MessageBuilder
.withPayload(order)
.setHeader(RocketMQHeaders.TRANSACTION_ID, order.getOrderId())
.build();
streamBridge.send("orderTransaction-out-0", message);
}
}
/**
* 事务消息监听器
*/
@RocketMQTransactionListener
class OrderTransactionListener implements RocketMQLocalTransactionListener {
@Autowired
private OrderService orderService;
/**
* 执行本地事务
*/
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
String transactionId = (String) msg.getHeaders()
.get(RocketMQHeaders.TRANSACTION_ID);
try {
// 执行本地事务：创建订单
Object payload = msg.getPayload();
boolean result = orderService.createOrder((Order) payload);
if (result) {
return RocketMQLocalTransactionState.COMMIT;
} else {
return RocketMQLocalTransactionState.ROLLBACK;
}
} catch (Exception e) {
return RocketMQLocalTransactionState.UNKNOWN;
}
}
/**
* 事务状态回查
*/
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
String transactionId = (String) msg.getHeaders()
.get(RocketMQHeaders.TRANSACTION_ID);
// 根据 transactionId 查询本地事务状态
Order order = orderService.getOrderByTransactionId(transactionId);
if (order != null && order.getStatus() == OrderStatus.CREATED) {
return RocketMQLocalTransactionState.COMMIT;
} else {
return RocketMQLocalTransactionState.ROLLBACK;
}
}
}yaml
# application.yml - 事务消息绑定配置
spring:
cloud:
stream:
bindings:
orderTransaction-out-0:
destination: order-topic
producer:
transactional: true # 开启事务消息
group: order-transaction-producer8.6 完整配置汇总
yaml
spring:
application:
name: order-service
cloud:
stream:
function:
definition: orderConsumer;orderPayConsumer;orderTransaction
rocketmq:
binder:
name-server: 127.0.0.1:9876
producers:
group: order-producer-group
send-message-timeout: 3000
retry-times-when-send-failed: 3
consumers:
group: order-consumer-group
consume-thread-max: 64
max-reconsume-times: 3
message-model: CLUSTERING
bindings:
# 生产者
orderProducer-out-0:
destination: order-topic
content-type: application/json
# 事务消息生产者
orderTransaction-out-0:
destination: order-topic
content-type: application/json
producer:
transactional: true
# 消费者：消费创建消息
orderConsumer-in-0:
destination: order-topic
content-type: application/json
group: order-create-consumer
consumer:
tags: order-create
concurrency: 5
# 消费者：消费支付消息
orderPayConsumer-in-0:
destination: order-topic
content-type: application/json
group: order-pay-consumer
consumer:
tags: order-pay
concurrency: 3九、RocketMQ vs Kafka vs RabbitMQ
9.1 综合对比
| 维度 | RocketMQ | Kafka | RabbitMQ |
|---|---|---|---|
| 开发语言 | Java | Java/Scala | Erlang |
| 所属 | Apache / 阿里巴巴 | Apache / LinkedIn | VMware / Pivotal |
| 吞吐量 | 极高（十万级/秒） | 极高（百万级/秒） | 中等（万级/秒） |
| 延迟 | 毫秒级 | 毫秒级 | 微秒级 |
| 可靠性 | 高（同步刷盘/同步复制） | 高（ISR 副本机制） | 高（持久化+镜像队列） |
| 消息堆积 | 极强（亿级堆积，性能衰减极小） | 强（依赖磁盘容量） | 弱（堆积严重影响性能） |
| 事务消息 | 原生支持（半消息+回查） | 支持（Kafka Transactions） | 不支持原生事务消息 |
| 顺序消息 | 原生支持（分区顺序 + 全局顺序） | 支持（分区内有序） | 支持（队列内有序） |
| 延迟消息 | 原生支持（18 个延迟级别） | 不支持（需自行实现） | 支持（插件或死信+TTL） |
| 消息过滤 | Tag + SQL92 表达式 | 不支持服务端过滤（需客户端过滤） | 支持 Routing Key |
| 消息重试 | 原生支持（重试队列+死信队列） | 不支持（需自行实现） | 支持（死信队列） |
| 消费模式 | 集群 / 广播 | 集群（ConsumerGroup） | 推模式 / 拉模式 |
| 消息回溯 | 支持（按时间回溯） | 支持（按 Offset 回溯） | 不支持 |
| 定时消息 | 支持（18 个延迟级别） | 不支持 | 不支持 |
| 适用场景 | 业务消息、事务消息、电商场景 | 日志采集、流计算、大数据 | 企业应用集成、复杂路由 |
| 社区活跃度 | 活跃（阿里+Apache） | 非常活跃（Confluent） | 活跃 |
| 运维复杂度 | 中等 | 较高（依赖 Zookeeper） | 较低 |
| 协议 | 自定义（Remoting） | 自定义（二进制） | AMQP、MQTT、STOMP |
9.2 选型建议
开始
│
▼
┌─ 是否大数据/日志采集场景？ ─┐
│ 是 │ 否
▼ ▼
Kafka ┌─ 是否需要事务消息？ ─┐
(高吞吐、流计算) │ 是 │ 否
▼ ▼
RocketMQ ┌─ 是否需要复杂路由？ ─┐
(电商、金融、分布式事务) │ 是 │ 否
▼ ▼
RabbitMQ RocketMQ
(企业集成、复杂路由) (通用业务消息)一句话总结：
| 场景 | 推荐 | 原因 |
|---|---|---|
| 电商业务（订单、支付、库存） | RocketMQ | 事务消息、顺序消息、延迟消息全支持 |
| 日志采集、大数据管道 | Kafka | 吞吐量最高，生态集成好 |
| 企业应用集成、复杂路由 | RabbitMQ | AMQP 标准协议，路由灵活，运维简单 |
| 金融级可靠性 | RocketMQ | 同步刷盘 + 同步复制，消息零丢失 |
| 微服务异步通信 | RocketMQ | Spring Cloud Stream 原生集成，开发体验好 |
十、面试要点
10.1 RocketMQ 如何保证消息不丢失？
消息不丢失需要从生产端、Broker 端、消费端三个阶段分别保证。
生产端 Broker 端 消费端
│ │ │
▼ ▼ ▼
┌─────────┐ ┌───────────┐ ┌──────────┐
│ 同步发送 │ │ 同步刷盘 │ │ 手动提交 │
│ 重试机制 │ │ 主从同步 │ │ 消费重试 │
│ 失败告警 │ │ 多副本部署 │ │ 幂等处理 │
└─────────┘ └───────────┘ └──────────┘生产端：
- 使用同步发送，等待 Broker 返回确认
- 配置重试次数（
retryTimesWhenSendFailed） - 发送失败记录日志 + 告警，必要时人工介入
Broker 端：
- 同步刷盘（
flushDiskType = SYNC_FLUSH），消息写入磁盘后才返回 ACK - 同步复制（
brokerRole = SYNC_MASTER），Master 和 Slave 都写入后才返回 ACK - 多 Broker 多副本部署，避免单点故障
消费端：
- 消费成功后再返回 CONSUME_SUCCESS，不要提前 ACK
- 消费失败返回 RECONSUME_LATER，利用重试队列重试
- 业务层实现幂等消费，防止重复消费导致的业务异常
10.2 事务消息的原理是什么？
RocketMQ 的事务消息基于"两阶段提交 + 事务状态回查"实现：
阶段一：发送半消息
Producer 发送一条"半消息"到 Broker
半消息对 Consumer 不可见（不会投递）
阶段二：执行本地事务 + 提交/回滚
Producer 收到半消息发送成功的响应后，执行本地事务
本地事务成功 → 向 Broker 发送 COMMIT
本地事务失败 → 向 Broker 发送 ROLLBACK
异常处理：事务状态回查
如果 Broker 长时间未收到 COMMIT/ROLLBACK（Producer 宕机、网络中断等）
Broker 会主动向 Producer 发起事务状态回查
Producer 查询本地事务的最终状态，返回 COMMIT 或 ROLLBACK
Broker 默认回查 15 次，每次间隔 60s核心要点：
- 半消息是存在 CommitLog 中的，只是对 Consumer 不可见
- 事务回查需要 Producer 端实现
checkLocalTransaction方法 - 不保证强一致性，只保证最终一致性
10.3 NameServer 的工作原理是什么？
- NameServer 是无状态的轻量级路由注册中心，各节点之间互不通信，完全对等
- Broker 启动时向所有 NameServer 注册路由信息（Topic、Queue、Broker 地址），之后每 30s 发送心跳
- Producer 和 Consumer 启动时从任意 NameServer 拉取路由信息，之后每 30s 定时更新
- 如果 NameServer 超过 120s 未收到 Broker 心跳，则将该 Broker 从路由表中移除
- Producer 和 Consumer 获取路由信息后，直接与 Broker 通信，不经过 NameServer
- NameServer 之间互不通信，因此可能出现短暂的路由不一致（AP 模型），但最终会一致
10.4 顺序消息是如何实现的？
RocketMQ 的顺序消息实现分为生产端和消费端两部分：
生产端（MessageQueueSelector）：
- 使用
MessageQueueSelector将需要顺序的消息路由到同一个 Queue - 比如按订单 ID 取模：
orderId % queueSize，确保同一订单的消息进入同一 Queue - 同一个 Queue 中的消息是严格有序的（FIFO）
消费端（MessageListenerOrderly）：
- 使用
MessageListenerOrderly而非MessageListenerConcurrently - 为每个 ConsumerQueue 加锁，确保同一时刻只有一个线程消费该 Queue
- 消费完成后再释放锁，然后拉取下一条消息
局限性：
- 顺序消费会牺牲并发能力，同一个 Queue 的消息是串行消费的
- 如果某个 Queue 的消息处理失败导致阻塞，该 Queue 后续消息都无法消费
- 全局顺序需要将 Topic 的 Queue 数量设置为 1
10.5 延迟消息的原理是什么？
- RocketMQ 不支持任意精度的延迟，只支持 18 个预定义的延迟级别
- 延迟消息发送后，Broker 将其原始 Topic 替换为内部的
SCHEDULE_TOPIC_XXXX - 消息被存储到 CommitLog，同时根据 delayLevel 投递到对应的 ConsumeQueue
- 后台定时任务（Timer）不断扫描 SCHEDULE_TOPIC_XXXX
- 当消息到达延迟时间后，恢复原始 Topic，将消息投递到真实的 ConsumeQueue
- Consumer 此时才能消费到该消息
18 个延迟级别： 1s / 5s / 10s / 30s / 1m / 2m / 3m / 4m / 5m / 6m / 7m / 8m / 9m / 10m / 20m / 30m / 1h / 2h
10.6 RocketMQ vs Kafka 如何选择？
| 对比维度 | RocketMQ | Kafka |
|---|---|---|
| 设计目标 | 业务消息处理 | 海量日志/流数据处理 |
| 消息堆积 | 亿级堆积，性能衰减极小 | 堆积越多性能越差 |
| 消息过滤 | 服务端 Tag 和 SQL92 过滤 | 不支持服务端过滤 |
| 事务消息 | 原生支持 | 支持但实现复杂 |
| 延迟消息 | 原生 18 个级别 | 不支持 |
| 消息重试 | 原生重试队列 + 死信队列 | 不支持 |
| 依赖 | 自身 NameServer（轻量） | Zookeeper（较重） |
| 适用场景 | 电商、金融、微服务 | 大数据、日志、流计算 |
选择建议： 业务场景选 RocketMQ，大数据场景选 Kafka。不要为了技术而技术，选择最适合业务的技术。
附录：常用命令速查
启动 RocketMQ
bash
# 启动 NameServer
nohup sh bin/mqnamesrv &
# 启动 Broker
nohup sh bin/mqbroker -n 127.0.0.1:9876 &
# 检查是否启动成功
jps
# 应该看到 NamesrvStartup 和 BrokerStartup管理命令
bash
# 查看所有 Topic
sh bin/mqadmin topicList -n 127.0.0.1:9876
# 创建 Topic
sh bin/mqadmin updateTopic -n 127.0.0.1:9876 -b 127.0.0.1:10911 -t order-topic
# 查看 Topic 详情
sh bin/mqadmin topicStatus -n 127.0.0.1:9876 -t order-topic
# 查看 ConsumerGroup 消费进度
sh bin/mqadmin consumerProgress -n 127.0.0.1:9876 -g order-consumer-group
# 查看集群状态
sh bin/mqadmin clusterList -n 127.0.0.1:9876
# 根据 Key 查询消息
sh bin/mqadmin queryMsgByKey -n 127.0.0.1:9876 -t order-topic -k ORDER123参考文档：