>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"

实现原理：

  1. 消息写入 CommitLog

  2. 如果是延迟消息，替换 Topic 为 SCHEDULE_TOPIC_XX

  3. 调度线程按级别将消息从 SCHEDULE_TOPIC 投递回原 Topic

  4. 消费者消费

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


# rocketmq # java # spring boot # rabbitmq
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

当 AI 修图师装上“万能插头“：MCP 如何改变修图工作流

avatar AI编程社区
cover

2026年8月开发者实战指南：ChatGPT Plus / Pro + Codex 如何协同完成需求分析、编码、测试、调试与代码审查

avatar AI编程社区
cover

Claude API + Python实战：结构化输出后，如何拦住“看起来正确”的JSON

avatar AI编程社区