【RocketMQ】阿里万亿级消息中间件MQ保姆级教程
>MQ主要解决应用解耦、异步消息、流量削峰、日志处理等问题
目录
一、RocketMQ概述
1.1 RocketMQ是什么
1.2 核心架构和概念
1.3 部署架构模式
二、快速上手
2.1 Docker部署
2.2 三中发送方式
2.3 顺序消息
2.4 事务消息
2.5 延迟消息
2.6 消息过滤
三、源码分析
3.1 Broker 启动流程
3.2 消息发送核心流程
3.3 消息消费核心流程
一、RocketMQ概述
1.1 RocketMQ是什么
Apache RocketMQ 是阿里巴巴开源的一款分布式、队列模型的消息中间件，后捐赠给 Apache 基金会并成为顶级项目（TLP）。
各mq之间的比较：
| 维度 | RocketMQ | Kafka | RabbitMQ |
|---|---|---|---|
| 定位 | 消息中间件（业务消息） | 日志流式处理 | 传统消息中间件 |
| 协议 | 自定义协议 | 自定义协议 | AMQP |
| 顺序消息 | ✅ 天然支持 | ⚠️ 仅分区有序 | ⚠️ 较弱 |
| 事务消息 | ✅ 完整方案 | ❌ 不支持 | ⚠️ 弱支持 |
| 延迟消息 | ✅ 内置 18 个级别 | ❌ 需外部方案 | ⚠️ 插件 |
| 消息堆积 | ✅ 亿级 | ✅ 强 | ❌ 弱 |
| 吞吐量 | 百万级 | 百万级 | 万级 |
| 运维复杂度 | 中 | 中 | 高 |
选型建议：
业务消息（订单、交易）→RocketMQ（推荐）
大数据日志流 → Kafka
传统企业集成、复杂路由 → RabbitMQ
1.2 核心架构和概念
┌──────────────────────────────────────────────┐ │ RocketMQ 集群 │ │ │ │ ┌────────────┐ ┌──────────────────┐ │ │ │ NameServer │ ←──→ │ NameServer（集群）│ │ │ └────────────┘ └──────────────────┘ │ │ ↑ │ │ │ 注册/发现 │ │ ↓ │ │ ┌────────────┐ 主从同步 ┌────────────┐ │ │ │ Broker-A │ ←───────→ │ Broker-A-S │ │ │ └────────────┘ └────────────┘ │ │ ↑ │ │ │ 发送/消费 │ │ ↓ │ │ ┌────────────┐ ┌────────────┐ │ │ │ Producer │ │ Consumer │ │ │ └────────────┘ └────────────┘ │ └──────────────────────────────────────────────┘
| 角色 | 说明 |
|---|---|
| NameServer | 路由注册中心，集群中各角色通过它获取路由信息（无状态） |
| Broker | 消息存储与中转，负责消息的存储、投递、查询、高可用 |
| Producer | 消息生产者，向 Broker 发送消息 |
| Consumer | 消息消费者，从 Broker 拉取消息进行消费 |
| Topic | 消息主题，一类消息的集合 |
| Message Queue | 消息队列，Topic 被切分为多个 Queue 以提高并行度 |
| Tag | 子主题，用于消息过滤 |
| Group | 一类 Producer 或 Consumer 的集合 |
| Offset | 消费位点，记录消费进度 |
1.3 部署架构模式
| 模式 | 说明 |
|---|---|
| 单 Master | 不推荐，仅用于本地调试 |
| 多 Master | 集群无 Slave， Broker 集群 |
| 多 Master 多 Slave（异步） | Master 写成功即返回，Slave 异步复制 |
| 多 Master 多 Slave（同步） | 双写成功才返回，强一致 |
| Dledger 集群 | 基于 Raft 协议的自动选主（4.5+） |
二、快速上手
2.1 Docker部署
docker-compose.yml文件如下，注意IP需要修改为自己的公网IP：
version: '3.8' services: # ==================== 1. NameServer ==================== rmqnamesrv: image: apache/rocketmq:4.9.4 container_name: rmqnamesrv ports: - "9876:9876" environment: - JAVA_OPT_EXT=-server -Xms256m -Xmx256m -Xmn128m command: sh mqnamesrv networks: - rocketmq_net # ==================== 2. Broker Master ==================== rmqbroker-master: image: apache/rocketmq:4.9.4 container_name: rmqbroker-master ports: - "10911:10911" # 业务端口 - "10909:10909" # VIP 端口 (10911-2) - "10912:10912" # HA 端口 environment: - NAMESRV_ADDR=rmqnamesrv:9876 - JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m command: > sh -c " echo 'brokerClusterName=DefaultCluster' > /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'brokerName=broker-a' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'brokerId=0' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'deleteWhen=04' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'fileReservedTime=48' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'brokerRole=ASYNC_MASTER' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'flushDiskType=ASYNC_FLUSH' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'brokerIP1=175.24.167.232' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'brokerIP2=rmqbroker-master' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'autoCreateTopicEnable=true' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'listenPort=10911' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'haListenPort=10912' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && sh mqbroker -n rmqnamesrv:9876 -c /home/rocketmq/rocketmq-4.9.4/conf/broker.conf & tail -f /home/rocketmq/logs/rocketmqlogs/broker.log " depends_on: - rmqnamesrv networks: - rocketmq_net # ==================== 3. Broker Slave ==================== rmqbroker-slave: image: apache/rocketmq:4.9.4 container_name: rmqbroker-slave ports: - "10921:10921" # 业务端口（Slave 独立） - "10919:10919" # VIP 端口（10921-2） - "10922:10922" # HA 端口 environment: - NAMESRV_ADDR=rmqnamesrv:9876 - JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m command: > sh -c " echo 'brokerClusterName=DefaultCluster' > /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'brokerName=broker-a' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'brokerId=1' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'deleteWhen=04' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'fileReservedTime=48' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'brokerRole=SLAVE' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'flushDiskType=ASYNC_FLUSH' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'brokerIP1=175.24.167.232' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'brokerIP2=175.24.167.232' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'autoCreateTopicEnable=true' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'listenPort=10921' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'haListenPort=10922' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && echo 'haMasterAddress=175.24.167.232:10912' >> /home/rocketmq/rocketmq-4.9.4/conf/broker.conf && sh mqbroker -n rmqnamesrv:9876 -c /home/rocketmq/rocketmq-4.9.4/conf/broker.conf & tail -f /home/rocketmq/logs/rocketmqlogs/broker.log " depends_on: - rmqnamesrv - rmqbroker-master networks: - rocketmq_net # ==================== 4. RocketMQ Dashboard ==================== rmqdashboard: image: apacherocketmq/rocketmq-dashboard:1.0.0 container_name: rmqdashboard ports: - "8180:8080" environment: - JAVA_OPTS=-Drocketmq.namesrv.addr=rmqnamesrv:9876 depends_on: - rmqnamesrv - rmqbroker-master - rmqbroker-slave networks: - rocketmq_net networks: rocketmq_net: driver: bridge浏览器访问：ip:8180，出现下面的页面说明mq集群部署成功
2.2 三中发送方式
添加Maven依赖
<dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.2.0</version> </dependency>不同的发送方式
| 方式 | 特点 | 适用场景 |
|---|---|---|
| 同步发送 | 等待 Broker 应答，可靠性高 | 重要通知、订单消息 |
| 异步发送 | 回调通知，性能高 | 对响应时间敏感 |
| 单向发送 | 不关心结果，性能最高 | 日志收集 |
发送结果状态：
| 返回状态 | 含义 |
|---|---|
| SEND_OK | 发送成功 |
| FLUSH_DISK_TIMEOUT | 刷盘超时（异步刷盘模式可能仍成功） |
| FLUSH_SLAVE_TIMEOUT | 主从同步超时（异步复制可能仍成功） |
| SLAVE_NOT_AVAILABLE | 从节点不可用 |
2.3 顺序消息
类型如下：
| 类型 | 描述 | 性能 |
|---|---|---|
| 全局顺序 | 整个 Topic 只有一个 Queue | 差 |
| 分区顺序 | 同一业务 key 投递到同一 Queue | 高 |
2.4 事务消息
流程图：
生产者代码：
消费者代码：
回查代码：
关键设计点
Half 消息： 对消费者不可见
Op 消息： Commit/Rollback 操作消息
反查机制： 解决本地事务执行时间过长的问题
2.5 延迟消息
内置18个延迟级别：
"1s", "5s", "10s", "30s","1m", "2m", "3m", "4m", "5m", "6m", "7m", "8m", "9m", "10m", "20m", "30m", "1h", "2h"
实现原理：
消息写入 CommitLog
如果是延迟消息，替换 Topic 为
SCHEDULE_TOPIC_XX调度线程按级别将消息从 SCHEDULE_TOPIC 投递回原 Topic
消费者消费
2.6 消息过滤
三、源码分析
3.1 Broker 启动流程
BrokerStartup.createBrokerController(args)
→ BrokerController.initialize()
→ 加载配置
→ 创建各种管理器（消息存储、消费者偏移、订阅等）
→ BrokerController.start()
→ 启动 Netty Server
→ 注册到 NameServer
→ 启动各种定时任务
关键线程：
| 线程 | 作用 |
|---|---|
| SendMessageProcessor | 处理消息发送请求 |
| PullMessageProcessor | 处理消息拉取请求 |
| FlushRealTimeService | 异步刷盘 |
| ReputMessageService | 构建 ConsumeQueue/Index |
| HAConnection | 主从同步 |
3.2 消息发送核心流程
DefaultMQProducerImpl.send()
→ MQFaultStrategy.selectOneMessageQueue() // 选队列
→ SendMessageProcessor.processRequest() // Broker 端
→ MessageStore.putMessage() // 存储
→ CommitLog.asyncPutMessage() // 顺序写
→ Broker 返回 SendResult
3.3消息消费核心流程
PullMessageHoldService
→ PullRequestHoldService 拉取请求
→ PullMessageProcessor 处理
→ PullResult 推送给消费者
→ ConsumeMessageConcurrentlyService 消费
→ 更新 Offset