RocketMQ 笔记

订单服务
    ↓ 同步调用
库存服务
    ↓ 同步调用
积分服务
    ↓ 同步调用
通知服务

用户提交订单
      ↓
订单服务保存订单
      ↓
发送 OrderCreatedEvent
      ↓
RocketMQ Broker 持久化消息
      ↓
 ┌────────────┬────────────┬────────────┐
库存服务      积分服务      通知服务
扣减库存      增加积分      发送短信

Topic：OrderTopic（共 6 个 Queue）
              │
          ┌───┴───┐
          ▼       ▼
       Broker A  Broker B
       Queue 0   Queue 3
       Queue 1   Queue 4
       Queue 2   Queue 5

┌─────────────────────────────────────────────────────────────┐
│                    Broker 集群（Cluster）                     │
│                                                             │
│   ┌─────────────────────┐      ┌─────────────────────┐      │
│   │    Broker A 主从     │      │    Broker B 主从    │      │
│   │  ┌─────┐ ┌─────┐    │      │  ┌─────┐ ┌─────┐    │      │
│   │  │Master│ │Slave│   │      │  │Master│ │Slave│   │      │
│   │  │(读写)│ │(同步)│    │      │  │(读写)│ │(同步)│   │      │
│   │  └──┬──┘ └──┬──┘    │      │  └──┬──┘ └──┬──┘    │      │
│   │     └────同步────┘   │      │     └────同步────┘  │       │
│   │存 Topic 的 Queue 0-2 │      │存 Topic 的 Queue 3-5 │      │
│   └─────────────────────┘      └─────────────────────┘      │
│                                                             │
│             ←── 集群：分散负载、水平扩展 ──→                     │
│             ←── 主从：数据备份、故障切换 ──→                    │
└─────────────────────────────────────────────────────────────┘

单个 Broker 故障时，Topic 仍有其他 Broker 和 Queue 可以承载新消息。

    Master Broker A          Slave Broker A
    (主，读写)               (从，同步复制)
         │                        │
         └──────── 同步复制 ──────┘

brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH

异步主从复制 + 异步刷盘

brokerRole=SYNC_MASTER
flushDiskType=SYNC_FLUSH

数据副本备份；
Master 不可用时，Consumer 仍可读取 Slave 中已经复制的消息。

传统主从：Master 故障后，Consumer 可以切换读取 Slave，
          但 Slave 不会自动成为 Master，写服务不能自动恢复。

DLedger / Controller：发生故障后可以自动选主，
                     从而恢复该 Broker 副本组的读写能力。

                     NameServer 集群
                            │
                  ┌─────────┴─────────┐
                  ▼                   ▼
          ┌─────────────┐     ┌─────────────┐
          │ Broker A 组  │     │ Broker B 组  │
          │ Master       │     │ Master       │
          │    +         │     │    +         │
          │ Slave        │     │ Slave        │
          └─────────────┘     └─────────────┘
                 ▲                   ▲
                 │                   │
          主从复制与数据备份    主从复制与数据备份

          ←──── 多 Broker：分散 Queue 和请求负载 ────→
          ←──── 主从副本：提高数据与读取可用性 ─────→

NameServer 集群：避免路由服务出现单点故障；
多 Master 集群：分散 Topic Queue，提高吞吐量和 Topic 整体可用性；
Master-Slave：保存数据副本，提高数据可靠性和消费端读可用性；
同步刷盘：降低机器断电导致的数据丢失风险；
同步复制：降低 Master 损坏导致的数据丢失风险；
DLedger / Controller：在 Master 故障后自动选主，恢复写服务；
Producer 重试：当前 Broker 或 Queue 发送失败时尝试其他可用 Queue。

NameServer 集群
       +
多 Broker 副本组
       +
主从数据复制
       +
合理的刷盘策略
       +
自动主从切换机制
       +
生产者重试和业务幂等

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");
DefaultMQPushConsumer consumer = new DefaultMQPullConsumer("test-consumer-group");

Producer 从 NameServer 获取 Topic 路由
        ↓
选择 Topic 下的某个 MessageQueue
        ↓
将消息发送到 MessageQueue 所在的 Broker
        ↓
Broker 持久化保存消息
        ↓
Consumer 从 NameServer 获取 Topic 路由
        ↓
ConsumerGroup 内的消费者分配 MessageQueue
        ↓
Consumer 从对应 Broker 拉取并处理消息

Producer 查询路由 → 消息写入 Broker → Consumer 查询路由 → 从 Broker 拉取消息

Topic：消息的业务分类
MessageQueue：Topic 下的分区，是发送和消费负载均衡的基本单位
Broker：真正保存和投递消息的服务器
NameServer：保存 Topic 与 Broker 之间的路由关系
ConsumerGroup：一组执行相同消费逻辑的消费者
Tag：Topic 内的二级分类
Key：消息的业务检索标识

Producer：发送消息
Consumer：消费消息
NameServer：提供路由注册与发现
Broker：存储、投递和管理消息
Proxy：RocketMQ 5.x 新客户端的接入层

                         ┌────────────────┐
                         │   NameServer   │
                         │ 保存路由信息    │
                         └───────▲────────┘
                                 │ Broker 注册
                                 │ 客户端查询路由
                 ┌───────────────┴────────────────┐
                 │                                │
        ┌────────┴────────┐              ┌────────┴────────┐
        │    Broker A     │              │    Broker B     │
        │ Topic / Queue   │              │ Topic / Queue   │
        │ 消息存储与投递    │              │ 消息存储与投递    │
        └──────▲─────┬────┘              └──────▲─────┬────┘
               │     │                           │     │
             发送   拉取消费                    发送   拉取消费
               │     │                           │     │
          Producer Consumer                 Producer Consumer

Topic：业务上的消息分类
MessageQueue：Topic 下的逻辑分区
Broker：实际存储和提供消息读写服务的服务器

Topic：order-event-topic
│
├── Broker A
│   ├── MessageQueue(topic=order-event-topic, broker=A, queueId=0)
│   ├── MessageQueue(topic=order-event-topic, broker=A, queueId=1)
│   └── MessageQueue(topic=order-event-topic, broker=A, queueId=2)
│
└── Broker B
    ├── MessageQueue(topic=order-event-topic, broker=B, queueId=0)
    ├── MessageQueue(topic=order-event-topic, broker=B, queueId=1)
    └── MessageQueue(topic=order-event-topic, broker=B, queueId=2)

Topic + BrokerName + QueueId

order-event-topic + Broker A + QueueId 0
order-event-topic + Broker B + QueueId 0

一个 Topic 包含多个 MessageQueue；
一个 Topic 的 MessageQueue 可以分布在多个 Broker 上；
一个 Broker 可以承载多个 Topic 的 MessageQueue。

Broker A
├── order-event-topic
│   ├── Queue 0一个 Topic 下面有多个 MessageQueue（默认是 4 个）。生产者发消息时，会根据负载均衡策略（比如轮询、哈希取模）选择其中一个 Queue 来发送。
│   ├── Queue 1
│   └── Queue 2
│
└── payment-event-topic
    ├── Queue 0
    └── Queue 1

Broker B
├── order-event-topic
│   ├── Queue 0
│   ├── Queue 1
│   └── Queue 2
│
└── file-event-topic
    ├── Queue 0
    └── Queue 1

从 Topic 角度：
一个 Topic 可以跨多个 Broker 部署。

从 Broker 角度：
一个 Broker 可以存储多个 Topic 的部分 MessageQueue。

Producer
   ├── 消息 1 → Broker A / Queue 0
   ├── 消息 2 → Broker B / Queue 1
   └── 消息 3 → Broker A / Queue 2

Consumer 1 → Broker A / Queue 0
Consumer 2 → Broker A / Queue 1
Consumer 3 → Broker B / Queue 0
Consumer 4 → Broker B / Queue 1

CommitLog：
保存完整消息内容，Broker 收到的消息通常按顺序追加写入。

ConsumeQueue：
保存某个 Topic、某个 Queue 对应的逻辑索引，
索引中记录消息在 CommitLog 中的位置、大小等信息。

CommitLog：真正存放消息正文的总账本
ConsumeQueue：每个 MessageQueue 对应的目录或索引

CommitLog

位置 100：order-event-topic / Queue 0 / 消息 A
位置 200：payment-event-topic / Queue 1 / 消息 B
位置 300：order-event-topic / Queue 1 / 消息 C
位置 400：order-event-topic / Queue 0 / 消息 D

ConsumeQueue：order-event-topic / Queue 0
├── 指向 CommitLog 位置 100
└── 指向 CommitLog 位置 400

ConsumeQueue：order-event-topic / Queue 1
└── 指向 CommitLog 位置 300

ConsumeQueue：payment-event-topic / Queue 1
└── 指向 CommitLog 位置 200

业务视角：

Topic
  ↓
多个 MessageQueue
  ↓
每个 MessageQueue 中存在一系列消息

物理存储视角：

完整消息统一顺序写入 CommitLog
  ↓
每个 Topic + Queue 对应一个 ConsumeQueue
  ↓
ConsumeQueue 保存指向 CommitLog 的逻辑索引

Topic 是逻辑上的业务分类；
MessageQueue 是 Topic 下的逻辑分区；
Broker 是 MessageQueue 的实际承载节点；
Topic 与 Broker 是通过 MessageQueue 建立的多对多关系；
消息正文实际存储在 Broker 的 CommitLog 中；
ConsumeQueue 为每个 Topic-Queue 保存消息索引。

order-event-topic
file-event-topic
payment-event-topic
notification-topic

Producer 向 Topic 发送消息
Consumer 订阅 Topic 消费消息

每个接口创建一个 Topic
每个用户创建一个 Topic
把完全无关的业务消息放进同一个 Topic

order-event-topic
├── Queue 0
├── Queue 1
├── Queue 2
└── Queue 3

Consumer 1 → Queue 0、Queue 3
Consumer 2 → Queue 1
Consumer 3 → Queue 2

ORDER_CREATED
ORDER_PAID
ORDER_SHIPPED

同一个 orderId
      ↓
始终选择同一个 MessageQueue
      ↓
同一订单的事件保持局部顺序

发送端：同一业务键选择同一个 Queue
Broker：该 Queue 内顺序存储
消费端：对该 Queue 顺序消费

Topic：order-event-topic

Tag：
ORDER_CREATED
ORDER_PAID
ORDER_CANCELLED
ORDER_FINISHED

Topic：大的业务领域
Tag：该领域中的具体事件类型

userId
orderId
fileId
时间戳

订单号
文件 ID
支付流水号
事件 ID

Topic = order-event-topic
Tag   = ORDER_CREATED
Key   = order-10001

eventId：事件唯一标识，主要用于消费幂等
businessId：订单号、文件 ID 等业务主键
traceId：整个调用链的追踪标识
Key：RocketMQ 中用于查询消息的索引标识

Topic：order-event-topic

ConsumerGroup：inventory-service-group
├── inventory-service-1
├── inventory-service-2
└── inventory-service-3

ConsumerGroup：notification-service-group
├── notification-service-1
└── notification-service-2

同一个 ConsumerGroup 内：
多个实例负载均衡，一条消息通常只由其中一个实例处理

不同 ConsumerGroup 之间：
消费进度相互独立，每个消费组都可以消费一遍消息

inventory-service-group
    由组内某一个库存服务实例消费

notification-service-group
    由组内某一个通知服务实例消费

search-service-group
    由组内某一个搜索服务实例消费

订阅 Topic
Tag 或过滤表达式
消费模式
消息处理语义

1. 创建 Producer，并指定 ProducerGroup
2. 配置 NameServer 或接入地址
3. 启动 Producer
4. 创建消息，设置 Topic、Tag、Key、属性和消息体
5. 发送消息
6. 应用关闭时关闭 Producer

{
  "eventId": "evt-20260711-0001",
  "orderId": 10001,
  "userId": 20001,
  "eventType": "ORDER_CREATED",
  "occurredAt": "2026-07-11T20:30:00"
}

Topic
Tag
Key
消息属性
消息体

Producer 查询 Topic 路由
        ↓
获得 Broker 和 MessageQueue 列表
        ↓
根据负载均衡策略选择一个 Queue
        ↓
将消息发送到该 Queue 所在 Broker
        ↓
Broker 写入消息
        ↓
Broker 返回发送结果

扣减库存
创建物流任务
更新搜索索引
发送短信
清理文件
生成缩略图

1. 创建 Consumer，并指定 ConsumerGroup
2. 配置 NameServer 或接入地址
3. 订阅 Topic 和 Tag
4. 注册消息监听器
5. 启动 Consumer
6. 接收并处理消息

Consumer 从 Broker 拉取消息
        ↓
执行本地业务逻辑
        ↓
业务处理成功
        ↓
返回消费成功
        ↓
更新消费进度

消费失败
    ↓
Broker 或客户端按照重试策略重新投递
    ↓
多次失败后进入死信队列

基于 eventId 建唯一索引
基于业务状态进行条件更新
使用去重表记录已消费事件
使用 Redis SETNX 做短期去重
利用数据库唯一约束阻止重复写入

有哪些 Broker
Broker 的地址和角色
每个 Topic 位于哪些 Broker
每个 Topic 有哪些 MessageQueue
Broker 的读写状态
Broker 的存活情况

Producer / Consumer
        ↓ 查询路由
NameServer
        ↓ 返回 Broker 地址和 Queue 信息
Producer / Consumer
        ↓ 直接通信
Broker

保存业务消息
转发每一条消息
执行消息消费
保存完整业务消费状态

首次查询新 Topic
发现新加入的 Broker
感知 Broker 路由变化
定期刷新路由

接收 Producer 消息
将消息写入 CommitLog
维护 ConsumeQueue 和 IndexFile
管理 Topic 和 MessageQueue
响应 Consumer 拉取请求
保存和管理消费进度
执行消息重试和死信处理
处理延时消息
处理事务消息回查
向 NameServer 注册路由
执行主从复制
参与故障恢复和主从切换

NameServer：地图或导航系统
Broker：消息仓库
Topic：仓库中的业务分类
MessageQueue：分类下的逻辑分区
Producer：入库方
Consumer：出库并处理消息的一方

Local 模式：
Proxy 与 Broker 同节点或同进程部署

Cluster 模式：
Proxy 独立组成集群，作为统一接入层

接收 gRPC 客户端请求
协议转换
客户端连接管理
路由和请求转发
为多语言客户端提供统一接入

传统 RocketMQ Java Client
    使用 Remoting 协议
    Producer 和 Consumer 通常直接访问 Broker
    Spring 项目常见 rocketmq-spring-boot-starter

RocketMQ 5.x gRPC Client
    客户端通常访问 Proxy
    使用 Endpoint 或 Proxy 地址
    更适合统一的多语言客户端体系

RocketMQ 服务端版本是 5.x
不代表项目必须使用 5.x gRPC Client

Broker 地址
Broker 名称
Master / Slave 信息
Topic 列表
MessageQueue 数量
读写权限

该 Topic 位于哪些 Broker
每个 Broker 有哪些可写 MessageQueue

Queue 0
Queue 1
Queue 2
Queue 3

消息进入某个 MessageQueue

消息主体顺序写入 CommitLog
对应 ConsumeQueue 写入逻辑索引

Topic 有 4 个 Queue
ConsumerGroup 有 2 个实例

Consumer 1 → Queue 0、Queue 2
Consumer 2 → Queue 1、Queue 3

Broker 返回消息
        ↓
Consumer 执行业务
        ↓
成功：确认消费并推进消费进度
失败：触发后续重试

Broker 向 NameServer 注册 Topic 路由
                ↓
Producer 查询 Topic 路由
                ↓
Producer 选择 MessageQueue
                ↓
Producer 直接向 Broker 发送消息
                ↓
Broker 持久化消息并进行主从复制
                ↓
Consumer 查询 Topic 路由
                ↓
ConsumerGroup 分配 MessageQueue
                ↓
Consumer 从 Broker 拉取消息
                ↓
业务成功后确认消费
                ↓
失败则进入重试与死信机制

Producer
    向 Topic 发送消息
        ↓
Topic
    表示业务消息类别
        ↓
MessageQueue
    是 Topic 下的分区，决定并行度和局部顺序
        ↓
Broker
    保存 MessageQueue 对应的消息数据
        ↓
ConsumerGroup
    订阅 Topic，并在组内分配 MessageQueue
        ↓
Consumer
    拉取消息并执行实际业务

Tag：
Topic 内的二级分类，用于订阅过滤

Key：
业务检索标识，用于查询和问题排查

Topic 决定消息属于哪类业务；
MessageQueue 决定消息写到哪个分区以及如何并行消费；
Broker 决定消息实际存在哪里；
Tag 决定消费者关注哪些事件类型；
Key 决定出现问题时如何检索消息；
ConsumerGroup 决定消息由哪一类业务消费；
Consumer 实例决定最终由哪台机器执行消息处理。

1. Producer 启动
2. Producer 从 NameServer 获取 Topic 路由
3. Producer 选择一个 MessageQueue
4. Producer 将消息发送给 Queue 所在 Broker
5. Broker 把消息追加到 CommitLog
6. Broker 建立 ConsumeQueue 和索引
7. Broker 返回发送结果
8. Consumer 从 Broker 获取消息
9. Consumer 执行业务逻辑
10. Consumer 提交消费结果
11. Broker 更新该 Consumer Group 的消费进度

queueIndex = hash(orderId) % queueCount;

客户端向 Broker 发起拉取请求
      ↓
没有消息时 Broker 暂时挂起请求
      ↓
新消息到达或超时
      ↓
Broker 返回消息
      ↓
客户端放入本地缓存
      ↓
消费线程池调用监听器

CommitLog
┌──────────────────────────────────────────┐
│ TopicA-Q0-message1                       │
│ TopicB-Q2-message1                       │
│ TopicA-Q1-message2                       │
│ TopicC-Q0-message1                       │
│ TopicA-Q0-message3                       │
└──────────────────────────────────────────┘

消息总长度
消息 ID
Topic
Queue ID
Queue Offset
消息体
Tag
Key
属性
时间戳
物理偏移量
CRC 等校验信息

consumequeue/
└── order-event-topic/
    ├── 0/
    ├── 1/
    ├── 2/
    └── 3/

消息在 CommitLog 中的物理偏移
消息长度
Tag Hash 等辅助信息

CommitLog：真正存放完整消息
ConsumeQueue：按 Topic + Queue 组织的消费索引

应用写入映射内存
      ↓
数据进入操作系统 Page Cache
      ↓
操作系统或刷盘线程写入磁盘

消息写入 Page Cache
      ↓
Broker 返回发送成功
      ↓
后台线程批量刷入磁盘

消息写入 Page Cache
      ↓
等待消息真正刷入磁盘
      ↓
Broker 返回发送成功

通知、日志、非核心事件：
    通常更关注吞吐量，可以接受异步刷盘

支付、结算、资金相关事件：
    更关注可靠性，需要结合同步刷盘、同步复制和业务补偿

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>${rocketmq-spring.version}</version>
</dependency>

【消息元数据（Header）】
KEYS = "uuid-xxx-eventId"
fileId = "file_10086"
schemaVersion = 1
TAG = "FILE_UPLOADED"
MSG_ID = "Broker生成的全局消息ID"
投递时间、重试次数、事务标记等系统内置头

【消息体 Body（二进制）】
{
  "eventId": "uuid-xxx-eventId",
  "fileId": "file_10086",
  "fileName": "a.pdf",
  "fileSize": 102400,
  "uploadTime": "2026-07-14"
}

spring:
  application:
    name: rocketmq-producer
rocketmq:
  # rocketMq的nameServer地址
  name-server: 127.0.0.1:9876     
  producer:
    # 生产者组别
    group: test-group
    # 消息发送的超时时间
    send-message-timeout: 3000
    # 异步消息发送失败重试次数
    retry-times-when-send-async-failed: 2
    # 发送消息的最大大小，单位字节，这里等于4M
    max-message-size: 4194304      

rocketmq:
  name-server: ${ROCKETMQ_NAME_SERVER}

/**
 * 注入rocketMQTemplate，我们使用它来操作mq
 */
@Autowired
private RocketMQTemplate rocketMQTemplate;
 
/**
 * 测试发送简单的消息
 *
 * @throws Exception
 */
@Test
public void testSimpleMsg() throws Exception {
    // 往test的主题里面发送一个简单的字符串消息
    // syncSend同步消息
    // asyncSend异步
    // 参数1:topic名 参数2：Object，数据
    SendResult sendResult = rocketMQTemplate.syncSend("test", "我是一个简单的消息");
    // 拿到消息的发送状态
    System.out.println(sendResult.getSendStatus());
    // 拿到消息的id
    System.out.println(sendResult.getMsgId());
}

spring:
  application:
    name: rocketmq-consumer
rocketmq:
  name-server: 127.0.0.1:9876
#    consumer：
#        group: aaa-group 不需要写，一个项目一般有很多消费者组

package com.test.listener;
 
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
 
@Component
@RocketMQMessageListener(topic = "test", consumerGroup = "test-group", messageModel = MessageModel.CLUSTERING)
public class SimpleMsgListener implements RocketMQListener<String> {
 
    /**
     * 消费消息的方法
     * 
     * @param message 消息内容，类型和上面的泛型一致。如果泛型指定了固定的类型，消息体就是我们的参数
     */
    @Override
    public void onMessage(String message) {
        System.out.println(message);
    }
}

@Component
@RocketMQMessageListener(topic = "test", consumerGroup = "test-group", messageModel = MessageModel.CLUSTERING)
public class SimpleMsgListener implements RocketMQListener<MessageExt> {
 
    /**
     * 消费消息的方法
     * 
     * @param message 消息内容，类型和上面的泛型一致。如果泛型指定了固定的类型，消息体就是我们的参数
     */
    @Override
    public void onMessage(MessageExt message) {
        System.out.println(new String(message.getBody()));
    }
}

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class FileUploadedEvent {

    /**
     * 全局唯一事件 ID，用于幂等和排障。
     */
    private String eventId;

    /**
     * 文件业务 ID。
     */
    private Long fileId;

    private Long userId;

    private String fileName;

    /**
     * 事件结构版本，便于以后兼容升级。
     */
    private Integer schemaVersion;

    private LocalDateTime occurredAt;
}

eventId
businessId
eventType
schemaVersion
occurredAt
traceId
业务数据

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

/**
 * 文件事件生产者
 * 负责发送文件相关事件到RocketMQ消息队列
 * Topic：file-event-topic，TAG：FILE_UPLOADED 区分消息业务类型
 */
@Service                    // ① 将当前类注册为Spring容器Bean，可被@Autowired注入使用
@RequiredArgsConstructor    // ② Lombok注解：生成仅包含final修饰字段的构造器，实现构造器注入（推荐，优于@Autowired字段注入）
@Slf4j                       // ③ Lombok注解：自动生成log日志对象，无需手动声明Logger
public class FileEventProducer {

    // 常量定义Topic名称，统一管理，避免硬编码散落在代码各处
    private static final String TOPIC = "file-event-topic";
    // 消息TAG：用于同一Topic下过滤不同业务消息（文件上传事件标记）
    private static final String TAG_FILE_UPLOAD = "FILE_UPLOADED";

    // RocketMQ操作模板类，SpringRocketMQ自动装配，封装发送、事务、批量等API
    private final RocketMQTemplate rocketMQTemplate;

    /**
     * 发送【文件已上传】事件消息
     * @param event 文件上传事件实体，承载业务数据
     */
    public void sendFileUploaded(FileUploadedEvent event) {
        // 1. 组装消息目的地：格式规则 Topic:TAG，RocketMQ标准分隔符
        String destination = TOPIC + ":" + TAG_FILE_UPLOAD;

        // 2. 构建Spring标准Message消息对象
        // withPayload(event)：设置消息体，真正要投递的业务实体FileUploadedEvent
        Message<FileUploadedEvent> message = MessageBuilder.withPayload(event)
                // RocketMQ消息KEY：消息唯一业务标识，用于消息查询、日志追踪、死信定位
                把事件唯一 ID 设置为 RocketMQ 消息的 Key，主要用于查询、排查和追踪消息。
                两个参数一个，键值对，一个是键，一个是值
                .setHeader(RocketMQHeaders.KEYS, event.getEventId())
                // 自定义消息头1：文件ID，消费者可直接从Header快速获取，无需解析消息体
                .setHeader("fileId", event.getFileId())
                // 自定义消息头2：消息数据结构版本，用于后续消息结构兼容、灰度升级
                .setHeader("schemaVersion", 1)
                // 完成消息构建，返回Message实例
                .build();

        // 3. 同步发送消息 syncSend：阻塞等待Broker返回发送结果，可靠性最高
        SendResult result = rocketMQTemplate.syncSend(destination, message);

        // 4. 打印发送成功日志，埋点关键业务字段，方便线上排查问题
        log.info(
                "文件上传事件发送成功, eventId={}, fileId={}, msgId={}, sendStatus={}",
                event.getEventId(),       // 业务事件唯一ID（自定义业务主键）
                event.getFileId(),        // 文件业务ID
                result.getMsgId(),        // RocketMQ服务端生成的全局唯一消息ID
                result.getSendStatus()    // 发送状态：SEND_OK 代表投递成功
        );
    }
}

try {
    rocketMQTemplate.syncSend(destination, message);
} catch (Exception e) {
    log.error(
            "消息发送失败, eventId={}, destination={}",
            event.getEventId(),
            destination,
            e
    );

    throw new MessageSendException(
            "发送文件事件失败",
            e
    );
}

// 发送异步消息，发送完以后会有一个异步通知
rocketMQTemplate.asyncSend(
        "notification-topic:EMAIL", // ① 目的地topic
        event,  // ② 消息内容
        new SendCallback() {    // ③ 回调对象（重点！）
			// 成功时的处理
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info(
                        "异步消息发送成功, eventId={}, msgId={}",
                        event.getEventId(),
                        sendResult.getMsgId()
                );
            }
		 // 失败时的处理
            @Override
            public void onException(Throwable throwable) {
                log.error(
                        "异步消息发送失败, eventId={}",
                        event.getEventId(),
                        throwable
                );
            }
        }
);

// 发送单向消息，没有返回值和结果
rocketMQTemplate.sendOneWay(
        "log-topic:ACCESS_LOG",
        accessLogEvent
);

Message<FileUploadedEvent> message =
        MessageBuilder.withPayload(event)
                .setHeader(
                        RocketMQHeaders.KEYS,
                        event.getEventId()
                )
                .setHeader("fileId", event.getFileId())
                .setHeader("schemaVersion", 1)
                .build();

rocketMQTemplate.syncSend(
        "file-event-topic:FILE_UPLOADED",
        message
);

/**
 * 测试延迟消息
 *
 * @throws Exception
 */
@Test
public void testDelay() throws Exception {
    // 构建消息对象
    Message<String> message = MessageBuilder.withPayload("我是一个延迟消息").build();
    // 发送一个延时消息，延迟等级为4级，也就是30s后被监听消费
    // 参数3：连接MQ的超时时间 参数4：延迟等级
    SendResult sendResult = rocketMQTemplate.syncSend("test", message, 2000, 4);
    System.out.println(sendResult.getSendStatus());
}

 rocketMQTemplate.syncSendDelayTimeSeconds(
        "topic",
        message,
        10
);

@Slf4j
@Component
// 绑定事务生产者组，一组生产者共用一个监听器
@RocketMQTransactionListener(txProducerGroup = "file-tx-producer-group")
public class FileTxListener implements RocketMQLocalTransactionListener {

    @Autowired
    private FileMapper fileMapper; // 本地数据库Mapper

    // 阶段2：半消息发送成功后执行本地事务
    @Override
    @Transactional(rollbackFor = Exception.class)
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        FileUploadedEvent event = (FileUploadedEvent) msg.getPayload();
        String eventId = (String) msg.getHeaders().get(RocketMQHeaders.KEYS);
        try {
            // 本地数据库事务：写入文件上传记录
            FileRecord record = new FileRecord();
            record.setEventId(eventId);
            record.setFileId(event.getFileId());
            fileMapper.insert(record);
            // 本地事务成功，通知Broker提交消息
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("本地文件事务失败 eventId:{}", eventId, e);
            // DB失败，通知Broker丢弃消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    // 阶段4：Broker回查时调用，校验本地事务是否存在
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        String eventId = (String) msg.getHeaders().get(RocketMQHeaders.KEYS);
        // 根据eventId查询数据库是否存在这条上传记录
        FileRecord record = fileMapper.selectByEventId(eventId);
        if (record != null) {
            // DB存在记录 → 本地事务成功，提交消息
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            // DB无记录 → 本地事务回滚，丢弃消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class FileEventTxProducer {
    private static final String TOPIC = "file-event-topic";
    private final RocketMQTemplate rocketMQTemplate;

    public void sendTxFileUpload(FileUploadedEvent event) {
        String dest = TOPIC + ":FILE_UPLOADED";
        // 构建消息，eventId作为KEY存入索引
        Message<FileUploadedEvent> message = MessageBuilder.withPayload(event)
                .setHeader(RocketMQHeaders.KEYS, event.getEventId())
                .setHeader("fileId", event.getFileId())
                .setHeader("schemaVersion", 1)
                .build();
        // 发送事务消息，第二个参数为自定义业务参数（可传null）
        rocketMQTemplate.sendMessageInTransaction(dest, message, null);
    }
}

Queue 0 → Consumer A
Queue 1 → Consumer B
Queue 2 → Consumer A
Queue 3 → Consumer B

@Component
@Slf4j
@RocketMQMessageListener(
        topic = "file-event-topic",              // ① 订阅的主题
        selectorExpression = "FILE_UPLOADED",    // ② 过滤标签，只消费带 `FILE_UPLOADED` Tag 的消息。
        consumerGroup = "networkdisk-ai-file-consumer", // ③ 消费组，消费者组名称，用于管理消费者集群。
        messageModel = MessageModel.CLUSTERING,  // ④ 消费模式，集群消费，组内多个消费者**负载均衡**消费消息
        consumeMode = ConsumeMode.CONCURRENTLY   // ⑤ 消费并发模式，多线程并发消费，**默认推荐**，性能高
)
//实现 RocketMQ 的监听接口，指定反序列化类型。
public class FileUploadedConsumer
        implements RocketMQListener<FileUploadedEvent> {

    @Override
    public void onMessage(FileUploadedEvent event) {
        log.info(
                "开始处理文件上传事件, eventId={}, fileId={}",
                event.getEventId(),
                event.getFileId()
        );

        // 执行文本解析、向量化或其他异步任务
        processFile(event);

        log.info(
                "文件上传事件处理成功, eventId={}, fileId={}",
                event.getEventId(),
                event.getFileId()
        );
    }

    private void processFile(FileUploadedEvent event) {
        // 业务逻辑
    }
}

┌─────────────────────────────────────────────────────────────────┐
│                    RocketMQ 消费流程                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  Broker（RocketMQ 服务端）                                     │
│       │                                                        │
│       │ 1. 推送消息给消费者                                    │
│       ▼                                                        │
│  ┌─────────────────────────────────────────────────────────────┐│
│  │ FileUploadedConsumer（你的消费端）                         ││
│  │                                                             ││
│  │ 2. @RocketMQMessageListener 启动监听                       ││
│  │    ├── topic = "file-event-topic"                         ││
│  │    ├── selectorExpression = "FILE_UPLOADED"               ││
│  │    └── consumerGroup = "networkdisk-ai-file-consumer"     ││
│  │                                                             ││
│  │ 3. 拉取到消息（JSON 格式）                                 ││
│  │    {                                                       ││
│  │      "eventId": "evt-001",                                ││
│  │      "fileId": 12345,                                     ││
│  │      "fileName": "report.pdf"                             ││
│  │    }                                                       ││
│  │                                                             ││
│  │ 4. 自动反序列化成 FileUploadedEvent 对象                  ││
│  │                                                             ││
│  │ 5. onMessage(event) 被调用                                ││
│  │    ├── 日志：开始处理                                     ││
│  │    ├── processFile(event)                                 ││
│  │    │   ├── 下载文件                                       ││
│  │    │   ├── 解析文本                                       ││
│  │    │   ├── 向量化                                         ││
│  │    │   └── 存入向量数据库                                 ││
│  │    └── 日志：处理成功                                     ││
│  │                                                             ││
│  │ 6. 方法正常返回 → 提交消费位点（Offset）                  ││
│  │                                                             ││
│  └─────────────────────────────────────────────────────────────┘│
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Message 1 → Consumer A
          → Consumer B
          → Consumer C

messageModel = MessageModel.BROADCASTING

ConsumerGroup: networkdisk-ai-file-consumer

Queue 0 offset = 1024
Queue 1 offset = 986
Queue 2 offset = 1102
Queue 3 offset = 1007

原来：
Consumer A → Queue 0、1、2、3

新增 Consumer B 后：
Consumer A → Queue 0、1
Consumer B → Queue 2、3

Producer 发送阶段
        ↓
Broker 存储阶段
        ↓
Consumer 消费阶段

业务数据库提交成功，但消息没发出去
消息已到 Broker，但 Producer 响应超时
Producer 进程在发送前宕机
网络闪断
发送失败重试后仍然失败

saveOrder();
rocketMQTemplate.syncSend(...);

订单提交成功
      ↓
应用宕机
      ↓
消息没有发送

本地事务：
    INSERT order
    INSERT outbox_event
提交成功

outbox_event

id
event_id
topic
tag
payload
status
retry_count
next_retry_time
created_at

发送确认
+ 有限重试
+ 本地记录
+ 定时补偿
+ 监控告警

同步刷盘还是异步刷盘
主从同步复制还是异步复制
Broker 部署副本数
磁盘健康状况
故障切换能力

同步刷盘：
    防止当前 Broker 进程或系统异常导致 Page Cache 数据丢失

同步复制：
    防止 Master 整机或磁盘损坏后消息没有副本

错误顺序：
Consumer 确认消息成功
      ↓
执行业务
      ↓
业务失败

执行业务
      ↓
业务成功
      ↓
返回消费成功

@Override
public void onMessage(FileUploadedEvent event) {
    try {
        fileIndexService.buildIndex(event.getFileId());
    } catch (Exception e) {
        log.error(
                "构建文件索引失败, eventId={}, fileId={}",
                event.getEventId(),
                event.getFileId(),
                e
        );

        // 不要吞掉异常，否则框架可能认为消费成功
        throw e;
    }
}

@Override
public void onMessage(FileUploadedEvent event) {
    try {
        fileIndexService.buildIndex(event.getFileId());
    } catch (Exception e) {
        log.error("处理失败", e);

        // 错误：异常被吞掉，方法正常结束
    }
}

Producer 发送成功，但响应超时，Producer 再次发送
Consumer 处理成功，但确认消息前宕机
网络异常导致 ACK 丢失
消费超时
Rebalance
人工重置 Offset
重试消息重新投递

CREATE TABLE mq_consume_record (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    event_id VARCHAR(64) NOT NULL,
    consumer_group VARCHAR(128) NOT NULL,
    consumed_at DATETIME NOT NULL,
    UNIQUE KEY uk_event_consumer (
        event_id,
        consumer_group
    )
);

@Transactional
public void consume(FileUploadedEvent event) {
    boolean inserted = consumeRecordMapper.insertIgnore(
            event.getEventId(),
            "networkdisk-ai-file-consumer"
    );

    if (!inserted) {
        // 已消费过，直接返回成功
        return;
    }

    fileIndexService.buildIndex(event.getFileId());
}

UPDATE orders
SET status = 'PAID'
WHERE id = #{orderId}
  AND status = 'CREATED';

SET mq:consume:{consumerGroup}:{eventId} 1 NX EX 86400

MQ 投递语义
+ 业务唯一标识
+ 数据库唯一约束
+ 状态机
+ 本地事务

// 失败的情况重发3次（同步）
producer.setRetryTimesWhenSendFailed(3);
// 失败的情况重发3次（异步）
producer.setRetryTimesWhenSendAsyncFailed(3);
// 消息在1S内没有发送成功，就会重试
producer.send(msg, 1000);

@Test
public void retryProducer() throws Exception {
    DefaultMQProducer producer = new DefaultMQProducer("retry-producer-group");
    producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
    producer.start();
    // 如果发送失败要重试几次（同步），不设置默认值是2
    producer.setRetryTimesWhenSendFailed(3);
    // 如果发送失败要重试几次（异步）
//        producer.setRetryTimesWhenSendAsyncFailed(3);
    String key = UUID.randomUUID().toString();
    System.out.println(key);
    Message message = new Message("retryTopic", "vip1", key, "我是vip666的文章".getBytes());
    producer.send(message);
    System.out.println("发送成功");
    producer.shutdown();
}

// 消费失败，重试几次
consumer.setMaxReconsumeTimes(5);

/**
 * 方案一：独立死信队列消费者
 * 原理：消息重试耗尽阈值后自动进入 %DLQ%xxx 死信Topic；单独起消费者监听死信Topic统一兜底
 * 适用：业务不拦截重试，依赖RocketMQ自动转死信，单独程序人工巡检失败消息
 * @throws Exception
 */
@Test
public void retryDeadConsumer() throws Exception {
    // 创建Push消费者，分组：retry-dead-consumer-group（死信专用分组，与业务消费组隔离）
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-dead-consumer-group");
    // 指定NameServer地址，连接MQ服务
    consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
    // 订阅死信Topic：固定前缀%DLQ% + 原业务消费组名，*代表接收所有TAG
    consumer.subscribe("%DLQ%retry-consumer-group", "*");
    // 注册并发消息监听器
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            // 批量消息取第一条（测试Demo简化写法，生产需循环遍历所有msg）
            MessageExt messageExt = msgs.get(0);
            System.out.println("死信消息接收时间：" + new Date());
            // 打印原始消息体
            System.out.println("消息内容：" + new String(messageExt.getBody()));
            // 死信处理逻辑：落地数据库/文件记录、推送告警通知运维人工排查
            System.out.println("记录到特别的位置 文件 mysql 通知人工处理");
            // 返回成功：这条死信处理完毕，Broker删除，不再重复投递
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    // 启动消费者
    consumer.start();
    // 阻塞主线程，防止测试方法直接结束
    System.in.read();
}

/**
 * 方案二：业务消费端自行控制重试阈值，超限直接归档，不丢去死信队列
 * 原理：消费时读取已重试次数reconsumeTimes，达到上限主动签收成功，停止重试
 * 适用：简单业务，不需要单独死信消费者，代码内直接兜底失败消息
 * @throws Exception
 */
@Test
public void retryConsumer2() throws Exception {
    // 业务正常消费分组
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-consumer-group");
    consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
    // 订阅业务正常Topic
    consumer.subscribe("retryTopic", "*");
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            MessageExt messageExt = msgs.get(0);
            System.out.println("消息接收时间：" + new Date());
            try {
                // 执行业务数据库逻辑（模拟报错：除零异常）
                handleDb();
            } catch (Exception e) {
                // 获取当前消息已经重试多少次
                int reconsumeTimes = messageExt.getReconsumeTimes();
                // 阈值：重试≥3次不再重试
                if (reconsumeTimes >= 3) {
                    // 超限：落地日志、入库、发告警人工处理
                    System.out.println("重试次数过大，归档记录，停止重试");
                    // 返回成功，Broker清除消息，不会进入死信队列
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                // 不足3次，告知Broker稍后重试
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            // 业务无异常，正常签收
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.in.read();
}

/** 模拟业务数据库异常 */
private void handleDb() {
    // 10/0 抛出算术异常，触发消费失败重试逻辑
    int i = 10 / 0;
}

@Test
public void testDeadMsgProducer() throws Exception {
    // 生产者分组
    DefaultMQProducer producer = new DefaultMQProducer("dead-group");
    producer.setNamesrvAddr("localhost:9876");
    producer.start();
    // 创建消息：Topic=dead-topic，消息体字符串字节数组
    Message message = new Message("dead-topic", "我是一个死信消息".getBytes());
    // 同步发送消息
    producer.send(message);
    // 关闭生产者释放资源
    producer.shutdown();
}

@Test
@Test
public void testDeadMsgConsumer() throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("dead-group");
    consumer.setNamesrvAddr("localhost:9876");
    consumer.subscribe("dead-topic", "*");
    // 全局配置：单条消息最大重试2次
    // 重试2次后依旧失败，RocketMQ自动将消息转入死信队列 %DLQ%dead-group
    consumer.setMaxReconsumeTimes(2);
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.println("收到业务消息：" + msgs);
            // 固定返回重试，每次都会重新投递，耗尽次数自动进DLQ
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    });
    consumer.start();
    System.in.read();
}

/**
 * 监听死信队列专用消费者
 * 底层规则：消息重试超过maxReconsumeTimes阈值，自动移入 %DLQ%消费组名 死信Topic
 * 普通业务消费者不会订阅DLQ，必须单独写消费者监听处理积压失败消息
 */
@Test
public void testDeadMq() throws  Exception{
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("dead-group");
    consumer.setNamesrvAddr("localhost:9876");
    // 订阅死信Topic，格式固定：%DLQ% + 原业务消费者分组名称
    consumer.subscribe("%DLQ%dead-group", "*");
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.println("捕获死信消息：" + msgs);
            // 人工处理、归档、修复后签收，消息彻底清理
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    // 阻塞等待消息
    System.in.read();
}

订单创建
   ↓
订单支付
   ↓
订单发货
   ↓
订单完成

/**
 * syncSendOrderly：同步有序发送API
 * 语法：syncSendOrderly(topicTag, payload, hashKey)
 * 原理：依靠第三个参数hashKey做哈希取模路由，相同hashKey消息固定投递到同一个消息队列
 * 作用：保证同一订单的所有状态变更消息进入同一队列，消费者顺序消费，避免状态错乱
 */
rocketMQTemplate.syncSendOrderly(
        // 目的地格式 Topic:Tag，Tag过滤只接收订单状态变更消息
        "order-event-topic:ORDER_STATUS_CHANGED",
        // 消息体：订单变更事件实体，存放完整订单业务数据
        event,
        // 路由哈希键：orderId
        // 相同orderId哈希值一致，会分发到同一个Queue，实现局部有序
        event.getOrderId().toString()
);

import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Component
// RocketMQ消费者注解，声明监听配置
@RocketMQMessageListener(
        topic = "order-event-topic",                // 监听的主主题
        selectorExpression = "ORDER_STATUS_CHANGED",// Tag过滤表达式，只消费该标签消息
        consumerGroup = "order-state-consumer",     // 消费者分组，同组负载均衡，不同组各自消费全量消息
        consumeMode = ConsumeMode.ORDERLY           // 消费模式：ORDERLY 顺序消费；CONCURRENT 并发消费
)
/**
 * 订单状态变更有序消费者
 * 原理：生产者用orderId路由到同一队列，消费者单线程串行消费单个队列消息
 * 作用：严格按消息发送顺序更新订单状态，防止出现「先收到已完成、后收到待支付」错乱
 */
public class OrderStateConsumer
        implements RocketMQListener<OrderStatusChangedEvent> {

    // 订单业务服务，处理状态更新逻辑
    private final OrderService orderService;

    // 构造器注入
    public OrderStateConsumer(OrderService orderService) {
        this.orderService = orderService;
    }

    /**
     * 消息消费入口方法
     * @param event 自动反序列化后的订单变更事件
     * 有序消费约束：上一条消息处理完成、方法执行完毕，才会拉取下一条同队列消息
     */
    @Override
    public void onMessage(OrderStatusChangedEvent event) {
        // 执行订单状态变更业务逻辑
        orderService.changeStatus(
                event.getOrderId(),
                event.getTargetStatus()
        );
    }
}

同一个 Queue 内有序，不等于整个 Topic 全局有序
同一业务键必须使用相同顺序键
生产者并发发送同一业务键时，业务侧也要保证发送先后关系
顺序消费失败会阻塞当前 Queue 后续消息
Queue 越少，全局顺序越强，但并发能力越低

订单创建后 30 分钟未支付自动取消
文件删除后 7 天彻底清理
验证码发送 1 分钟后检查状态
任务失败后延迟 10 分钟重试
会议开始前发送提醒

rocketMQTemplate.syncSend(
        "order-delay-topic:CHECK_PAYMENT",
        message,
        3000,
        delayLevel
);

定时任务：
    按固定时间或周期扫描
    例如每天凌晨清理数据

延时消息：
    每个业务事件产生自己的未来触发时间
    例如每个订单创建后 30 分钟检查一次

@Transactional
public void cancelIfUnpaid(Long orderId) {
    Order order = orderMapper.selectById(orderId);

    if (order == null) {
        return;
    }

    if (!OrderStatus.CREATED.equals(order.getStatus())) {
        // 已支付、已取消或已完成，不再处理
        return;
    }

    orderMapper.cancelOrder(orderId);
}

数据库事务成功，但消息没发送
消息发送成功，但数据库事务失败

1. Producer 发送 Half Message
2. Broker 保存消息，但暂不投递给 Consumer
3. Broker 返回 Half Message 成功
4. Producer 执行本地事务
5. Producer 向 Broker 提交 Commit 或 Rollback
6. Commit 后消息才对 Consumer 可见
7. 状态未知时，Broker 回查 Producer

/**
 * 订单事务消息生产者
 * 作用：发送事务半消息，绑定本地创建订单DB事务，保证「建订单库写入」和「MQ消息投递」原子一致
 * 原理：2PC事务消息，先发半消息，再执行本地事务，根据本地事务结果决定消息对外可见/丢弃
 */
@Service
@RequiredArgsConstructor // Lombok：final字段构造器注入rocketMQTemplate
public class OrderTransactionProducer {

    // RocketMQ操作模板，封装事务消息发送能力
    private final RocketMQTemplate rocketMQTemplate;

    /**
     * 创建订单入口，发送事务消息
     * @param command 创建订单请求载体，携带订单所有业务参数
     */
    public void createOrder(CreateOrderCommand command) {
        // 1、构建标准Spring Message消息
        Message<CreateOrderCommand> message =
                MessageBuilder.withPayload(command) // Payload：消息主体，消费者接收解析
                        // MQ内置KEY索引字段：requestId作为全局唯一标识
                        // 用途1：写入IndexFile哈希索引，控制台可按requestId检索消息
                        // 用途2：事务回查时唯一查询条件，精准匹配本地订单记录
                        .setHeader(
                                RocketMQHeaders.KEYS,
                                command.getRequestId()
                        )
                        .build();

        // 2、发送事务消息核心方法
        // 参数1：目的地 Topic:Tag 格式，ORDER_CREATED标记「订单创建」事件
        // 参数2：完整消息对象
        // 参数3：自定义透传参数，会原样传入 executeLocalTransaction 的argument参数
        // 返回值：TransactionSendResult 封装本次本地事务执行结果
        TransactionSendResult result =
                rocketMQTemplate.sendMessageInTransaction(
                        "order-transaction-topic:ORDER_CREATED",
                        message,
                        command
                );

        // 3、判断本地事务执行状态，事务回滚则主动抛出业务异常
        if (result.getLocalTransactionState()
                == LocalTransactionState.ROLLBACK_MESSAGE) {
            throw new IllegalStateException(
                    "订单事务消息执行失败"
            );
        }
    }
}

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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.messaging.Message;

/**
 * RocketMQ事务消息监听器
 * 两个核心方法：
 * 1. executeLocalTransaction：半消息发送成功后同步执行本地DB事务
 * 2. checkLocalTransaction：Broker定时回查，生产者宕机/网络丢失时兜底校验事务状态
 */
@RocketMQTransactionListener // 标识事务监听，自动绑定对应事务生产者组
@RequiredArgsConstructor
@Slf4j // 自动生成log日志对象
public class OrderTransactionListener
        implements RocketMQLocalTransactionListener {

    // 订单业务服务：封装创建订单数据库逻辑
    private final OrderService orderService;
    // 订单Mapper：回查时查询数据库订单记录
    private final OrderMapper orderMapper;

    /**
     * 阶段1：半消息持久化Broker后，同步回调执行本地事务
     * @param message MQ原始消息对象
     * @param argument sendMessageInTransaction第三个参数，透传CreateOrderCommand
     * @return 事务状态：COMMIT提交消息 / ROLLBACK丢弃消息 / UNKNOWN状态未知等待回查
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(
            Message message,
            Object argument
    ) {
        // 取出发送时透传的订单创建参数
        CreateOrderCommand command =
                (CreateOrderCommand) argument;

        try {
            // 执行本地数据库事务：生成订单、扣库存、落订单表等
            orderService.createOrder(command);
            // 本地事务成功，通知Broker把半消息转正，投递到业务Topic供消费者消费
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error(
                    "订单本地事务执行失败, requestId={}",
                    command.getRequestId(),
                    e
            );
            // DB事务异常，通知Broker删除半消息，下游永远收不到该订单消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 阶段2：事务补偿回查接口
     * 触发条件：生产者发送半消息后宕机/网络断连，Broker长期未收到commit/rollback指令
     * 逻辑：通过消息头里唯一requestId查询数据库，判断订单事务是否实际创建成功
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(
            Message message
    ) {
        // 从内置KEY头取出全局唯一请求ID
        String requestId = String.valueOf(
                message.getHeaders().get(
                        RocketMQHeaders.KEYS
                )
        );

        // 根据requestId查询订单表，判断本地事务是否执行成功
        Order order = orderMapper.selectByRequestId(
                requestId
        );

        if (order != null) {
            // 数据库存在订单记录 → 本地事务已完成，通知Broker提交消息
            return RocketMQLocalTransactionState.COMMIT;
        }
        // 无订单记录 → 本地事务回滚/未执行，Broker删除半消息
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}

根据 requestId 查询订单是否存在
根据 paymentNo 查询支付记录是否成功
根据 eventId 查询本地事务日志

创建订单后发送订单事件
支付成功后发送支付结果
账户变更后发送记账事件
文件状态变化后触发异步处理

生产速度：每秒 10,000 条
消费速度：每秒 6,000 条
净积压：每秒 4,000 条

消费者实例太少
Topic Queue 数量不足
单条消息处理耗时过长
数据库慢 SQL
第三方接口超时
消费者频繁抛异常重试
顺序消息被某条异常消息阻塞
消息体过大
消费线程池配置不合理
消费者发生 Full GC
Broker、网络或磁盘性能异常

生产 TPS
消费 TPS
消费延迟
最大积压量
失败重试数
消费者实例数
数据库响应时间
Broker 磁盘利用率

Topic 只有 4 个 Queue
Consumer Group 启动 20 个实例

逐条写数据库：
1000 条消息 → 1000 次数据库请求

批量写数据库：
1000 条消息 → 10 次批量请求

根据下游承载能力设置消费并发
根据消息大小控制本地缓存
根据平均耗时评估线程数
对第三方调用增加超时、限流和熔断
对数据库写入使用批量和索引优化

1 个 NameServer
1 个 Broker

3 个 NameServer

Broker Group A
    Broker A 主节点
    Broker A 副本

Broker Group B
    Broker B 主节点
    Broker B 副本

多个 Producer 实例
多个 Consumer 实例
Dashboard
Prometheus + Grafana
集中日志系统

分散 Topic Queue
提高集群吞吐量
避免单节点成为容量瓶颈
降低单机故障影响
实现水平扩展

Master 写入
    ↓
等待 Slave 确认
    ↓
向 Producer 返回成功

Master 写入
    ↓
立即返回成功
    ↓
后台复制给 Slave

Topic Queue 是否迁移或新增
Producer 路由是否刷新
Consumer 是否发生 Rebalance
顺序键映射是否变化
磁盘容量与水位
网络带宽
消息副本
监控和告警

消费成功：
    更新消费组 Offset

物理删除：
    后台按文件保留和磁盘策略清理 CommitLog

Broker 是否存活
NameServer 是否存活
Producer 发送成功率
Producer 发送耗时
Consumer 消费 TPS
Consumer 失败率
Consumer 重试次数
消息积压量
消费延迟
死信消息数量
Broker 磁盘使用率
CommitLog 写入耗时
主从复制延迟
JVM 堆、GC 和线程数
网络连接数

eventId
businessId
messageId
traceId

log.info(
        "消息发送成功, eventId={}, businessId={}, msgId={}, topic={}",
        eventId,
        businessId,
        sendResult.getMsgId(),
        topic
);

log.info(
        "消息消费开始, eventId={}, businessId={}, consumerGroup={}",
        eventId,
        businessId,
        consumerGroup
);

1. 业务数据库是否提交成功
2. Producer 是否生成事件
3. Producer 是否发送成功
4. Broker 是否能按 Key 查到消息
5. 消费组是否订阅正确 Topic 和 Tag
6. 消费组 Offset 是否推进
7. 消息是否进入重试队列
8. 消息是否进入死信队列
9. Consumer 是否异常或频繁重启
10. 下游数据库、Redis、ES 是否异常

file-event-topic:FILE_UPLOADED

topic = files-event-topic
selectorExpression = FILE_UPLOAD

{
  "method": "updateFile",
  "args": [...]
}

{
  "eventId": "evt-001",
  "eventType": "FILE_UPLOADED",
  "fileId": 1001,
  "userId": 2001,
  "occurredAt": "2026-07-11T20:30:00",
  "schemaVersion": 1
}

{
  "orderId": 10001
}

{
  "orderId": 10001,
  "userId": 20001,
  "amount": 99.00,
  "status": "PAID"
}

发送业务主键
+ 消费者真正需要的稳定字段
+ 事件发生时状态

新增可选字段
保持旧字段语义
使用 schemaVersion
灰度升级 Consumer，再升级 Producer
避免直接修改字段类型

Topic:
networkdisk-file-event
networkdisk-user-event
networkdisk-share-event

Consumer Group:
networkdisk-ai-file-index-consumer
networkdisk-search-file-sync-consumer
networkdisk-notification-file-consumer

test-topic
topic1
consumer-group
default-group

fileId
objectKey
bucket
metadata

面向业务消息和事件驱动
支持事务消息
支持顺序消息
支持延时消息
支持消费重试和死信
Topic、Queue、ConsumerGroup 模型清晰
在 Java 和微服务业务系统中使用广泛

日志采集
CDC
大数据流处理
埋点
实时计算
长期事件流

复杂路由
中小规模业务消息
工作队列
灵活确认机制

消息模型
事务和延时需求
顺序要求
吞吐量
运维经验
团队技术栈
云服务支持
现有基础设施

生产阶段：
同步发送、失败重试、本地消息表、事务消息、补偿任务

Broker 阶段：
持久化、同步刷盘、主从复制、高可用部署

消费阶段：
业务成功后再确认、失败重试、死信告警、消费幂等

业务服务产生事件
        ↓
Producer 从 NameServer 查询路由
        ↓
Producer 选择 Topic 下的 MessageQueue
        ↓
Broker 将完整消息顺序写入 CommitLog
        ↓
Broker 构建 ConsumeQueue 和 Key 索引
        ↓
Consumer Group 获取并分配 Queue
        ↓
Consumer 拉取消息并执行业务
        ↓
成功后推进消费 Offset
        ↓
失败则进入重试流程
        ↓
超过最大次数进入死信队列

异步解耦与数据最终一致性
可靠投递与重复消费
消费重试与业务幂等
MessageQueue 数量与消费并行度
局部顺序与系统吞吐量
同步刷盘与性能
主从复制与可用性
事务消息与本地事务回查
消息积压与下游承载能力

规范的 Topic 和 Consumer Group
稳定的消息 Schema
全局唯一事件 ID
Producer 发送确认
本地消息表或事务消息
Broker 多副本和持久化
Consumer 业务幂等
失败重试与死信处理
积压和延迟监控
定时对账与业务补偿
完整的 TraceId、Key 和 MessageId 日志