【rocket-mq】RocketMQ集成和RocketMQ核心技术

浅浅的花香味﹌ 2024-03-26 07:14 70阅读 0赞

> 提示：RocketMQ集成和RocketMQ核心技术

#### 文章目录 ####

*  前言
 *  一、RocketMQ环境准备
 *   *  1.编译rocketmq源码并运行
     *  2.编译rocketmq控制台源码并运行
 *  二、集成到spring-cloud-alibab
 *   *  1.消息发送进程
     *  2.消息接收进程
 *  三、RocketMQ核心技术
 *   *  1.rocketmq基本概念
     *  2.rocketmq特性
     *  3.rocketmq最佳实践
     *   *  生产者
         *  消费者
         *  Broker（代理）
         *  Name Server（集群注册、路由）
         *  客户端配置
         *  系统配置

--------------------

## 前言 ##

目前在javaweb应用中最常用的消息中间件便是rocketmq、rabbitmq和kafka了。最近涉猎cloud-aliaba项目整合rocketmq，本文记录项目中涉及的一些资料和问题处理，以便后续参考。  
github文档：[https://github.com/apache/rocketmq/tree/master/docs/cn][https_github.com_apache_rocketmq_tree_master_docs_cn]

--------------------

## 一、RocketMQ环境准备 ##

### 1.编译rocketmq源码并运行 ###

`注意：在windows系统中进行测试，不要使用PowerShell，使用cmd窗口进入后切换到ROCKETMQ_HOME\bin目录进行操作！`

1. 下载rocketmq源码
    	git clone --branch rocketmq-all-4.9.3 https://github.com/apache/rocketmq.git
    
    2. 编译（windows上需要加单引号）
    	mvn clean install '-Dmaven.test.skip=true' '-Prelease-all'
    	编译好之后在distribution\target目录下：
    	--distribution\target
    		--rocketmq-5.0.1-SNAPSHOT.tar.gz (linux)
    		--rocketmq-5.0.1-SNAPSHOT.zip (windows)
    
    3. 将windows压缩包解压后，conf目录下修改broker.conf文件（主要配置存储路径）
    	brokerClusterName = DefaultCluster
    	brokerName = broker-a
    	brokerId = 0
    	deleteWhen = 04
    	fileReservedTime = 48
    	brokerRole = ASYNC_MASTER
    	flushDiskType = ASYNC_FLUSH
    	# 自动创建Topic
    	autoCreateTopicEnable=true
    	# nameServ地址
    	namesrvAddr=127.0.0.1:9876
    	diskMaxUsedSpaceRatio=99
    	
    	# 存储路径
    	storePathRootDir=F:/software/RocketMQ/dataDir
    	# commitLog路径
    	storePathCommitLog=F:/software/RocketMQ/dataDir/commitlog
    	# 消息队列存储路径
    	storePathConsumeQueue=F:/software/RocketMQ/dataDir/consumequeue
    	# 消息索引存储路径
    	storePathIndex=F:/software/RocketMQ/dataDir/index
    	# checkpoint文件路径
    	storeCheckpoint=F:/software/RocketMQ/dataDir/checkpoint
    	# abort文件存储路径
    	abortFile=F:/software/RocketMQ/dataDir/abort
    	
    4. 配置ROCKETMQ_HOME环境变量
    	ROCKETMQ_HOME=F:\学习与工作\安装包\项目需要\rocketmq-4.9.3 
    	
    5. 启动NameServer
    	执行bin\mqnamesrv.cmd
    	
    6. 启动broker
    	执行bin\mqbroker.cmd
    
    【使用rocketmq内置工具测试：】
    	消息发送：
    	set NAMESRV_ADDR=localhost:9876
    	tools.cmd org.apache.rocketmq.example.quickstart.Producer
    	消息接收：
    	set NAMESRV_ADDR=localhost:9876
    	tools.cmd org.apache.rocketmq.example.quickstart.Consumer
    
    观察到生产和消费日志即说明，服务启动成功！

### 2.编译rocketmq控制台源码并运行 ###

1. 下载
    	git clone https://github.com/apache/rocketmq-dashboard.git
    2. 修改yml配置（主要是删除多余的namesrvAddrs地址和进程服务端口）
    	server:
    	  port: 10003 #防止和其他应用的默认端口冲突
    	  
    	rocketmq:
    	config:
    		# if this value is empty,use env value rocketmq.config.namesrvAddr  NAMESRV_ADDR | now, default localhost:9876
    		# configure multiple namesrv addresses to manage multiple different clusters
    		namesrvAddrs:
    		- 127.0.0.1:9876
    3. 编译
    	mvn install -Pq -Dmaven.test.skip=true
    4. 启动
    	java -jar rocketmq-dashboard-1.0.1-SNAPSHOT.jar
    5. 访问127.0.0.1:10003进行管理。

rocketmq管理页面如图：  
![在这里插入图片描述][745e31b32ebe4f20b488729827f339bb.png]  
jps可以看到nameserver、broker、consumer和dashboard四个java进程正在运行（producer测试进程发送完消息后停止了）：  
![在这里插入图片描述][276d458ae83047e8ab216e5b1d64216e.png]

## 二、集成到spring-cloud-alibab ##

### 1.消息发送进程 ###

1. 导入依赖
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.5.2</version>
        </dependency>
        
    2. 在进程的application.yml文件中设置rocketmq的name-server地址和分组
    	rocketmq:
    	  name-server: 127.0.0.1:9876
    	  producer:
    		group: order-group
    3. 发送消息
    	@Autowired
    	private RocketMQTemplate rocketMQTemplate;
    	
    	rocketMQTemplate.convertAndSend("order-topic", order);

### 2.消息接收进程 ###

1. 导入依赖
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.5.2</version>
        </dependency>
        
    2. 修改消费进程的application.yml配置，声明name-server地址
    	rocketmq:
    	  name-server: 127.0.0.1:9876
    	  
    3. 监听并消费消息
    	@Slf4j
    	@Component
    	// 消费组，消息主题
    	@RocketMQMessageListener(consumerGroup = "user-group", topic = "order-topic")
    	public class RocketConsumeListener implements RocketMQListener<Order> {
        
    		@Override
    		public void onMessage(Order order) {
        
    			log.info("用户微服务收到了订单信息：{}", JSONObject.toJSONString(order));
    		}
    	}

## 三、RocketMQ核心技术 ##

### 1.rocketmq基本概念 ###

**消息模型**  
RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中Producer 负责生产消息，  
Consumer 负责消费消息，Broker 负责存储消息。Broker 在实际部署过程中对应一台服务器，每个  
Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。Message  
Queue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。  
ConsumerGroup 由多个Consumer 实例构成。  
**消息生产者**  
负责生产消息，一般由业务系统负责生产消息。一个消息生产者会把业务应用系统里产生的消息发送到  
broker服务器。RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异  
步方式均需要Broker返回确认信息，单向发送不需要。  
**消息消费者**  
负责消费消息，一般是后台系统负责异步消费。一个消息消费者会从Broker服务器拉取消息、并将其提  
供给应用程序。从用户应用的角度而言提供了两种消费形式：拉取式消费、推动式消费。  
**主题**  
表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行消息  
订阅的基本单位。  
**代理服务器**  
消息中转角色，负责存储消息、转发消息。代理服务器在RocketMQ系统中负责接收从生产者发送来的  
消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、  
消费进度偏移和主题和队列消息等。  
**名字服务**  
名称服务充当路由消息的提供者。生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。  
多个Namesrv实例组成集群，但相互独立，没有信息交换。  
**拉取式消费**  
Consumer消费的一种类型，应用通常主动调用Consumer的拉消息方法从Broker服务器拉消息、主动  
权由应用控制。一旦获取了批量消息，应用就会启动消费过程。  
**推动式消费**  
Consumer消费的一种类型，该模式下Broker收到数据后会主动推送给消费端，该消费模式一般实时性  
较高。  
**生产者组**  
同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。如果发送的是事务消息且原始  
生产者在发送之后崩溃，则Broker服务器会联系同一生产者组的其他生产者实例以提交或回溯消费。  
**消费者组**  
同一类Consumer的集合，这类Consumer通常消费同一类消息且消费逻辑一致。消费者组使得在消息  
消费方面，实现负载均衡和容错的目标变得非常容易。要注意的是，消费者组的消费者实例必须订阅完  
全相同的Topic。RocketMQ 支持两种消息模式：集群消费（Clustering）和广播消费  
（Broadcasting）。  
**集群消费**  
集群消费模式下,相同Consumer Group的每个Consumer实例平均分摊消息。  
**广播消费**  
广播消费模式下，相同Consumer Group的每个Consumer实例都接收全量的消息。  
**普通顺序消息**  
普通顺序消费模式下，消费者通过同一个消息队列（ Topic 分区，称作 Message Queue） 收到的消息  
是有顺序的，不同消息队列收到的消息则可能是无顺序的。  
**严格顺序消息**  
严格顺序消息模式下，消费者收到的所有消息均是有顺序的。  
**消息**  
消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。  
RocketMQ中每个消息拥有唯一的Message ID，且可以携带具有业务标识的Key。系统提供了通过  
Message ID和Key查询消息的功能。  
**标签**  
为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业  
务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提  
供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。

### 2.rocketmq特性 ###

![在这里插入图片描述][f92dff96fb04421ba341f9e65deacd49.png]  
**顺序消息**分为全局顺序消息与分区顺序消息，全局顺序是指某个Topic下的所有消息都要保证顺序；部分  
顺序消息只要保证每一组消息被顺序消费即可。

**消息可靠性**  
RocketMQ支持消息的高可靠，影响消息可靠性的几种情况：

1.  Broker非正常关闭
2.  Broker异常Crash
3.  OS Crash
4.  机器掉电，但是能立即恢复供电情况
5.  机器无法开机（可能是cpu、主板、内存等关键设备损坏）
6.  磁盘设备损坏  
    1)、2)、3)、4) 四种情况都属于硬件资源可立即恢复情况，RocketMQ在这四种情况下能保证消息不丢，  
    或者丢失少量数据（依赖刷盘方式是同步还是异步）。  
    5)、6)属于单点故障，且无法恢复，一旦发生，在此单点上的消息全部丢失。RocketMQ在这两种情况  
    下，通过异步复制，可保证99%的消息不丢，但是仍然会有极少量的消息可能丢失。通过同步双写技术  
    可以完全避免单点，同步双写势必会影响性能，适合对消息可靠性要求极高的场合，例如与Money相关  
    的应用。注：RocketMQ从3.0版本开始支持同步双写。

**至少一次**  
至少一次(At least Once)指每个消息必须投递一次。Consumer先Pull消息到本地，消费完成后，才向服  
务器返回ack，如果没有消费一定不会ack消息，所以RocketMQ可以很好的支持此特性。

**回溯消费**  
回溯消费是指Consumer已经消费成功的消息，由于业务上需求需要重新消费，要支持此功能，Broker  
在向Consumer投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度，例如由于  
Consumer系统故障，恢复后需要重新消费1小时前的数据，那么Broker要提供一种机制，可以按照时间  
维度来回退消费进度。RocketMQ支持按照时间回溯消费，时间维度精确到毫秒。

**事务消息**  
RocketMQ事务消息（Transactional Message）是指应用本地事务和发送消息操作可以被定义到全局事  
务中，要么同时成功，要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通  
过事务消息能达到分布式事务的最终一致。

**定时消息**  
定时消息（延迟队列）是指消息发送到broker后，不会立即被消费，等待特定时间投递给真正的topic。  
broker有配置项messageDelayLevel，默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m  
10m 20m 30m 1h 2h”，18个level。

**死信队列**  
死信队列用于处理无法被正常消费的消息。当一条消息初次消费失败，消息队列会自动进行消息重试；  
达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消  
息队列 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。

### 3.rocketmq最佳实践 ###

#### 生产者 ####

一个应用尽可能用一个Topic，而**消息子类型则可以用tags来标识**。message.setTags(“TagA”)。  
每个消息在业务层面的**唯一标识码要设置到keys字段**，方便将来定位消息丢失问题。message.setKeys(orderId);  
日志的打印，消息发送成功或者失败要打印消息日志，务必要打印SendResult和key字段。

Producer的send方法本身支持内部重试，重试逻辑如下：

*  至多重试2次
 *  如果同步模式发送失败，则轮转到下一个Broker，如果异步模式发送失败，则只会在当前Broker进  
    行重试。这个方法的总耗时时间不超过sendMsgTimeout设置的值，默认10s。
 *  如果本身向broker发送消息产生超时异常，就不会再重试。  
    **如果业务对消息可靠性要求比较高，建议应用增加相应的重试逻辑。**

某些场景要求耗时非常短，但是对可靠性要求并不高，例如日志收集类应用，此类应用可以采用oneway形式调用，，oneway形式只发送请求不等待应答，而发送请求在客户端实现层面仅仅是一个操作系统系统调用的开销，即**将数据写入客户端的socket缓冲区**，此过程耗时通常在微秒级。

#### 消费者 ####

**消费过程幂等：**  
RocketMQ无法避免消息重复（Exactly-Once），所以如果业务对消费重复非常敏感，务必要在业务层  
面进行去重处理。

**消费速度：**

*  提高消费并行度  
    同ConsumerGroup下，增加Consumer实例数量
 *  **批量方式消费（推荐的处理方式）**
 *  跳过非重要消息，选择丢弃不重要的消息
 *  优化消费过程

建议务必打印消费入口日志，利于问题排查定位。

**线程数设置：**  
消费者使用 ThreadPoolExecutor 在内部对消息进行消费，所以你可以通过设置setConsumeThreadMin 或 setConsumeThreadMax 来改变它。

**消费位点：**  
当建立一个新的消费者组时，需要决定是否需要消费已经存在于 Broker 中的历史消息CONSUME\_FROM\_LAST\_OFFSET 将会忽略历史消息，并消费之后生成的任何消息。  
CONSUME\_FROM\_FIRST\_OFFSET 将会消费每个存在于 Broker 中的信息。你也可以使用CONSUME\_FROM\_TIMESTAMP 来消费在指定时间戳后产生的消息。

#### Broker（代理） ####

Broker 角色分为 ASYNC\_MASTER（异步主机）、SYNC\_MASTER（同步主机）以及SLAVE（从机）。  
如果对消息的可靠性要求比较严格，可以采用 SYNC\_MASTER加SLAVE的部署方式。如果对消息可靠性要求不高，可以采用ASYNC\_MASTER加SLAVE的部署方式。如果只是测试方便，可以选择仅主机部署方式。

**FlushDiskType**  
\*\*SYNC\_FLUSH（同步刷新）**相比于**ASYNC\_FLUSH（异步处理）\*\*会损失很多性能，但是也更可靠，所以需要根据实际的业务场景做好权衡。

**Broker 配置**  
![在这里插入图片描述][431ae7e626774001a1ac1c01249c2984.png]

#### Name Server（集群注册、路由） ####

RocketMQ 中，Name Servers 被设计用来做简单的路由管理。其职责包括：

*  Brokers 定期向每个名称服务器注册路由数据。
 *  名称服务器为客户端，包括生产者，消费者和命令行客户端提供最新的路由信息。

#### 客户端配置 ####

**客户端寻址方式**  
RocketMQ可以令客户端找到Name Server, 然后通过Name Server再找到Broker。如下所示有多种配置  
方式，优先级由高到低，高优先级会覆盖低优先级。

*  代码中指定Name Server地址，多个namesrv地址之间用分号分割

producer.setNamesrvAddr("192.168.0.1:9876;192.168.0.2:9876");
    consumer.setNamesrvAddr("192.168.0.1:9876;192.168.0.2:9876");

*  Java启动参数中指定Name Server地址

-Drocketmq.namesrv.addr=192.168.0.1:9876;192.168.0.2:9876

*  环境变量指定Name Server地址

export NAMESRV_ADDR=192.168.0.1:9876;192.168.0.2:9876

*  HTTP静态服务器寻址（默认）  
    客户端启动后，会定时访问一个静态HTTP服务器，地址如下：http://jmenv.tbsite.net:8080/rocketm  
    q/nsaddr，这个URL的返回内容如下：  
    `192.168.0.1:9876;192.168.0.2:9876`  
    客户端默认每隔2分钟访问一次这个HTTP服务器，并更新本地的Name Server地址。URL已经在代码中硬编码，可通过修改/etc/hosts文件来改变要访问的服务器，例如在/etc/hosts增加如下配置：  
    `export NAMESRV_ADDR=192.168.0.1:9876;192.168.0.2:9876`  
    推荐使用HTTP静态服务器寻址方式，好处是客户端部署简单，且Name Server集群可以热升级。

**客户端的公共配置**  
![客户端的公共配置][95c3ec9c24d34b709dc05d1cf8730b34.png]

#### 系统配置 ####

**JVM相关**

*  设置相同的Xms和Xmx值来防止JVM调整堆大小以获得更好的性能。

-server -Xms8g -Xmx8g -Xmn4g

*  如果不关心RocketMQ Broker的启动时间，通过“预触摸”Java堆以确保在JVM初始化期间每个页面都将被分配

-XX:+AlwaysPreTouch

*  禁用偏置锁定可能会减少JVM暂停

-XX:-UseBiasedLocking

*  垃圾回收，建议使用带JDK 1.8的G1收集器

-XX:+UseG1GC -XX:G1HeapRegionSize=16m
    -XX:G1ReservePercent=25
    -XX:InitiatingHeapOccupancyPercent=30

*  不要把-XX:MaxGCPauseMillis的值设置太小，否则JVM将使用一个小的年轻代来实现这个目标，这将导致非常频繁的minor GC

-XX:+UseGCLogFileRotation
    -XX:NumberOfGCLogFiles=5
    -XX:GCLogFileSize=30m

`-Xloggc:/dev/shm/mq_gc_%p.log123`

*  建议使用rolling GC日志文件，如果写入GC文件会增加代理的延迟，可以考虑将GC日志文件重定向到内存文件系统

-server -Xms8g -Xmx8g -Xmn4g

**Linux内核参数**

*  vm.extra\_free\_kbytes，告诉VM在后台回收（kswapd）启动的阈值与直接回收（通过分配进程）的阈值之间保留额外的可用内存。RocketMQ使用此参数来避免内存分配中的长延迟。（与具  
    体内核版本相关）
 *  vm.min\_free\_kbytes，如果将其设置为低于1024KB，将会巧妙的将系统破坏，并且系统在高负载下容易出现死锁。
 *  vm.max\_map\_count，限制一个进程可能具有的最大内存映射区域数。RocketMQ将使用mmap加载CommitLog和ConsumeQueue，因此建议将为此参数设置较大的值。（agressiveness -->  
    aggressiveness）
 *  **vm.swappiness**，定义内核交换内存页面的积极程度。较高的值会增加攻击性，较低的值会减少交换量。建议将值设置为10来避免交换延迟。
 *  **File descriptor limits**，RocketMQ需要为文件（CommitLog和ConsumeQueue）和网络连接打开文件描述符。我们建议设置文件描述符的值为655350。
 *  Disk scheduler，RocketMQ建议使用I/O截止时间调度器，它试图为请求提供有保证的延迟。

--------------------

参考：  
[https://github.com/apache/rocketmq/tree/master/docs/cn][https_github.com_apache_rocketmq_tree_master_docs_cn]

[https_github.com_apache_rocketmq_tree_master_docs_cn]: https://github.com/apache/rocketmq/tree/master/docs/cn
[745e31b32ebe4f20b488729827f339bb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/09f15ae7f7d6490d98616476de949d9a.png
[276d458ae83047e8ab216e5b1d64216e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/3084506cde6f4261ae7da61a87b4f84b.png
[f92dff96fb04421ba341f9e65deacd49.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/3f5c79d520af484bae5274e07b9250ad.png
[431ae7e626774001a1ac1c01249c2984.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/343bf0df8c0a49728a948ee72ec51ab2.png
[95c3ec9c24d34b709dc05d1cf8730b34.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/c761a8b8d6c94eb2939ddad790ba345c.png