当前位置：首页 > news >正文

RocketMQ

news 来源：原创 2025/7/19 15:54:28

一、引言

Message Queue（消息队列），从字⾯上理解：⾸先它是⼀个队列。FIFO先进先出的数据结构——队列。消息队列就是所谓的存放消息的队列。
消息队列解决的不是存放消息的队列的⽬的，解决的是通信问题。

⽐如以电商订单系统为例，如果各服务之间使⽤同步通信，不仅耗时较久，且过程中受到⽹络波动的影响，不能保证⾼成功率。因此，使⽤异步的通信⽅式对架构进⾏改造。

使⽤异步的通信⽅式对模块间的调⽤进⾏解耦，可以快速的提升系统的吞吐量。上游执⾏完消息的发送业务后⽴即获得结果，下游多个服务订阅到消息后各⾃消费。通过消息队列，屏蔽底层的通信协议，使得解藕和并⾏消费得以实现。

二、RocketMQ介绍

1、RocketMQ的由来

随着使⽤中队列和虚拟主题的增加，阿⾥巴巴团队使⽤的ActiveMQ IO 模块达到了瓶颈。为了尽⼒通过节流、断路器或降级来解决这个问题，但效果不佳。所以开始关注当时流⾏的消息传递解决⽅案Kafka。不幸的是，Kafka ⽆法满⾜要求，尤其是在低延迟和⾼可靠性⽅⾯。在这种情况下，决定发明⼀种新的消息传递引擎来处理更⼴泛的⽤例，从传统的发布/订阅场景到⼤容量实时零丢失交易系统。⽬前 RocketMQ已经开源给Apache基⾦会。如今，已有 100 多家公司在其业务中使⽤开源版本的 RocketMQ。

2、RocketMQ vs. ActiveMQ vs. Kafka

消息产品	ActiveMQ	Kafka	RocketMQ
客户端 SDK	Java 、 .NET 、 C++ 等。	Java 、 Scala等。	Java 、 C++、 Go
协议和规范	推送模型，⽀持 OpenWire 、 STOMP、 AMQP、MQTT、 JMS	拉取模型，⽀持TCP	拉取模型，⽀持 TCP 、 JMS 、 OpenMessaging
订购信息	Exclusive Consumer 或 Exclusive Queues 可以保证排序	确保分区内消息的排序	确保消息的严格排序，并且可以优雅地横向扩展
预定消息	⽀持的	不⽀持	⽀持的
批量消息	不⽀持	⽀持，带有异步⽣产者	⽀持，具有同步模式以避免消息丢失
⼴播消息	⽀持的	不⽀持	⽀持的
消息过滤器	⽀持的	⽀持，可以使⽤KafkaStreams 过滤消息	⽀持的基于SQL92的属性过滤器表达式
服务器触发的重新交付	不⽀持	不⽀持	⽀持的
消息存储	使⽤ JDBC 和⾼性能⽇志⽀持⾮常快速的持久化，例如levelDB、kahaDB	⾼性能⽂件存储	⾼性能和低延迟的⽂件存储
消息追溯	⽀持的	⽀持的偏移量指示	⽀持的时间戳和偏移量两种表示
消息优先级	⽀持的	不⽀持	不⽀持
⾼可⽤性和故障转移	⽀持，取决于存储，如果使⽤levelDB 则需要ZooKeeper服务器	⽀持，需要ZooKeeper服务器	⽀持的主从模型，⽆需其他套件
消息跟踪	不⽀持	不⽀持	⽀持的
配置	默认配置为低级，⽤户需优化配置参数	Kafka 使⽤键值对格式进⾏配置。这些值可以从⽂件或以编程⽅式提供。	开箱即⽤，⽤户只需注意⼀些配置
管理和运营⼯具	⽀持的	⽀持，使⽤终端命令公开核⼼指标	⽀持的、丰富的Web 和终端命令以公开核⼼指标

三、RocketMQ的基本概念

1、技术架构

RocketMQ架构上主要分为四部分，如上图所示:

Producer：消息发布的⻆⾊，⽀持分布式集群⽅式部署。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进⾏消息投递，投递的过程⽀持快速失败并且低延迟。

Consumer：消息消费的⻆⾊，⽀持分布式集群⽅式部署。⽀持以push推，pull拉两种模式对消息进⾏消费。同时也⽀持集群⽅式和⼴播⽅式的消费，它提供实时消息订阅机制，可以满⾜⼤多数⽤户的需求。

        NameServer：NameServer是⼀个⾮常简单的Topic路由注册中⼼，其⻆⾊类似Dubbo中的zookeeper，⽀持Broker的动态注册与发现。主要包括两个功能：
        Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供⼼跳检测机制，检查Broker是否还存活；
        路由信息管理，每个NameServer将保存关于Broker集群的整个路由信息和⽤于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从⽽进⾏消息的投递和消费。NameServer通常也是集群的⽅式部署，各实例间相互不进⾏信息通讯。Broker是向每⼀台NameServer注册⾃⼰的路由信息，所以每⼀个NameServer实例上⾯都保存⼀份完整的路由信息。当某个NameServer因某种原因下线了，Broker仍然可以向其它NameServer同步其路由信息，Producer,Consumer仍然可以动态感知Broker的路由的信息。

BrokerServer：Broker主要负责消息的存储、投递和查询以及服务⾼可⽤保证，为了实现这些功能，Broker包含了以下⼏个重要⼦模块。

                Remoting Module：整个Broker的实体，负责处理来⾃clients端的请求。
                Client Manager：负责管理客户端(Producer/Consumer)和维护Consumer的Topic订阅信息
                Store Service：提供⽅便简单的API接⼝处理消息存储到物理硬盘和查询功

能。

HA Service：⾼可⽤服务，提供Master Broker 和 Slave Broker 之间的数据同步功能。

Index Service：根据特定的Message key 对投递到 Broker 的消息进⾏索引服务，以提供消息的快速查询。

2、部署架构

RocketMQ ⽹络部署特点：

NameServer是⼀个⼏乎⽆状态节点，可集群部署，节点之间⽆任何信息同步。

Broker部署相对复杂，Broker分为Master与Slave，⼀个Master可以对应多个Slave，但是⼀个Slave只能对应⼀个Master，Master与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId 来定义，BrokerId为0表示Master，⾮0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建⽴⻓连接，定时注册Topic信息到所有NameServer。注意：当前RocketMQ版本在部署架构上⽀持⼀Master多Slave，但只有BrokerId=1的从服务器才会参与消息的读负载。

Producer与NameServer集群中的其中⼀个节点（随机选择）建⽴⻓连接，定期从NameServer获取Topic路由信息，并向提供Topic 服务的Master建⽴⻓连接，且定时向Master发送⼼跳。Producer完全⽆状态，可集群部署。

Consumer与NameServer集群中的其中⼀个节点（随机选择）建⽴⻓连接，定期从NameServer获取Topic路由信息，并向提供Topic服务的Master、Slave建⽴⻓连接，且定时向Master、Slave发送⼼跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，消费者在向Master拉取消息时，Master服务器会根据拉取偏移量与最⼤偏移量的距离（判断是否读⽼消息，产⽣读I/O），以及从服务器是否可读等因素建议下⼀次是从Master还是Slave拉取。

结合部署架构图，描述集群⼯作流程：

启动NameServer，NameServer起来后监听端⼝，等待Broker、Producer、Consumer连上来，相当于⼀个路由控制中⼼。

Broker启动，跟所有的NameServer保持⻓连接，定时发送⼼跳包。⼼跳包中包含当前Broker信息(IP+端⼝等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。

收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时⾃动创建Topic。

Producer发送消息，启动时先跟NameServer集群中的其中⼀台建⽴⻓连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择⼀个队列，然后与队列所在的Broker建⽴⻓连接从⽽向Broker发消息。

Consumer跟Producer类似，跟其中⼀台NameServer建⽴⻓连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建⽴连接通道，开始消费消息。

四、快速开始

1、下载RocketMQ

本教程使⽤的是RocketMQ4.7.1版本，建议使⽤该版本进⾏之后的demo训练。建议使⽤该版本进⾏之后的demo训练。

运⾏版：https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.7.1/rocketmq-all-4.7.1-bin-release.zip

源码：https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.7.1/rocketmq-all-4.7.1-source-release.zip

2、安装RocketMQ

        准备⼀台装有Linux系统的虚拟机。
        安装jdk，上传jdk-8u191安装包并解压缩在 /usr/local/java ⽬录下。
        安装rocketmq，上传rocketmq安装包并使⽤unzip命令解压缩在 /usr/local/rocketmq ⽬录下。         配置jdk和rocketmq的环境变量

CentOS 系统配置环境变量：

对所有用户生效：

sudo vim /etc/profile

只对当前用户生效：

vim ~/.bash_profile

添加以下内容，首先mq一定安装在上面的指定目录：

export JAVA_HOME=/usr/local/java/jdk1.8.0_191

export JRE_HOME=/usr/local/java/jdk1.8.0_191/jre

export ROCKETMQ_HOME=/usr/local/rocketmq/rocketmq-all-4.7.1-bin-release

export CLASSPATH=$CLASSPATH:$JAVA_HOME/lib:$JAVA_HOME/jre/lib

export PATH=$JAVA_HOME/bin:$JAVA_HOME/jre/bin:$ROCKETMQ_HOME/bin:$PATH:$HO

ME/bin

注意，RocketMQ的环境变量⽤来加载 ROCKETMQ_HOME/conf 下的配置⽂件，如果不配置则⽆法启动NameServer和Broker。

完成后执⾏命令，让环境变量⽣效

source /etc/profile

然后修改bin/runserver.sh⽂件，由于RocketMQ默认设置的JVM内存为4G，但虚拟机⼀般没有这么4G内存，因此调整为512mb。

JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g - XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

在runserver.sh⽂件中找到上⾯这段内容，改为下⾯的参数。

JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

3、启动NameServer

在上⼀章中介绍了RocketMQ的架构，启动RocketMQ服务需要先启动NameServer。在bin⽬录内使⽤静默⽅式启动。

nohup ./mqnamesrv -n localhost:9876 &

查看bin/nohup.out显示如下内容表示启动成功：

root@ubuntu:/usr/local/rocketmq/rocketmq-all-4.7.1-bin-release/bin#cat nohup.out

Java HotSpot(TM) 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release

Java HotSpot(TM) 64-Bit Server VM warning:

UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release.

The Name Server boot success. serializeType=JSON

4、启动Broker

先修改broker的JVM参数配置，将默认8G内存修改为512m。修改 bin/runbroker.sh ⽂件

JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m"

在 conf/broker.conf ⽂件中加⼊如下配置，开启⾃动创建Topic功能

autoCreateTopicEnable=true

以静默⽅式启动broker

nohup ./mqbroker -n localhost:9876 &

查看 bin/nohup.out ⽇志，显示如下内容表示启动成功

The broker[ubuntu, 172.17.0.1:10911] boot success. serializeType=JSON

5、使⽤发送和接收消息验证MQ

配置nameserver的环境变量：在发送/接收消息之前，需要告诉客户端nameserver的位置。配置环境变量 NAMESRV_ADDR ：（也就是说编辑vim /etc/profile环境变量，然后记得执行source /etc/profile ，使配置生效）

export NAMESRV_ADDR=localhost:9876

使⽤bin/tools.sh⼯具验证消息的发送，默认会发1000条消息，但是在使用之前也需需要修改 tools.sh 中的内存堆的大小。

./tools.sh org.apache.rocketmq.example.quickstart.Producer

发送的消息⽇志：

...

SendResult [sendStatus=SEND_OK,

msgId=FD154BA55A2B1008020C29FFFED6A0855CFC12A3A380885CB70A0235,

offsetMsgId=AC11000100002A9F000000000001F491,

messageQueue=MessageQueue [topic=TopicTest, brokerName=ubuntu,

queueId=0], queueOffset=141]

使⽤bin/tools.sh⼯具验证消息的接收：

./tools.sh org.apache.rocketmq.example.quickstart.Consumer

看到接收到的消息：

...

ConsumeMessageThread_12 Receive New Messages: [MessageExt

[brokerName=ubuntu, queueId=1, storeSize=227, queueOffset=245,

sysFlag=0, bornTimestamp=1658892578234,

bornHost=/172.16.253.100:48524, storeTimestamp=1658892578235,

storeHost=/172.17.0.1:10911,

msgId=AC11000100002A9F0000000000036654, commitLogOffset=222804,

bodyCRC=683694034, reconsumeTimes=0, preparedTransactionOffset=0,

toString()=Message{topic='TopicTest', flag=0, properties=

{MIN_OFFSET=0, MAX_OFFSET=250, CONSUME_START_TIME=1658892813497,

UNIQ_KEY=FD154BA55A2B1008020C29FFFED6A0855CFC12A3A380885CB9BA03D6,

CLUSTER=DefaultCluster, WAIT=true, TAGS=TagA}, body=[72, 101, 108,

108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 57, 56, 50],

transactionId='null'}]]

6、关闭服务器

关闭broker：

./mqshutdown broker

关闭nameserver

./mqshutdown namesrv

五、搭建RocketMQ集群

1、RocketMQ集群模式

为了追求更好的性能，RocketMQ的最佳实践⽅式都是在集群模式下完成。 RocketMQ官⽅提供了三种集群搭建⽅式。

        2主2从异步通信⽅式：
        使⽤异步⽅式进⾏主从之间的数据复制，吞吐量⼤，但可能会丢消息。
        使⽤ conf/2m-2s-async ⽂件夹内的配置⽂件做集群配置。

        2主2从同步通信⽅式：
        使⽤同步⽅式进⾏主从之间的数据复制，保证消息安全投递，不会丢失，但影响吞吐量
        使⽤ conf/2m-2s-sync ⽂件夹内的配置⽂件做集群配置。

        2主⽆从⽅式：
        会存在单点故障，且读的性能没有前两种⽅式好。
        使⽤ conf/2m-noslave ⽂件夹内的配置⽂件做集群配置。

        Dledger⾼可⽤集群：|
        上述三种官⽅提供的集群没办法实现⾼可⽤，即在master节点挂掉后，slave节点没办法⾃动被选举为新的master，⽽需要⼈⼯实现。
        RocketMQ在4.5版本之后引⼊了第三⽅的Dleger⾼可⽤集群。