1. MQ介绍
##1.1 为什么要用MQ
消息队列是一种“先进先出”的数据结构
其应用场景主要包含以下3个方面
- 应用解耦
系统的耦合性越高，容错性就越低。以电商应用为例，用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障或者因为升级等原因暂时不可用，都会造成下单操作异常，影响用户使用体验。
使用消息队列解耦合，系统的耦合性就会提高了。比如物流系统发生故障，需要几分钟才能来修复，在这段时间内，物流系统要处理的数据被缓存到消息队列中，用户的下单操作正常完成。当物流系统回复后，补充处理存在消息队列中的订单消息即可，终端系统感知不到物流系统发生过几分钟故障。
- 流量削峰
应用系统如果遇到系统请求流量的瞬间猛增，有可能会将系统压垮。有了消息队列可以将大量请求缓存起来，分散到很长一段时间处理，这样可以大大提到系统的稳定性和用户体验。
一般情况，为了保证系统的稳定性，如果系统负载超过阈值，就会阻止用户请求，这会影响用户体验，而如果使用消息队列将请求缓存起来，等待系统处理完毕后通知用户下单完毕，这样总不能下单体验要好。
处于经济考量目的：
业务系统正常时段的QPS如果是1000，流量最高峰是10000，为了应对流量高峰配置高性能的服务器显然不划算，这时可以使用消息队列对峰值流量削峰
- 数据分发
通过消息队列可以让数据在多个系统更加之间进行流通。数据的产生方不需要关心谁来使用数据，只需要将数据发送到消息队列，数据使用方直接在消息队列中直接获取数据即可
1.2 MQ的优点和缺点
优点：解耦、削峰、数据分发
缺点包含以下几点：
- 系统可用性降低 系统引入的外部依赖越多，系统稳定性越差。一旦MQ宕机，就会对业务造成影响。 如何保证MQ的高可用？
- 系统复杂度提高 MQ的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过MQ进行异步调用。 如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？
- 一致性问题 A系统处理完业务，通过MQ给B、C、D三个系统发消息数据，如果B系统、C系统处理成功，D系统处理失败。 如何保证消息数据处理的一致性？
1.3 各种MQ产品的比较
常见的MQ产品包括Kafka、ActiveMQ、RabbitMQ、RocketMQ。
2. RocketMQ快速入门
RocketMQ是阿里巴巴2016年MQ中间件，使用Java语言开发，在阿里内部，RocketMQ承接了例如“双11”等高并发场景的消息流转，能够处理万亿级别的消息。
2.1 准备工作
2.1.1 下载RocketMQ
RocketMQ最新版本：4.5.1
下载地址
2.2.2 环境要求
- Linux64位系统
- JDK1.8(64位)
- 源码安装需要安装Maven 3.2.x
2.2 安装RocketMQ
2.2.1 安装步骤
本教程以二进制包方式安装
- 解压安装包
- 进入安装目录
2.2.2 目录介绍
- bin：启动脚本，包括shell脚本和CMD脚本
- conf：实例配置文件 ，包括broker配置文件、logback配置文件等
- lib：依赖jar包，包括Netty、commons-lang、FastJSON等
2.3 启动RocketMQ
- 启动NameServer
# 1.启动NameServer
nohup sh bin/mqnamesrv &
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/namesrv.log- 启动Broker
# 1.启动Broker
nohup sh bin/mqbroker -n localhost:9876 &
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/broker.log - 问题描述： RocketMQ默认的虚拟机内存较大，启动Broker如果因为内存不足失败，需要编辑如下两个配置文件，修改JVM内存大小
# 编辑runbroker.sh和runserver.sh修改默认JVM大小
vi runbroker.sh
vi runserver.sh- 参考设置：
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
2.4 测试RocketMQ
2.4.1 发送消息
代码语言：javascript复制# 1.设置环境变量
export NAMESRV_ADDR=localhost:9876
# 2.使用安装包的Demo发送消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer2.4.2 接收消息
代码语言：javascript复制# 1.设置环境变量
export NAMESRV_ADDR=localhost:9876
# 2.接收消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer2.5 关闭RocketMQ
代码语言：javascript复制# 1.关闭NameServer
sh bin/mqshutdown namesrv
# 2.关闭Broker
sh bin/mqshutdown broker3. RocketMQ集群搭建
3.1 各角色介绍
- Producer：消息的发送者；举例：发信者
- Consumer：消息接收者；举例：收信者
- Broker：暂存和传输消息；举例：邮局
- NameServer：管理Broker；举例：各个邮局的管理机构
- Topic：区分消息的种类；一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者可以订阅一个或者多个Topic消息
- Message Queue：相当于是Topic的分区；用于并行发送和接收消息
3.2 集群搭建方式
3.2.1 集群特点
- NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
- Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。
- Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。
- Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。
3.2.3 集群模式
1）单Master模式
这种方式风险较大，一旦Broker重启或者宕机时，会导致整个服务不可用。不建议线上环境使用,可以用于本地测试。
2）多Master模式
一个集群无Slave，全是Master，例如2个Master或者3个Master，这种模式的优缺点如下：
- 优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高；
- 缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。
3）多Master多Slave模式（异步）
每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式，主备有短暂消息延迟（毫秒级），这种模式的优缺点如下：
- 优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎一样；
- 缺点：Master宕机，磁盘损坏情况下会丢失少量消息。
4）多Master多Slave模式（同步）
每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功，这种模式的优缺点如下：
- 优点：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；
- 缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。
3.3 双主双从集群搭建
3.3.1 总体架构
消息高可用采用2m-2s（同步双写）方式
3.3.2 集群工作流程
- 启动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建立连接通道，开始消费消息。
3.3.3 服务器环境
序号 | IP | 角色 | 架构模式 |
|---|---|---|---|
1 | 192.168.25.135 | nameserver、brokerserver | Master1、Slave2 |
2 | 192.168.25.138 | nameserver、brokerserver | Master2、Slave1 |
3.3.4 Host添加信息
代码语言：javascript复制vim /etc/hosts配置如下:
代码语言：javascript复制# nameserver
192.168.25.135 rocketmq-nameserver1
192.168.25.138 rocketmq-nameserver2
# broker
192.168.25.135 rocketmq-master1
192.168.25.135 rocketmq-slave2
192.168.25.138 rocketmq-master2
192.168.25.138 rocketmq-slave1配置完成后, 重启网卡
代码语言：javascript复制systemctl restart network3.3.5 防火墙配置
宿主机需要远程访问虚拟机的rocketmq服务和web服务，需要开放相关的端口号，简单粗暴的方式是直接关闭防火墙
代码语言：javascript复制# 关闭防火墙
systemctl stop firewalld.service
# 查看防火墙的状态
firewall-cmd --state
# 禁止firewall开机启动
systemctl disable firewalld.service或者为了安全，只开放特定的端口号，RocketMQ默认使用3个端口：9876 、10911 、11011 。如果防火墙没有关闭的话，那么防火墙就必须开放这些端口：
nameserver默认使用 9876 端口master默认使用 10911 端口slave默认使用11011 端口
执行以下命令：
代码语言：javascript复制# 开放name server默认端口
firewall-cmd --remove-port=9876/tcp --permanent
# 开放master默认端口
firewall-cmd --remove-port=10911/tcp --permanent
# 开放slave默认端口 (当前集群模式可不开启)
firewall-cmd --remove-port=11011/tcp --permanent
# 重启防火墙
firewall-cmd --reload3.3.6 环境变量配置
代码语言：javascript复制vim /etc/profile在profile文件的末尾加入如下命令
代码语言：javascript复制#set rocketmq
ROCKETMQ_HOME=/usr/local/rocketmq/rocketmq-all-4.4.0-bin-release
PATH=$PATH:$ROCKETMQ_HOME/bin
export ROCKETMQ_HOME PATH输入:wq! 保存并退出， 并使得配置立刻生效：
代码语言：javascript复制source /etc/profile3.3.7 创建消息存储路径
代码语言：javascript复制mkdir /usr/local/rocketmq/store
mkdir /usr/local/rocketmq/store/commitlog
mkdir /usr/local/rocketmq/store/consumequeue
mkdir /usr/local/rocketmq/store/index3.3.8 broker配置文件
1）master1
服务器：192.168.25.135
代码语言：javascript复制vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a.properties修改配置如下：
代码语言：javascript复制#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=SYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=1282）slave2
服务器：192.168.25.135
代码语言：javascript复制vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b-s.properties修改配置如下：
代码语言：javascript复制#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0 表示 Master，>0 表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=11011
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SLAVE
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=1283）master2
服务器：192.168.25.138
代码语言：javascript复制vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b.properties修改配置如下：
代码语言：javascript复制#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=SYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=1284）slave1
服务器：192.168.25.138
代码语言：javascript复制vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a-s.properties修改配置如下：
代码语言：javascript复制#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=11011
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SLAVE
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=1283.3.9 修改启动脚本文件
1）runbroker.sh
代码语言：javascript复制vi /usr/local/rocketmq/bin/runbroker.sh需要根据内存大小进行适当的对JVM参数进行调整：
代码语言：javascript复制#===================================================
# 开发环境配置 JVM Configuration
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"####2）runserver.sh
代码语言：javascript复制vim /usr/local/rocketmq/bin/runserver.shJAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"3.3.10 服务启动
1）启动NameServe集群
分别在192.168.25.135和192.168.25.138启动NameServer
代码语言：javascript复制cd /usr/local/rocketmq/bin
nohup sh mqnamesrv &2）启动Broker集群
- 在192.168.25.135上启动master1和slave2
master1：
代码语言：javascript复制cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-syncbroker-a.properties &slave2：
代码语言：javascript复制cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b-s.properties &- 在192.168.25.138上启动master2和slave2
master2
代码语言：javascript复制cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b.properties &slave1
代码语言：javascript复制cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-a-s.properties &3.3.11 查看进程状态
启动后通过JPS查看启动进程
3.3.12 查看日志
代码语言：javascript复制# 查看nameServer日志
tail -500f ~/logs/rocketmqlogs/namesrv.log
# 查看broker日志
tail -500f ~/logs/rocketmqlogs/broker.log3.4 mqadmin管理工具
3.4.1 使用方式
进入RocketMQ安装位置，在bin目录下执行./mqadmin {command} {args}
###3.4.2 命令介绍
####1）Topic相关
名称 | 含义 | 命令选项 | 说明 |
|---|---|---|---|
updateTopic | 创建更新Topic配置 | -b | Broker 地址，表示 topic 所在 Broker，只支持单台Broker，地址为ip:port |
-c | cluster 名称，表示 topic 所在集群（集群可通过 clusterList 查询） | ||
-h- | 打印帮助 | ||
-n | NameServer服务地址，格式 ip:port | ||
-p | 指定新topic的读写权限( W=2|R=4|WR=6 ) | ||
-r | 可读队列数（默认为 8） | ||
-w | 可写队列数（默认为 8） | ||
-t | topic 名称（名称只能使用字符 ^[a-zA-Z0-9_-] $ ） | ||
deleteTopic | 删除Topic | -c | cluster 名称，表示删除某集群下的某个 topic （集群 可通过 clusterList 查询） |
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
-t | topic 名称（名称只能使用字符 ^[a-zA-Z0-9_-] $ ） | ||
topicList | 查看 Topic 列表信息 | -h | 打印帮助 |
-c | 不配置-c只返回topic列表，增加-c返回clusterName, topic, consumerGroup信息，即topic的所属集群和订阅关系，没有参数 | ||
-n | NameServer 服务地址，格式 ip:port | ||
topicRoute | 查看 Topic 路由信息 | -t | topic 名称 |
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
topicStatus | 查看 Topic 消息队列offset | -t | topic 名称 |
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
topicClusterList | 查看 Topic 所在集群列表 | -t | topic 名称 |
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
updateTopicPerm | 更新 Topic 读写权限 | -t | topic 名称 |
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
-b | Broker 地址，表示 topic 所在 Broker，只支持单台Broker，地址为ip:port | ||
-p | 指定新 topic 的读写权限( W=2|R=4|WR=6 ) | ||
-c | cluster 名称，表示 topic 所在集群（集群可通过 clusterList 查询），-b优先，如果没有-b，则对集群中所有Broker执行命令 | ||
updateOrderConf | 从NameServer上创建、删除、获取特定命名空间的kv配置，目前还未启用 | -h | 打印帮助 |
-n | NameServer 服务地址，格式 ip:port | ||
-t | topic，键 | ||
-v | orderConf，值 | ||
-m | method，可选get、put、delete | ||
allocateMQ | 以平均负载算法计算消费者列表负载消息队列的负载结果 | -t | topic 名称 |
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
-i | ipList，用逗号分隔，计算这些ip去负载Topic的消息队列 | ||
statsAll | 打印Topic订阅关系、TPS、积累量、24h读写总量等信息 | -h | 打印帮助 |
-n | NameServer 服务地址，格式 ip:port | ||
-a | 是否只打印活跃topic | ||
-t | 指定topic |
####2）集群相关
名称 | 含义 | 命令选项 | 说明 |
|---|---|---|---|
clusterList | 查看集群信息，集群、BrokerName、BrokerId、TPS等信息 | -m | 打印更多信息 (增加打印出如下信息 #InTotalYest, #OutTotalYest, #InTotalToday ,#OutTotalToday) |
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
-i | 打印间隔，单位秒 | ||
clusterRT | 发送消息检测集群各Broker RT。消息发往${BrokerName} Topic。 | -a | amount，每次探测的总数，RT = 总时间 / amount |
-s | 消息大小，单位B | ||
-c | 探测哪个集群 | ||
-p | 是否打印格式化日志，以|分割，默认不打印 | ||
-h | 打印帮助 | ||
-m | 所属机房，打印使用 | ||
-i | 发送间隔，单位秒 | ||
-n | NameServer 服务地址，格式 ip:port |
####3）Broker相关
名称 | 含义 | 命令选项 | 说明 |
|---|---|---|---|
updateBrokerConfig | 更新 Broker 配置文件，会修改Broker.conf | -b | Broker 地址，格式为ip:port |
-c | cluster 名称 | ||
-k | key 值 | ||
-v | value 值 | ||
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
brokerStatus | 查看 Broker 统计信息、运行状态（你想要的信息几乎都在里面） | -b | Broker 地址，地址为ip:port |
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
brokerConsumeStats | Broker中各个消费者的消费情况，按Message Queue维度返回Consume Offset，Broker Offset，Diff，TImestamp等信息 | -b | Broker 地址，地址为ip:port |
-t | 请求超时时间 | ||
-l | diff阈值，超过阈值才打印 | ||
-o | 是否为顺序topic，一般为false | ||
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
getBrokerConfig | 获取Broker配置 | -b | Broker 地址，地址为ip:port |
-n | NameServer 服务地址，格式 ip:port | ||
wipeWritePerm | 从NameServer上清除 Broker写权限 | -b | Broker 地址，地址为ip:port |
-n | NameServer 服务地址，格式 ip:port | ||
-h | 打印帮助 | ||
cleanExpiredCQ | 清理Broker上过期的Consume Queue，如果手动减少对列数可能产生过期队列 | -n | NameServer 服务地址，格式 ip:port |
-h | 打印帮助 | ||
-b | Broker 地址，地址为ip:port | ||
-c | 集群名称 | ||
cleanUnusedTopic | 清理Broker上不使用的Topic，从内存中释放Topic的Consume Queue，如果手动删除Topic会产生不使用的Topic | -n | NameServer 服务地址，格式 ip:port |
-h | 打印帮助 | ||
-b | Broker 地址，地址为ip:port | ||
-c | 集群名称 | ||
sendMsgStatus | 向Broker发消息，返回发送状态和RT | -n | NameServer 服务地址，格式 ip:port |
-h | 打印帮助 | ||
-b | BrokerName，注意不同于Broker地址 | ||
-s | 消息大小，单位B | ||
-c | 发送次数 |
####4）消息相关
名称 | 含义 | 命令选项 | 说明 |
|---|---|---|---|
queryMsgById | 根据offsetMsgId查询msg，如果使用开源控制台，应使用offsetMsgId，此命令还有其他参数，具体作用请阅读QueryMsgByIdSubCommand。 | -i | msgId |
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
queryMsgByKey | 根据消息 Key 查询消息 | -k | msgKey |
-t | Topic 名称 | ||
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
queryMsgByOffset | 根据 Offset 查询消息 | -b | Broker 名称，（这里需要注意 填写的是 Broker 的名称，不是 Broker 的地址，Broker 名称可以在 clusterList 查到） |
-i | query 队列 id | ||
-o | offset 值 | ||
-t | topic 名称 | ||
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
queryMsgByUniqueKey | 根据msgId查询，msgId不同于offsetMsgId，区别详见常见运维问题。-g，-d配合使用，查到消息后尝试让特定的消费者消费消息并返回消费结果 | -h | 打印帮助 |
-n | NameServer 服务地址，格式 ip:port | ||
-i | uniqe msg id | ||
-g | consumerGroup | ||
-d | clientId | ||
-t | topic名称 | ||
checkMsgSendRT | 检测向topic发消息的RT，功能类似clusterRT | -h | 打印帮助 |
-n | NameServer 服务地址，格式 ip:port | ||
-t | topic名称 | ||
-a | 探测次数 | ||
-s | 消息大小 | ||
sendMessage | 发送一条消息，可以根据配置发往特定Message Queue，或普通发送。 | -h | 打印帮助 |
-n | NameServer 服务地址，格式 ip:port | ||
-t | topic名称 | ||
-p | body，消息体 | ||
-k | keys | ||
-c | tags | ||
-b | BrokerName | ||
-i | queueId | ||
consumeMessage | 消费消息。可以根据offset、开始&结束时间戳、消息队列消费消息，配置不同执行不同消费逻辑，详见ConsumeMessageCommand。 | -h | 打印帮助 |
-n | NameServer 服务地址，格式 ip:port | ||
-t | topic名称 | ||
-b | BrokerName | ||
-o | 从offset开始消费 | ||
-i | queueId | ||
-g | 消费者分组 | ||
-s | 开始时间戳，格式详见-h | ||
-d | 结束时间戳 | ||
-c | 消费多少条消息 | ||
printMsg | 从Broker消费消息并打印，可选时间段 | -h | 打印帮助 |
-n | NameServer 服务地址，格式 ip:port | ||
-t | topic名称 | ||
-c | 字符集，例如UTF-8 | ||
-s | subExpress，过滤表达式 | ||
-b | 开始时间戳，格式参见-h | ||
-e | 结束时间戳 | ||
-d | 是否打印消息体 | ||
printMsgByQueue | 类似printMsg，但指定Message Queue | -h | 打印帮助 |
-n | NameServer 服务地址，格式 ip:port | ||
-t | topic名称 | ||
-i | queueId | ||
-a | BrokerName | ||
-c | 字符集，例如UTF-8 | ||
-s | subExpress，过滤表达式 | ||
-b | 开始时间戳，格式参见-h | ||
-e | 结束时间戳 | ||
-p | 是否打印消息 | ||
-d | 是否打印消息体 | ||
-f | 是否统计tag数量并打印 | ||
resetOffsetByTime | 按时间戳重置offset，Broker和consumer都会重置 | -h | 打印帮助 |
-n | NameServer 服务地址，格式 ip:port | ||
-g | 消费者分组 | ||
-t | topic名称 | ||
-s | 重置为此时间戳对应的offset | ||
-f | 是否强制重置，如果false，只支持回溯offset，如果true，不管时间戳对应offset与consumeOffset关系 | ||
-c | 是否重置c 客户端offset |
5）消费者、消费组相关
名称 | 含义 | 命令选项 | 说明 |
|---|---|---|---|
consumerProgress | 查看订阅组消费状态，可以查看具体的client IP的消息积累量 | -g | 消费者所属组名 |
-s | 是否打印client IP | ||
-h | 打印帮助 | ||
-n | NameServer 服务地址，格式 ip:port | ||
consumerStatus | 查看消费者状态，包括同一个分组中是否都是相同的订阅，分析Process Queue是否堆积，返回消费者jstack结果，内容较多，使用者参见ConsumerStatusSubCommand | -h | 打印帮助 |
-n | NameServer 服务地址，格式 ip:port | ||
-g | consumer group | ||
-i | clientId | ||
-s | 是否执行jstack | ||
getConsumerStatus | 获取 Consumer 消费进度 | -g | 消费者所属组名 |
-t | 查询主题 | ||
-i | Consumer 客户端 ip | ||
-n | NameServer 服务地址，格式 ip:port | ||
-h | 打印帮助 | ||
updateSubGroup | 更新或创建订阅关系 | -n | NameServer 服务地址，格式 ip:port |
-h | 打印帮助 | ||
-b | Broker地址 | ||
-c | 集群名称 | ||
-g | 消费者分组名称 | ||
-s | 分组是否允许消费 | ||
-m | 是否从最小offset开始消费 | ||
-d | 是否是广播模式 | ||
-q | 重试队列数量 | ||
-r | 最大重试次数 | ||
-i | 当slaveReadEnable开启时有效，且还未达到从slave消费时建议从哪个BrokerId消费，可以配置备机id，主动从备机消费 | ||
-w | 如果Broker建议从slave消费，配置决定从哪个slave消费，配置BrokerId，例如1 | ||
-a | 当消费者数量变化时是否通知其他消费者负载均衡 | ||
deleteSubGroup | 从Broker删除订阅关系 | -n | NameServer 服务地址，格式 ip:port |
-h | 打印帮助 | ||
-b | Broker地址 | ||
-c | 集群名称 | ||
-g | 消费者分组名称 | ||
cloneGroupOffset | 在目标群组中使用源群组的offset | -n | NameServer 服务地址，格式 ip:port |
-h | 打印帮助 | ||
-s | 源消费者组 | ||
-d | 目标消费者组 | ||
-t | topic名称 | ||
-o | 暂未使用 |
6）连接相关
名称 | 含义 | 命令选项 | 说明 |
|---|---|---|---|
consumerConnec tion | 查询 Consumer 的网络连接 | -g | 消费者所属组名 |
-n | NameServer 服务地址，格式 ip:port | ||
-h | 打印帮助 | ||
producerConnec tion | 查询 Producer 的网络连接 | -g | 生产者所属组名 |
-t | 主题名称 | ||
-n | NameServer 服务地址，格式 ip:port | ||
-h | 打印帮助 |
7）NameServer相关
名称 | 含义 | 命令选项 | 说明 |
|---|---|---|---|
updateKvConfig | 更新NameServer的kv配置，目前还未使用 | -s | 命名空间 |
-k | key | ||
-v | value | ||
-n | NameServer 服务地址，格式 ip:port | ||
-h | 打印帮助 | ||
deleteKvConfig | 删除NameServer的kv配置 | -s | 命名空间 |
-k | key | ||
-n | NameServer 服务地址，格式 ip:port | ||
-h | 打印帮助 | ||
getNamesrvConfig | 获取NameServer配置 | -n | NameServer 服务地址，格式 ip:port |
-h | 打印帮助 | ||
updateNamesrvConfig | 修改NameServer配置 | -n | NameServer 服务地址，格式 ip:port |
-h | 打印帮助 | ||
-k | key | ||
-v | value |
8）其他
名称 | 含义 | 命令选项 | 说明 |
|---|---|---|---|
startMonitoring | 开启监控进程，监控消息误删、重试队列消息数等 | -n | NameServer 服务地址，格式 ip:port |
-h | 打印帮助 |
3.4.3 注意事项
- 几乎所有命令都需要配置-n表示NameServer地址，格式为ip:port
- 几乎所有命令都可以通过-h获取帮助
- 如果既有Broker地址（-b）配置项又有clusterName（-c）配置项，则优先以Broker地址执行命令；如果不配置Broker地址，则对集群中所有主机执行命令
3.5 集群监控平台搭建
3.5.1 概述
RocketMQ有一个对其扩展的开源项目incubator-rocketmq-externals，这个项目中有一个子模块叫rocketmq-console，这个便是管理控制台项目了，先将incubator-rocketmq-externals拉到本地，因为我们需要自己对rocketmq-console进行编译打包运行。
3.5.2 下载并编译打包
代码语言：javascript复制git clone https://github.com/apache/rocketmq-externals
cd rocketmq-console
mvn clean package -Dmaven.test.skip=true注意：打包前在rocketmq-console中配置namesrv集群地址：
rocketmq.config.namesrvAddr=192.168.25.135:9876;192.168.25.138:9876启动rocketmq-console：
代码语言：javascript复制java -jar rocketmq-console-ng-1.0.0.jar启动成功后，我们就可以通过浏览器访问http://localhost:8080进入控制台界面了，如下图：
集群状态：
4. 消息发送样例
- 导入MQ客户端依赖
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.4.0</version>
</dependency>- 消息发送者步骤分析r
1.创建消息生产者producer，并制定生产者组名
2.指定Nameserver地址
3.启动producer
4.创建消息对象，指定主题Topic、Tag和消息体
5.发送消息
6.关闭生产者producer- 消息消费者步骤分析
1.创建消费者Consumer，制定消费者组名
2.指定Nameserver地址
3.订阅主题Topic和Tag
4.设置回调函数，处理消息
5.启动消费者consumer4.1 基本样例
4.1.1 消息发送
1）发送同步消息
这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。
代码语言：javascript复制public class SyncProducer {
public static void main(String[] args) throws Exception {
// 实例化消息生产者Producer
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
// 设置NameServer的地址
producer.setNamesrvAddr("localhost:9876");
// 启动Producer实例
producer.start();
for (int i = 0; i < 100; i ) {
// 创建消息，并指定Topic，Tag和消息体
Message msg = new Message("TopicTest" /* Topic */,
"TagA" /* Tag */,
("Hello RocketMQ " i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
// 发送消息到一个Broker
SendResult sendResult = producer.send(msg);
// 通过sendResult返回消息是否成功送达
System.out.printf("%s%n", sendResult);
}
// 如果不再发送消息，关闭Producer实例。
producer.shutdown();
}
}2）发送异步消息
异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。
代码语言：javascript复制public class AsyncProducer {
public static void main(String[] args) throws Exception {
// 实例化消息生产者Producer
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
// 设置NameServer的地址
producer.setNamesrvAddr("localhost:9876");
// 启动Producer实例
producer.start();
producer.setRetryTimesWhenSendAsyncFailed(0);
for (int i = 0; i < 100; i ) {
final int index = i;
// 创建消息，并指定Topic，Tag和消息体
Message msg = new Message("TopicTest",
"TagA",
"OrderID188",
"Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
// SendCallback接收异步返回结果的回调
producer.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.printf("%-10d OK %s %n", index,
sendResult.getMsgId());
}
@Override
public void onException(Throwable e) {
System.out.printf("%-10d Exception %s %n", index, e);
e.printStackTrace();
}
});
}
// 如果不再发送消息，关闭Producer实例。
producer.shutdown();
}
}3）单向发送消息
这种方式主要用在不特别关心发送结果的场景，例如日志发送。
代码语言：javascript复制public class OnewayProducer {
public static void main(String[] args) throws Exception{
// 实例化消息生产者Producer
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
// 设置NameServer的地址
producer.setNamesrvAddr("localhost:9876");
// 启动Producer实例
producer.start();
for (int i = 0; i < 100; i ) {
// 创建消息，并指定Topic，Tag和消息体
Message msg = new Message("TopicTest" /* Topic */,
"TagA" /* Tag */,
("Hello RocketMQ " i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
// 发送单向消息，没有任何返回结果
producer.sendOneway(msg);
}
// 如果不再发送消息，关闭Producer实例。
producer.shutdown();
}
}4.1.2 消费消息
1）负载均衡模式
消费者采用负载均衡方式消费消息，多个消费者共同消费队列消息，每个消费者处理的消息不同
代码语言：javascript复制public static void main(String[] args) throws Exception {
// 实例化消息生产者,指定组名
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
// 指定Namesrv地址信息.
consumer.setNamesrvAddr("localhost:9876");
// 订阅Topic
consumer.subscribe("Test", "*");
//负载均衡模式消费
consumer.setMessageModel(MessageModel.CLUSTERING);
// 注册回调函数，处理消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.printf("%s Receive New Messages: %s %n",
Thread.currentThread().getName(), msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//启动消息者
consumer.start();
System.out.printf("Consumer Started.%n");
}2）广播模式
消费者采用广播的方式消费消息，每个消费者消费的消息都是相同的
代码语言：javascript复制public static void main(String[] args) throws Exception {
// 实例化消息生产者,指定组名
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
// 指定Namesrv地址信息.
consumer.setNamesrvAddr("localhost:9876");
// 订阅Topic
consumer.subscribe("Test", "*");
//广播模式消费
consumer.setMessageModel(MessageModel.BROADCASTING);
// 注册回调函数，处理消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.printf("%s Receive New Messages: %s %n",
Thread.currentThread().getName(), msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//启动消息者
consumer.start();
System.out.printf("Consumer Started.%n");
}4.2 顺序消息
消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为分区有序或者全局有序。
顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。
下面用订单进行分区有序的示例。一个订单的顺序流程是：创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中，消费时，同一个OrderId获取到的肯定是同一个队列。
4.2.1 顺序消息生产
代码语言：javascript复制/**
* Producer，发送顺序消息
*/
public class Producer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
String[] tags = new String[]{"TagA", "TagC", "TagD"};
// 订单列表
List<OrderStep> orderList = new Producer().buildOrders();
Date date = new Date();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String dateStr = sdf.format(date);
for (int i = 0; i < 10; i ) {
// 加个时间前缀
String body = dateStr " Hello RocketMQ " orderList.get(i);
Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" i, body.getBytes());
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Long id = (Long) arg; //根据订单id选择发送queue
long index = id % mqs.size();
return mqs.get((int) index);
}
}, orderList.get(i).getOrderId());//订单id
System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
sendResult.getSendStatus(),
sendResult.getMessageQueue().getQueueId(),
body));
}
producer.shutdown();
}
/**
* 订单的步骤
*/
private static class OrderStep {
private long orderId;
private String desc;
public long getOrderId() {
return orderId;
}
public void setOrderId(long orderId) {
this.orderId = orderId;
}
public String getDesc() {
return desc;
}
public void setDesc(String desc) {
this.desc = desc;
}
@Override
public String toString() {
return "OrderStep{"
"orderId=" orderId
", desc='" desc '''
'}';
}
}
/**
* 生成模拟订单数据
*/
private List<OrderStep> buildOrders() {
List<OrderStep> orderList = new ArrayList<OrderStep>();
OrderStep orderDemo = new OrderStep();
orderDemo.setOrderId(15103111039L);
orderDemo.setDesc("创建");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111065L);
orderDemo.setDesc("创建");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111039L);
orderDemo.setDesc("付款");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103117235L);
orderDemo.setDesc("创建");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111065L);
orderDemo.setDesc("付款");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103117235L);
orderDemo.setDesc("付款");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111065L);
orderDemo.setDesc("完成");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111039L);
orderDemo.setDesc("推送");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103117235L);
orderDemo.setDesc("完成");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111039L);
orderDemo.setDesc("完成");
orderList.add(orderDemo);
return orderList;
}
}4.2.2 顺序消费消息
代码语言：javascript复制/**
* 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）
*/
public class ConsumerInOrder {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("please_rename_unique_group_name_3");
consumer.setNamesrvAddr("127.0.0.1:9876");
/**
* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
* 如果非第一次启动，那么按照上次消费的位置继续消费
*/
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicTest", "TagA || TagC || TagD");
consumer.registerMessageListener(new MessageListenerOrderly() {
Random random = new Random();
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
context.setAutoCommit(true);
for (MessageExt msg : msgs) {
// 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序
System.out.println("consumeThread=" Thread.currentThread().getName() "queueId=" msg.getQueueId() ", content:" new String(msg.getBody()));
}
try {
//模拟业务逻辑处理中...
TimeUnit.SECONDS.sleep(random.nextInt(10));
} catch (Exception e) {
e.printStackTrace();
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started.");
}
}4.3 延时消息
比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。
4.3.1 启动消息消费者
代码语言：javascript复制public class ScheduledMessageConsumer {
public static void main(String[] args) throws Exception {
// 实例化消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
// 订阅Topics
consumer.subscribe("TestTopic", "*");
// 注册消息监听者
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
for (MessageExt message : messages) {
// Print approximate delay time period
System.out.println("Receive message[msgId=" message.getMsgId() "] " (System.currentTimeMillis() - message.getStoreTimestamp()) "ms later");
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// 启动消费者
consumer.start();
}
}4.3.2 发送延时消息
代码语言：javascript复制public class ScheduledMessageProducer {
public static void main(String[] args) throws Exception {
// 实例化一个生产者来产生延时消息
DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
// 启动生产者
producer.start();
int totalMessagesToSend = 100;
for (int i = 0; i < totalMessagesToSend; i ) {
Message message = new Message("TestTopic", ("Hello scheduled message " i).getBytes());
// 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
message.setDelayTimeLevel(3);
// 发送消息
producer.send(message);
}
// 关闭生产者
producer.shutdown();
}
}###4.3.3 验证
您将会看到消息的消费比存储时间晚10秒
4.3.4 使用限制
代码语言：javascript复制// org/apache/rocketmq/store/config/MessageStoreConfig.java
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18
4.4 批量消息
批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic，相同的waitStoreMsgOK，而且不能是延时消息。此外，这一批消息的总大小不应超过4MB。
4.4.1 发送批量消息
如果您每次只发送不超过4MB的消息，则很容易使用批处理，样例如下：
代码语言：javascript复制String topic = "BatchTest";
List<Message> messages = new ArrayList<>();
messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
try {
producer.send(messages);
} catch (Exception e) {
e.printStackTrace();
//处理error
}如果消息的总长度可能大于4MB时，这时候最好把消息进行分割
代码语言：javascript复制public class ListSplitter implements Iterator<List<Message>> {
private final int SIZE_LIMIT = 1024 * 1024 * 4;
private final List<Message> messages;
private int currIndex;
public ListSplitter(List<Message> messages) {
this.messages = messages;
}
@Override
public boolean hasNext() {
return currIndex < messages.size();
}
@Override
public List<Message> next() {
int nextIndex = currIndex;
int totalSize = 0;
for (; nextIndex < messages.size(); nextIndex ) {
Message message = messages.get(nextIndex);
int tmpSize = message.getTopic().length() message.getBody().length;
Map<String, String> properties = message.getProperties();
for (Map.Entry<String, String> entry : properties.entrySet()) {
tmpSize = entry.getKey().length() entry.getValue().length();
}
tmpSize = tmpSize 20; // 增加日志的开销20字节
if (tmpSize > SIZE_LIMIT) {
//单个消息超过了最大的限制
//忽略,否则会阻塞分裂的进程
if (nextIndex - currIndex == 0) {
//假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环
nextIndex ;
}
break;
}
if (tmpSize totalSize > SIZE_LIMIT) {
break;
} else {
totalSize = tmpSize;
}
}
List<Message> subList = messages.subList(currIndex, nextIndex);
currIndex = nextIndex;
return subList;
}
}
//把大的消息分裂成若干个小的消息
ListSplitter splitter = new ListSplitter(messages);
while (splitter.hasNext()) {
try {
List<Message> listItem = splitter.next();
producer.send(listItem);
} catch (Exception e) {
e.printStackTrace();
//处理error
}
}4.5 过滤消息
在大多数情况下，TAG是一个简单而有用的设计，其可以来选择您想要的消息。例如：
代码语言：javascript复制DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");消费者将接收包含TAGA或TAGB或TAGC的消息。但是限制是一个消息只能有一个标签，这对于复杂的场景可能不起作用。在这种情况下，可以使用SQL表达式筛选消息。SQL特性可以通过发送消息时的属性来进行计算。在RocketMQ定义的语法下，可以实现一些简单的逻辑。下面是一个例子：
代码语言：javascript复制------------
| message |
|----------| a > 5 AND b = 'abc'
| a = 10 | --------------------> Gotten
| b = 'abc'|
| c = true |
------------
------------
| message |
|----------| a > 5 AND b = 'abc'
| a = 1 | --------------------> Missed
| b = 'abc'|
| c = true |
------------4.5.1 SQL基本语法
RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。
- 数值比较，比如：>，>=，<，<=，BETWEEN，=；
- 字符比较，比如：=，<>，IN；
- IS NULL 或者 IS NOT NULL；
- 逻辑符号 AND，OR，NOT；
常量支持类型为：
- 数值，比如：123，3.1415；
- 字符，比如：‘abc’，必须用单引号包裹起来；
- NULL，特殊的常量
- 布尔值，TRUE 或 FALSE
只有使用push模式的消费者才能用使用SQL92标准的sql语句，接口如下：
代码语言：javascript复制public void subscribe(finalString topic, final MessageSelector messageSelector)4.5.2 消息生产者
发送消息时，你能通过putUserProperty来设置消息的属性
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
producer.start();
Message msg = new Message("TopicTest",
tag,
("Hello RocketMQ " i).getBytes(RemotingHelper.DEFAULT_CHARSET)
);
// 设置一些属性
msg.putUserProperty("a", String.valueOf(i));
SendResult sendResult = producer.send(msg);
producer.shutdown();4.5.3 消息消费者
用MessageSelector.bySql来使用sql筛选消息
代码语言：javascript复制DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
// 只有订阅的消息有这个属性a, a >=0 and a <= 3
consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();4.6 事务消息
###4.6.1 流程分析
上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程。
####1）事务消息发送及提交
(1) 发送消息（half消息）。
(2) 服务端响应消息写入结果。
(3) 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）。
(4) 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）
2）事务补偿
(1) 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”
(2) Producer收到回查消息，检查回查消息对应的本地事务的状态
(3) 根据本地事务状态，重新Commit或者Rollback
其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。
3）事务消息状态
事务消息共有三种状态，提交状态、回滚状态、中间状态：
- TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
- TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
- TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。
###4.6.1 发送事务消息
1) 创建事务性生产者
使用 TransactionMQProducer类创建生产者，并指定唯一的 ProducerGroup，就可以设置自定义线程池来处理这些检查请求。执行本地事务后、需要根据执行结果对消息队列进行回复。回传的事务状态在请参考前一节。
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
//创建事务监听器
TransactionListener transactionListener = new TransactionListenerImpl();
//创建消息生产者
TransactionMQProducer producer = new TransactionMQProducer("group6");
producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
//生产者这是监听器
producer.setTransactionListener(transactionListener);
//启动消息生产者
producer.start();
String[] tags = new String[]{"TagA", "TagB", "TagC"};
for (int i = 0; i < 3; i ) {
try {
Message msg = new Message("TransactionTopic", tags[i % tags.length], "KEY" i,
("Hello RocketMQ " i).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.sendMessageInTransaction(msg, null);
System.out.printf("%s%n", sendResult);
TimeUnit.SECONDS.sleep(1);
} catch (MQClientException | UnsupportedEncodingException e) {
e.printStackTrace();
}
}
//producer.shutdown();
}
}2）实现事务的监听接口
当发送半消息成功时，我们使用 executeLocalTransaction 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。checkLocalTranscation 方法用于检查本地事务状态，并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。
public class TransactionListenerImpl implements TransactionListener {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
System.out.println("执行本地事务");
if (StringUtils.equals("TagA", msg.getTags())) {
return LocalTransactionState.COMMIT_MESSAGE;
} else if (StringUtils.equals("TagB", msg.getTags())) {
return LocalTransactionState.ROLLBACK_MESSAGE;
} else {
return LocalTransactionState.UNKNOW;
}
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
System.out.println("MQ检查消息Tag【" msg.getTags() "】的本地事务执行结果");
return LocalTransactionState.COMMIT_MESSAGE;
}
}4.6.2 使用限制
- 事务消息不支持延时消息和批量消息。
- 为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的
transactionCheckMax参数来修改此限制。如果已经检查某条消息超过 N 次的话（ N =transactionCheckMax） 则 Broker 将丢弃此消息，并在默认情况下同时打印错误日志。用户可以通过重写AbstractTransactionCheckListener类来修改这个行为。 - 事务消息将在 Broker 配置文件中的参数 transactionMsgTimeout 这样的特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于
transactionMsgTimeout参数。 - 事务性消息可能不止一次被检查或消费。
- 提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。
- 事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。
1. 案例介绍
1.1 业务分析
模拟电商网站购物场景中的【下单】和【支付】业务
1）下单
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-nSfCyNti-1631150365497)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/下单组件图.png)]
- 用户请求订单系统下单
- 订单系统通过RPC调用订单服务下单
- 订单服务调用优惠券服务，扣减优惠券
- 订单服务调用调用库存服务，校验并扣减库存
- 订单服务调用用户服务，扣减用户余额
- 订单服务完成确认订单
2）支付
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-kWIZMwsl-1631150365497)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/支付组件图.png)]
- 用户请求支付系统
- 支付系统调用第三方支付平台API进行发起支付流程
- 用户通过第三方支付平台支付成功后，第三方支付平台回调通知支付系统
- 支付系统调用订单服务修改订单状态
- 支付系统调用积分服务添加积分
- 支付系统调用日志服务记录日志
1.2 问题分析
问题1
用户提交订单后，扣减库存成功、扣减优惠券成功、使用余额成功，但是在确认订单操作失败，需要对库存、库存、余额进行回退。
如何保证数据的完整性？
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ECRf3Z3K-1631150365497)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/下单失败流程图.png)]
使用MQ保证在下单失败后系统数据的完整性
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-wMETFUbh-1631150365497)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/下单时序图(2)].png)
###问题2
用户通过第三方支付平台（支付宝、微信）支付成功后，第三方支付平台要通过回调API异步通知商家支付系统用户支付结果，支付系统根据支付结果修改订单状态、记录支付日志和给用户增加积分。
商家支付系统如何保证在收到第三方支付平台的异步通知时，如何快速给第三方支付凭条做出回应？
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-PBojCZ5e-1631150365498)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/支付流程.png)]
通过MQ进行数据分发，提高系统处理性能
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-kN19YuEg-1631150365498)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/支付成功数据分发流程图.png)]
2. 技术分析
2.1 技术选型
- SpringBoot
- Dubbo
- Zookeeper
- RocketMQ
- Mysql
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LnwZf2Wq-1631150365498)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/项目结构图.png)]
2.2 SpringBoot整合RocketMQ
下载rocketmq-spring项目
将rocketmq-spring安装到本地仓库
代码语言：javascript复制mvn install -Dmaven.skip.test=true2.2.1 消息生产者
1）添加依赖
代码语言：javascript复制<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.0.1.RELEASE</version>
</parent>
<properties>
<rocketmq-spring-boot-starter-version>2.0.3</rocketmq-spring-boot-starter-version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>${rocketmq-spring-boot-starter-version}</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.6</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>2）配置文件
代码语言：javascript复制# application.properties
rocketmq.name-server=192.168.25.135:9876;192.168.25.138:9876
rocketmq.producer.group=my-group3）启动类
代码语言：javascript复制@SpringBootApplication
public class MQProducerApplication {
public static void main(String[] args) {
SpringApplication.run(MQSpringBootApplication.class);
}
}4）测试类
代码语言：javascript复制@RunWith(SpringRunner.class)
@SpringBootTest(classes = {MQSpringBootApplication.class})
public class ProducerTest {
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Test
public void test1(){
rocketMQTemplate.convertAndSend("springboot-mq","hello springboot rocketmq");
}
}2.2.2 消息消费者
1）添加依赖
同消息生产者
2）配置文件
同消息生产者
3）启动类
代码语言：javascript复制@SpringBootApplication
public class MQConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(MQSpringBootApplication.class);
}
}4）消息监听器
代码语言：javascript复制@Slf4j
@Component
@RocketMQMessageListener(topic = "springboot-mq",consumerGroup = "springboot-mq-consumer-1")
public class Consumer implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
log.info("Receive message：" message);
}
}2.3 SpringBoot整合Dubbo
下载dubbo-spring-boot-starter依赖包
将dubbo-spring-boot-starter安装到本地仓库
mvn install -Dmaven.skip.test=true[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-21GxV9r7-1631150365499)(…/文档/ https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/dubbo.png)]
2.3.1 搭建Zookeeper集群
1）准备工作
- 安装JDK
- 将Zookeeper上传到服务器
- 解压Zookeeper，并创建data目录，将conf下的zoo_sample.cfg文件改名为zoo.cfg
- 建立
/user/local/zookeeper-cluster,将解压后的Zookeeper复制到以下三个目录
/usr/local/zookeeper-cluster/zookeeper-1
/usr/local/zookeeper-cluster/zookeeper-2
/usr/local/zookeeper-cluster/zookeeper-3- 配置每一个 Zookeeper 的 dataDir（zoo.cfg） clientPort 分别为 2181 2182 2183
修改
/usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg
clientPort=2181
dataDir=/usr/local/zookeeper-cluster/zookeeper-1/data 修改/usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg
代码语言：javascript复制clientPort=2182
dataDir=/usr/local/zookeeper-cluster/zookeeper-2/data 修改/usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg
代码语言：javascript复制clientPort=2183
dataDir=/usr/local/zookeeper-cluster/zookeeper-3/data2）配置集群
- 在每个 zookeeper 的 data 目录下创建一个 myid 文件，内容分别是 1、2、3 。这个文件就是记录每个服务器的 ID
- 在每一个 zookeeper 的 zoo.cfg 配置客户端访问端口（clientPort）和集群服务器 IP 列表。 集群服务器 IP 列表如下
server.1=192.168.25.140:2881:3881
server.2=192.168.25.140:2882:3882
server.3=192.168.25.140:2883:3883解释：server.服务器 ID=服务器 IP 地址：服务器之间通信端口：服务器之间投票选举端口
3）启动集群
启动集群就是分别启动每个实例。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Vjn93ehm-1631150365499)(…/文档/ https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/zk.png)]
2.3.2 RPC服务接口
代码语言：javascript复制public interface IUserService {
public String sayHello(String name);
}2.3.3 服务提供者
1）添加依赖
代码语言：javascript复制<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.0.1.RELEASE</version>
</parent>
<dependencies>
<!--dubbo-->
<dependency>
<groupId>com.alibaba.spring.boot</groupId>
<artifactId>dubbo-spring-boot-starter</artifactId>
<version>2.0.0</version>
</dependency>
<!--spring-boot-stater-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
<exclusions>
<exclusion>
<artifactId>log4j-to-slf4j</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
</exclusions>
</dependency>
<!--zookeeper-->
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.10</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.9</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<!--API-->
<dependency>
<groupId>com.itheima.demo</groupId>
<artifactId>dubbo-api</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
</dependencies>2）配置文件
代码语言：javascript复制# application.properties
spring.application.name=dubbo-demo-provider
spring.dubbo.application.id=dubbo-demo-provider
spring.dubbo.application.name=dubbo-demo-provider
spring.dubbo.registry.address=zookeeper://192.168.25.140:2181;zookeeper://192.168.25.140:2182;zookeeper://192.168.25.140:2183
spring.dubbo.server=true
spring.dubbo.protocol.name=dubbo
spring.dubbo.protocol.port=208803）启动类
代码语言：javascript复制@EnableDubboConfiguration
@SpringBootApplication
public class ProviderBootstrap {
public static void main(String[] args) throws IOException {
SpringApplication.run(ProviderBootstrap.class,args);
}
}4）服务实现
代码语言：javascript复制@Component
@Service(interfaceClass = IUserService.class)
public class UserServiceImpl implements IUserService{
@Override
public String sayHello(String name) {
return "hello:" name;
}
}2.3.4 服务消费者
1）添加依赖
代码语言：javascript复制<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.0.1.RELEASE</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--dubbo-->
<dependency>
<groupId>com.alibaba.spring.boot</groupId>
<artifactId>dubbo-spring-boot-starter</artifactId>
<version>2.0.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
<exclusions>
<exclusion>
<artifactId>log4j-to-slf4j</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
</exclusions>
</dependency>
<!--zookeeper-->
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.10</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.9</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<!--API-->
<dependency>
<groupId>com.itheima.demo</groupId>
<artifactId>dubbo-api</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
</dependencies>2）配置文件
代码语言：javascript复制# application.properties
spring.application.name=dubbo-demo-consumer
spring.dubbo.application.name=dubbo-demo-consumer
spring.dubbo.application.id=dubbo-demo-consumer
spring.dubbo.registry.address=zookeeper://192.168.25.140:2181;zookeeper://192.168.25.140:2182;zookeeper://192.168.25.140:21833）启动类
代码语言：javascript复制@EnableDubboConfiguration
@SpringBootApplication
public class ConsumerBootstrap {
public static void main(String[] args) {
SpringApplication.run(ConsumerBootstrap.class);
}
}4）Controller
代码语言：javascript复制@RestController
@RequestMapping("/user")
public class UserController {
@Reference
private IUserService userService;
@RequestMapping("/sayHello")
public String sayHello(String name){
return userService.sayHello(name);
}
}3. 环境搭建
3.1 数据库
1）优惠券表
Field | Type | Comment |
|---|---|---|
coupon_id | bigint(50) NOT NULL | 优惠券ID |
coupon_price | decimal(10,2) NULL | 优惠券金额 |
user_id | bigint(50) NULL | 用户ID |
order_id | bigint(32) NULL | 订单ID |
is_used | int(1) NULL | 是否使用 0未使用 1已使用 |
used_time | timestamp NULL | 使用时间 |
2）商品表
Field | Type | Comment |
|---|---|---|
goods_id | bigint(50) NOT NULL | 主键 |
goods_name | varchar(255) NULL | 商品名称 |
goods_number | int(11) NULL | 商品库存 |
goods_price | decimal(10,2) NULL | 商品价格 |
goods_desc | varchar(255) NULL | 商品描述 |
add_time | timestamp NULL | 添加时间 |
3）订单表
Field | Type | Comment |
|---|---|---|
order_id | bigint(50) NOT NULL | 订单ID |
user_id | bigint(50) NULL | 用户ID |
order_status | int(1) NULL | 订单状态 0未确认 1已确认 2已取消 3无效 4退款 |
pay_status | int(1) NULL | 支付状态 0未支付 1支付中 2已支付 |
shipping_status | int(1) NULL | 发货状态 0未发货 1已发货 2已退货 |
address | varchar(255) NULL | 收货地址 |
consignee | varchar(255) NULL | 收货人 |
goods_id | bigint(50) NULL | 商品ID |
goods_number | int(11) NULL | 商品数量 |
goods_price | decimal(10,2) NULL | 商品价格 |
goods_amount | decimal(10,0) NULL | 商品总价 |
shipping_fee | decimal(10,2) NULL | 运费 |
order_amount | decimal(10,2) NULL | 订单价格 |
coupon_id | bigint(50) NULL | 优惠券ID |
coupon_paid | decimal(10,2) NULL | 优惠券 |
money_paid | decimal(10,2) NULL | 已付金额 |
pay_amount | decimal(10,2) NULL | 支付金额 |
add_time | timestamp NULL | 创建时间 |
confirm_time | timestamp NULL | 订单确认时间 |
pay_time | timestamp NULL | 支付时间 |
4）订单商品日志表
Field | Type | Comment |
|---|---|---|
goods_id | int(11) NOT NULL | 商品ID |
order_id | varchar(32) NOT NULL | 订单ID |
goods_number | int(11) NULL | 库存数量 |
log_time | datetime NULL | 记录时间 |
5）用户表
Field | Type | Comment |
|---|---|---|
user_id | bigint(50) NOT NULL | 用户ID |
user_name | varchar(255) NULL | 用户姓名 |
user_password | varchar(255) NULL | 用户密码 |
user_mobile | varchar(255) NULL | 手机号 |
user_score | int(11) NULL | 积分 |
user_reg_time | timestamp NULL | 注册时间 |
user_money | decimal(10,0) NULL | 用户余额 |
6）用户余额日志表
Field | Type | Comment |
|---|---|---|
user_id | bigint(50) NOT NULL | 用户ID |
order_id | bigint(50) NOT NULL | 订单ID |
money_log_type | int(1) NOT NULL | 日志类型 1订单付款 2 订单退款 |
use_money | decimal(10,2) NULL | 操作金额 |
create_time | timestamp NULL | 日志时间 |
7）订单支付表
Field | Type | Comment |
|---|---|---|
pay_id | bigint(50) NOT NULL | 支付编号 |
order_id | bigint(50) NULL | 订单编号 |
pay_amount | decimal(10,2) NULL | 支付金额 |
is_paid | int(1) NULL | 是否已支付 1否 2是 |
8）MQ消息生产表
Field | Type | Comment |
|---|---|---|
id | varchar(100) NOT NULL | 主键 |
group_name | varchar(100) NULL | 生产者组名 |
msg_topic | varchar(100) NULL | 消息主题 |
msg_tag | varchar(100) NULL | Tag |
msg_key | varchar(100) NULL | Key |
msg_body | varchar(500) NULL | 消息内容 |
msg_status | int(1) NULL | 0:未处理;1:已经处理 |
create_time | timestamp NOT NULL | 记录时间 |
###9）MQ消息消费表
Field | Type | Comment |
|---|---|---|
msg_id | varchar(50) NULL | 消息ID |
group_name | varchar(100) NOT NULL | 消费者组名 |
msg_tag | varchar(100) NOT NULL | Tag |
msg_key | varchar(100) NOT NULL | Key |
msg_body | varchar(500) NULL | 消息体 |
consumer_status | int(1) NULL | 0:正在处理;1:处理成功;2:处理失败 |
consumer_times | int(1) NULL | 消费次数 |
consumer_timestamp | timestamp NULL | 消费时间 |
remark | varchar(500) NULL | 备注 |
3.2 项目初始化
shop系统基于Maven进行项目管理
3.1.1 工程浏览
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lmc2uSxw-1631150365500)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/项目初始化.png)]
- 父工程：shop-parent
- 订单系统：shop-order-web
- 支付系统：shop-pay-web
- 优惠券服务：shop-coupon-service
- 订单服务：shop-order-service
- 支付服务：shop-pay-service
- 商品服务：shop-goods-service
- 用户服务：shop-user-service
- 实体类：shop-pojo
- 持久层：shop-dao
- 接口层：shop-api
- 工具工程：shop-common
共12个系统
3.1.2 工程关系
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-h0XnFagV-1631150365500)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/项目结构图.png)]
3.3 Mybatis逆向工程使用
1）代码生成
使用Mybatis逆向工程针对数据表生成CURD持久层代码
###2）代码导入
- 将实体类导入到shop-pojo工程
- 在服务层工程中导入对应的Mapper类和对应配置文件
3.4 公共类介绍
- ID生成器 IDWorker：Twitter雪花算法
- 异常处理类 CustomerException：自定义异常类 CastException：异常抛出类
- 常量类 ShopCode：系统状态类
- 响应实体类 Result：封装响应状态和响应信息
4. 下单业务
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-RsR0KCmH-1631150365500)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/下单时序图(2)].png)
4.1 下单基本流程
1）接口定义
- IOrderService
public interface IOrderService {
/**
* 确认订单
* @param order
* @return Result
*/
Result confirmOrder(TradeOrder order);
}###2）业务类实现
代码语言：javascript复制@Slf4j
@Component
@Service(interfaceClass = IOrderService.class)
public class OrderServiceImpl implements IOrderService {
@Override
public Result confirmOrder(TradeOrder order) {
//1.校验订单
//2.生成预订单
try {
//3.扣减库存
//4.扣减优惠券
//5.使用余额
//6.确认订单
//7.返回成功状态
} catch (Exception e) {
//1.确认订单失败,发送消息
//2.返回失败状态
}
}
}###3）校验订单
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZubQ2Ykk-1631150365501)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/校验订单(2)].png)
代码语言：javascript复制private void checkOrder(TradeOrder order) {
//1.校验订单是否存在
if(order==null){
CastException.cast(ShopCode.SHOP_ORDER_INVALID);
}
//2.校验订单中的商品是否存在
TradeGoods goods = goodsService.findOne(order.getGoodsId());
if(goods==null){
CastException.cast(ShopCode.SHOP_GOODS_NO_EXIST);
}
//3.校验下单用户是否存在
TradeUser user = userService.findOne(order.getUserId());
if(user==null){
CastException.cast(ShopCode.SHOP_USER_NO_EXIST);
}
//4.校验商品单价是否合法
if(order.getGoodsPrice().compareTo(goods.getGoodsPrice())!=0){
CastException.cast(ShopCode.SHOP_GOODS_PRICE_INVALID);
}
//5.校验订单商品数量是否合法
if(order.getGoodsNumber()>=goods.getGoodsNumber()){
CastException.cast(ShopCode.SHOP_GOODS_NUM_NOT_ENOUGH);
}
log.info("校验订单通过");
}###4）生成预订单
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fnAcdgB3-1631150365501)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/生成预订单.png)]
代码语言：javascript复制private Long savePreOrder(TradeOrder order) {
//1.设置订单状态为不可见
order.setOrderStatus(ShopCode.SHOP_ORDER_NO_CONFIRM.getCode());
//2.订单ID
order.setOrderId(idWorker.nextId());
//核算运费是否正确
BigDecimal shippingFee = calculateShippingFee(order.getOrderAmount());
if (order.getShippingFee().compareTo(shippingFee) != 0) {
CastException.cast(ShopCode.SHOP_ORDER_SHIPPINGFEE_INVALID);
}
//3.计算订单总价格是否正确
BigDecimal orderAmount = order.getGoodsPrice().multiply(new BigDecimal(order.getGoodsNumber()));
orderAmount.add(shippingFee);
if (orderAmount.compareTo(order.getOrderAmount()) != 0) {
CastException.cast(ShopCode.SHOP_ORDERAMOUNT_INVALID);
}
//4.判断优惠券信息是否合法
Long couponId = order.getCouponId();
if (couponId != null) {
TradeCoupon coupon = couponService.findOne(couponId);
//优惠券不存在
if (coupon == null) {
CastException.cast(ShopCode.SHOP_COUPON_NO_EXIST);
}
//优惠券已经使用
if ((ShopCode.SHOP_COUPON_ISUSED.getCode().toString())
.equals(coupon.getIsUsed().toString())) {
CastException.cast(ShopCode.SHOP_COUPON_INVALIED);
}
order.setCouponPaid(coupon.getCouponPrice());
} else {
order.setCouponPaid(BigDecimal.ZERO);
}
//5.判断余额是否正确
BigDecimal moneyPaid = order.getMoneyPaid();
if (moneyPaid != null) {
//比较余额是否大于0
int r = order.getMoneyPaid().compareTo(BigDecimal.ZERO);
//余额小于0
if (r == -1) {
CastException.cast(ShopCode.SHOP_MONEY_PAID_LESS_ZERO);
}
//余额大于0
if (r == 1) {
//查询用户信息
TradeUser user = userService.findOne(order.getUserId());
if (user == null) {
CastException.cast(ShopCode.SHOP_USER_NO_EXIST);
}
//比较余额是否大于用户账户余额
if (user.getUserMoney().compareTo(order.getMoneyPaid().longValue()) == -1) {
CastException.cast(ShopCode.SHOP_MONEY_PAID_INVALID);
}
order.setMoneyPaid(order.getMoneyPaid());
}
} else {
order.setMoneyPaid(BigDecimal.ZERO);
}
//计算订单支付总价
order.setPayAmount(orderAmount.subtract(order.getCouponPaid())
.subtract(order.getMoneyPaid()));
//设置订单添加时间
order.setAddTime(new Date());
//保存预订单
int r = orderMapper.insert(order);
if (ShopCode.SHOP_SUCCESS.getCode() != r) {
CastException.cast(ShopCode.SHOP_ORDER_SAVE_ERROR);
}
log.info("订单:[" order.getOrderId() "]预订单生成成功");
return order.getOrderId();
}###5）扣减库存
- 通过dubbo调用商品服务完成扣减库存
private void reduceGoodsNum(TradeOrder order) {
TradeGoodsNumberLog goodsNumberLog = new TradeGoodsNumberLog();
goodsNumberLog.setGoodsId(order.getGoodsId());
goodsNumberLog.setOrderId(order.getOrderId());
goodsNumberLog.setGoodsNumber(order.getGoodsNumber());
Result result = goodsService.reduceGoodsNum(goodsNumberLog);
if (result.getSuccess().equals(ShopCode.SHOP_FAIL.getSuccess())) {
CastException.cast(ShopCode.SHOP_REDUCE_GOODS_NUM_FAIL);
}
log.info("订单:[" order.getOrderId() "]扣减库存[" order.getGoodsNumber() "个]成功");
}- 商品服务GoodsService扣减库存
@Override
public Result reduceGoodsNum(TradeGoodsNumberLog goodsNumberLog) {
if (goodsNumberLog == null ||
goodsNumberLog.getGoodsNumber() == null ||
goodsNumberLog.getOrderId() == null ||
goodsNumberLog.getGoodsNumber() == null ||
goodsNumberLog.getGoodsNumber().intValue() <= 0) {
CastException.cast(ShopCode.SHOP_REQUEST_PARAMETER_VALID);
}
TradeGoods goods = goodsMapper.selectByPrimaryKey(goodsNumberLog.getGoodsId());
if(goods.getGoodsNumber()<goodsNumberLog.getGoodsNumber()){
//库存不足
CastException.cast(ShopCode.SHOP_GOODS_NUM_NOT_ENOUGH);
}
//减库存
goods.setGoodsNumber(goods.getGoodsNumber()-goodsNumberLog.getGoodsNumber());
goodsMapper.updateByPrimaryKey(goods);
//记录库存操作日志
goodsNumberLog.setGoodsNumber(-(goodsNumberLog.getGoodsNumber()));
goodsNumberLog.setLogTime(new Date());
goodsNumberLogMapper.insert(goodsNumberLog);
return new Result(ShopCode.SHOP_SUCCESS.getSuccess(),ShopCode.SHOP_SUCCESS.getMessage());
}###6）扣减优惠券
- 通过dubbo完成扣减优惠券
private void changeCoponStatus(TradeOrder order) {
//判断用户是否使用优惠券
if (!StringUtils.isEmpty(order.getCouponId())) {
//封装优惠券对象
TradeCoupon coupon = couponService.findOne(order.getCouponId());
coupon.setIsUsed(ShopCode.SHOP_COUPON_ISUSED.getCode());
coupon.setUsedTime(new Date());
coupon.setOrderId(order.getOrderId());
Result result = couponService.changeCouponStatus(coupon);
//判断执行结果
if (result.getSuccess().equals(ShopCode.SHOP_FAIL.getSuccess())) {
//优惠券使用失败
CastException.cast(ShopCode.SHOP_COUPON_USE_FAIL);
}
log.info("订单:[" order.getOrderId() "]使用扣减优惠券[" coupon.getCouponPrice() "元]成功");
}
}- 优惠券服务CouponService更改优惠券状态
@Override
public Result changeCouponStatus(TradeCoupon coupon) {
try {
//判断请求参数是否合法
if (coupon == null || StringUtils.isEmpty(coupon.getCouponId())) {
CastException.cast(ShopCode.SHOP_REQUEST_PARAMETER_VALID);
}
//更新优惠券状态为已使用
couponMapper.updateByPrimaryKey(coupon);
return new Result(ShopCode.SHOP_SUCCESS.getSuccess(), ShopCode.SHOP_SUCCESS.getMessage());
} catch (Exception e) {
return new Result(ShopCode.SHOP_FAIL.getSuccess(), ShopCode.SHOP_FAIL.getMessage());
}
}###7）扣减用户余额
- 通过用户服务完成扣减余额
private void reduceMoneyPaid(TradeOrder order) {
//判断订单中使用的余额是否合法
if (order.getMoneyPaid() != null && order.getMoneyPaid().compareTo(BigDecimal.ZERO) == 1) {
TradeUserMoneyLog userMoneyLog = new TradeUserMoneyLog();
userMoneyLog.setOrderId(order.getOrderId());
userMoneyLog.setUserId(order.getUserId());
userMoneyLog.setUseMoney(order.getMoneyPaid());
userMoneyLog.setMoneyLogType(ShopCode.SHOP_USER_MONEY_PAID.getCode());
//扣减余额
Result result = userService.changeUserMoney(userMoneyLog);
if (result.getSuccess().equals(ShopCode.SHOP_FAIL.getSuccess())) {
CastException.cast(ShopCode.SHOP_USER_MONEY_REDUCE_FAIL);
}
log.info("订单:[" order.getOrderId() "扣减余额[" order.getMoneyPaid() "元]成功]");
}
}- 用户服务UserService,更新余额
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-AQjY46Di-1631150365502)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/更改用户余额.png)]
代码语言：javascript复制@Override
public Result changeUserMoney(TradeUserMoneyLog userMoneyLog) {
//判断请求参数是否合法
if (userMoneyLog == null
|| userMoneyLog.getUserId() == null
|| userMoneyLog.getUseMoney() == null
|| userMoneyLog.getOrderId() == null
|| userMoneyLog.getUseMoney().compareTo(BigDecimal.ZERO) <= 0) {
CastException.cast(ShopCode.SHOP_REQUEST_PARAMETER_VALID);
}
//查询该订单是否存在付款记录
TradeUserMoneyLogExample userMoneyLogExample = new TradeUserMoneyLogExample();
userMoneyLogExample.createCriteria()
.andUserIdEqualTo(userMoneyLog.getUserId())
.andOrderIdEqualTo(userMoneyLog.getOrderId());
int count = userMoneyLogMapper.countByExample(userMoneyLogExample);
TradeUser tradeUser = new TradeUser();
tradeUser.setUserId(userMoneyLog.getUserId());
tradeUser.setUserMoney(userMoneyLog.getUseMoney().longValue());
//判断余额操作行为
//【付款操作】
if (userMoneyLog.getMoneyLogType().equals(ShopCode.SHOP_USER_MONEY_PAID.getCode())) {
//订单已经付款，则抛异常
if (count > 0) {
CastException.cast(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY);
}
//用户账户扣减余额
userMapper.reduceUserMoney(tradeUser);
}
//【退款操作】
if (userMoneyLog.getMoneyLogType().equals(ShopCode.SHOP_USER_MONEY_REFUND.getCode())) {
//如果订单未付款,则不能退款,抛异常
if (count == 0) {
CastException.cast(ShopCode.SHOP_ORDER_PAY_STATUS_NO_PAY);
}
//防止多次退款
userMoneyLogExample = new TradeUserMoneyLogExample();
userMoneyLogExample.createCriteria()
.andUserIdEqualTo(userMoneyLog.getUserId())
.andOrderIdEqualTo(userMoneyLog.getOrderId())
.andMoneyLogTypeEqualTo(ShopCode.SHOP_USER_MONEY_REFUND.getCode());
count = userMoneyLogMapper.countByExample(userMoneyLogExample);
if (count > 0) {
CastException.cast(ShopCode.SHOP_USER_MONEY_REFUND_ALREADY);
}
//用户账户添加余额
userMapper.addUserMoney(tradeUser);
}
//记录用户使用余额日志
userMoneyLog.setCreateTime(new Date());
userMoneyLogMapper.insert(userMoneyLog);
return new Result(ShopCode.SHOP_SUCCESS.getSuccess(),ShopCode.SHOP_SUCCESS.getMessage());
}###8）确认订单
代码语言：javascript复制private void updateOrderStatus(TradeOrder order) {
order.setOrderStatus(ShopCode.SHOP_ORDER_CONFIRM.getCode());
order.setPayStatus(ShopCode.SHOP_ORDER_PAY_STATUS_NO_PAY.getCode());
order.setConfirmTime(new Date());
int r = orderMapper.updateByPrimaryKey(order);
if (r <= 0) {
CastException.cast(ShopCode.SHOP_ORDER_CONFIRM_FAIL);
}
log.info("订单:[" order.getOrderId() "]状态修改成功");
}9）小结
代码语言：javascript复制@Override
public Result confirmOrder(TradeOrder order) {
//1.校验订单
checkOrder(order);
//2.生成预订单
Long orderId = savePreOrder(order);
order.setOrderId(orderId);
try {
//3.扣减库存
reduceGoodsNum(order);
//4.扣减优惠券
changeCoponStatus(order);
//5.使用余额
reduceMoneyPaid(order);
//6.确认订单
updateOrderStatus(order);
log.info("订单:[" orderId "]确认成功");
return new Result(ShopCode.SHOP_SUCCESS.getSuccess(), ShopCode.SHOP_SUCCESS.getMessage());
} catch (Exception e) {
//确认订单失败,发送消息
...
return new Result(ShopCode.SHOP_FAIL.getSuccess(), ShopCode.SHOP_FAIL.getMessage());
}
}4.2 失败补偿机制
4.2.1 消息发送方
- 配置RocketMQ属性值
rocketmq.name-server=192.168.25.135:9876;192.168.25.138:9876
rocketmq.producer.group=orderProducerGroup
mq.order.consumer.group.name=order_orderTopic_cancel_group
mq.order.topic=orderTopic
mq.order.tag.confirm=order_confirm
mq.order.tag.cancel=order_cancel- 注入模板类和属性值信息
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Value("${mq.order.topic}")
private String topic;
@Value("${mq.order.tag.cancel}")
private String cancelTag;- 发送下单失败消息
@Override
public Result confirmOrder(TradeOrder order) {
//1.校验订单
//2.生成预订
try {
//3.扣减库存
//4.扣减优惠券
//5.使用余额
//6.确认订单
} catch (Exception e) {
//确认订单失败,发送消息
CancelOrderMQ cancelOrderMQ = new CancelOrderMQ();
cancelOrderMQ.setOrderId(order.getOrderId());
cancelOrderMQ.setCouponId(order.getCouponId());
cancelOrderMQ.setGoodsId(order.getGoodsId());
cancelOrderMQ.setGoodsNumber(order.getGoodsNumber());
cancelOrderMQ.setUserId(order.getUserId());
cancelOrderMQ.setUserMoney(order.getMoneyPaid());
try {
sendMessage(topic,
cancelTag,
cancelOrderMQ.getOrderId().toString(),
JSON.toJSONString(cancelOrderMQ));
} catch (Exception e1) {
e1.printStackTrace();
CastException.cast(ShopCode.SHOP_MQ_SEND_MESSAGE_FAIL);
}
return new Result(ShopCode.SHOP_FAIL.getSuccess(), ShopCode.SHOP_FAIL.getMessage());
}
}private void sendMessage(String topic, String tags, String keys, String body) throws Exception {
//判断Topic是否为空
if (StringUtils.isEmpty(topic)) {
CastException.cast(ShopCode.SHOP_MQ_TOPIC_IS_EMPTY);
}
//判断消息内容是否为空
if (StringUtils.isEmpty(body)) {
CastException.cast(ShopCode.SHOP_MQ_MESSAGE_BODY_IS_EMPTY);
}
//消息体
Message message = new Message(topic, tags, keys, body.getBytes());
//发送消息
rocketMQTemplate.getProducer().send(message);
}4.2.2 消费接收方
- 配置RocketMQ属性值
rocketmq.name-server=192.168.25.135:9876;192.168.25.138:9876
mq.order.consumer.group.name=order_orderTopic_cancel_group
mq.order.topic=orderTopic- 创建监听类，消费消息
@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",
consumerGroup = "${mq.order.consumer.group.name}",
messageModel = MessageModel.BROADCASTING)
public class CancelOrderConsumer implements RocketMQListener<MessageExt>{
@Override
public void onMessage(MessageExt messageExt) {
...
}
}1）回退库存
- 流程分析
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-epHjRHqK-1631150365502)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/回退库存.png)]
- 消息消费者
@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup = "${mq.order.consumer.group.name}",messageModel = MessageModel.BROADCASTING )
public class CancelMQListener implements RocketMQListener<MessageExt>{
@Value("${mq.order.consumer.group.name}")
private String groupName;
@Autowired
private TradeGoodsMapper goodsMapper;
@Autowired
private TradeMqConsumerLogMapper mqConsumerLogMapper;
@Autowired
private TradeGoodsNumberLogMapper goodsNumberLogMapper;
@Override
public void onMessage(MessageExt messageExt) {
String msgId=null;
String tags=null;
String keys=null;
String body=null;
try {
//1. 解析消息内容
msgId = messageExt.getMsgId();
tags= messageExt.getTags();
keys= messageExt.getKeys();
body= new String(messageExt.getBody(),"UTF-8");
log.info("接受消息成功");
//2. 查询消息消费记录
TradeMqConsumerLogKey primaryKey = new TradeMqConsumerLogKey();
primaryKey.setMsgTag(tags);
primaryKey.setMsgKey(keys);
primaryKey.setGroupName(groupName);
TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(primaryKey);
if(mqConsumerLog!=null){
//3. 判断如果消费过...
//3.1 获得消息处理状态
Integer status = mqConsumerLog.getConsumerStatus();
//处理过...返回
if(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode().intValue()==status.intValue()){
log.info("消息:" msgId ",已经处理过");
return;
}
//正在处理...返回
if(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode().intValue()==status.intValue()){
log.info("消息:" msgId ",正在处理");
return;
}
//处理失败
if(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode().intValue()==status.intValue()){
//获得消息处理次数
Integer times = mqConsumerLog.getConsumerTimes();
if(times>3){
log.info("消息:" msgId ",消息处理超过3次,不能再进行处理了");
return;
}
mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
//使用数据库乐观锁更新
TradeMqConsumerLogExample example = new TradeMqConsumerLogExample();
TradeMqConsumerLogExample.Criteria criteria = example.createCriteria();
criteria.andMsgTagEqualTo(mqConsumerLog.getMsgTag());
criteria.andMsgKeyEqualTo(mqConsumerLog.getMsgKey());
criteria.andGroupNameEqualTo(groupName);
criteria.andConsumerTimesEqualTo(mqConsumerLog.getConsumerTimes());
int r = mqConsumerLogMapper.updateByExampleSelective(mqConsumerLog, example);
if(r<=0){
//未修改成功,其他线程并发修改
log.info("并发修改,稍后处理");
}
}
}else{
//4. 判断如果没有消费过...
mqConsumerLog = new TradeMqConsumerLog();
mqConsumerLog.setMsgTag(tags);
mqConsumerLog.setMsgKey(keys);
mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
mqConsumerLog.setMsgBody(body);
mqConsumerLog.setMsgId(msgId);
mqConsumerLog.setConsumerTimes(0);
//将消息处理信息添加到数据库
mqConsumerLogMapper.insert(mqConsumerLog);
}
//5. 回退库存
MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
Long goodsId = mqEntity.getGoodsId();
TradeGoods goods = goodsMapper.selectByPrimaryKey(goodsId);
goods.setGoodsNumber(goods.getGoodsNumber() mqEntity.getGoodsNum());
goodsMapper.updateByPrimaryKey(goods);
//记录库存操作日志
TradeGoodsNumberLog goodsNumberLog = new TradeGoodsNumberLog();
goodsNumberLog.setOrderId(mqEntity.getOrderId());
goodsNumberLog.setGoodsId(goodsId);
goodsNumberLog.setGoodsNumber(mqEntity.getGoodsNum());
goodsNumberLog.setLogTime(new Date());
goodsNumberLogMapper.insert(goodsNumberLog);
//6. 将消息的处理状态改为成功
mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode());
mqConsumerLog.setConsumerTimestamp(new Date());
mqConsumerLogMapper.updateByPrimaryKey(mqConsumerLog);
log.info("回退库存成功");
} catch (Exception e) {
e.printStackTrace();
TradeMqConsumerLogKey primaryKey = new TradeMqConsumerLogKey();
primaryKey.setMsgTag(tags);
primaryKey.setMsgKey(keys);
primaryKey.setGroupName(groupName);
TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(primaryKey);
if(mqConsumerLog==null){
//数据库未有记录
mqConsumerLog = new TradeMqConsumerLog();
mqConsumerLog.setMsgTag(tags);
mqConsumerLog.setMsgKey(keys);
mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode());
mqConsumerLog.setMsgBody(body);
mqConsumerLog.setMsgId(msgId);
mqConsumerLog.setConsumerTimes(1);
mqConsumerLogMapper.insert(mqConsumerLog);
}else{
mqConsumerLog.setConsumerTimes(mqConsumerLog.getConsumerTimes() 1);
mqConsumerLogMapper.updateByPrimaryKeySelective(mqConsumerLog);
}
}
}
}2）回退优惠券
代码语言：javascript复制@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup = "${mq.order.consumer.group.name}",messageModel = MessageModel.BROADCASTING )
public class CancelMQListener implements RocketMQListener<MessageExt>{
@Autowired
private TradeCouponMapper couponMapper;
@Override
public void onMessage(MessageExt message) {
try {
//1. 解析消息内容
String body = new String(message.getBody(), "UTF-8");
MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
log.info("接收到消息");
//2. 查询优惠券信息
TradeCoupon coupon = couponMapper.selectByPrimaryKey(mqEntity.getCouponId());
//3.更改优惠券状态
coupon.setUsedTime(null);
coupon.setIsUsed(ShopCode.SHOP_COUPON_UNUSED.getCode());
coupon.setOrderId(null);
couponMapper.updateByPrimaryKey(coupon);
log.info("回退优惠券成功");
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
log.error("回退优惠券失败");
}
}
}3）回退余额
代码语言：javascript复制@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup = "${mq.order.consumer.group.name}",messageModel = MessageModel.BROADCASTING )
public class CancelMQListener implements RocketMQListener<MessageExt>{
@Autowired
private IUserService userService;
@Override
public void onMessage(MessageExt messageExt) {
try {
//1.解析消息
String body = new String(messageExt.getBody(), "UTF-8");
MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
log.info("接收到消息");
if(mqEntity.getUserMoney()!=null && mqEntity.getUserMoney().compareTo(BigDecimal.ZERO)>0){
//2.调用业务层,进行余额修改
TradeUserMoneyLog userMoneyLog = new TradeUserMoneyLog();
userMoneyLog.setUseMoney(mqEntity.getUserMoney());
userMoneyLog.setMoneyLogType(ShopCode.SHOP_USER_MONEY_REFUND.getCode());
userMoneyLog.setUserId(mqEntity.getUserId());
userMoneyLog.setOrderId(mqEntity.getOrderId());
userService.updateMoneyPaid(userMoneyLog);
log.info("余额回退成功");
}
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
log.error("余额回退失败");
}
}
}4）取消订单
代码语言：javascript复制@Override
public void onMessage(MessageExt messageExt) {
String body = new String(messageExt.getBody(), "UTF-8");
String msgId = messageExt.getMsgId();
String tags = messageExt.getTags();
String keys = messageExt.getKeys();
log.info("CancelOrderProcessor receive message:" messageExt);
CancelOrderMQ cancelOrderMQ = JSON.parseObject(body, CancelOrderMQ.class);
TradeOrder order = orderService.findOne(cancelOrderMQ.getOrderId());
order.setOrderStatus(ShopCode.SHOP_ORDER_CANCEL.getCode());
orderService.changeOrderStatus(order);
log.info("订单:[" order.getOrderId() "]状态设置为取消");
return order;
}4.3 测试
1）准备测试环境
代码语言：javascript复制@RunWith(SpringRunner.class)
@SpringBootTest(classes = ShopOrderServiceApplication.class)
public class OrderTest {
@Autowired
private IOrderService orderService;
}###1）准备测试数据
- 用户数据
- 商品数据
- 优惠券数据
###2）测试下单成功流程
代码语言：javascript复制@Test
public void add(){
Long goodsId=XXXL;
Long userId=XXXL;
Long couponId=XXXL;
TradeOrder order = new TradeOrder();
order.setGoodsId(goodsId);
order.setUserId(userId);
order.setGoodsNumber(1);
order.setAddress("北京");
order.setGoodsPrice(new BigDecimal("5000"));
order.setOrderAmount(new BigDecimal("5000"));
order.setMoneyPaid(new BigDecimal("100"));
order.setCouponId(couponId);
order.setShippingFee(new BigDecimal(0));
orderService.confirmOrder(order);
}执行完毕后,查看数据库中用户的余额、优惠券数据，及订单的状态数据
###3）测试下单失败流程
代码同上。
执行完毕后，查看用户的余额、优惠券数据是否发生更改，订单的状态是否为取消。
5. 支付业务
5.1 创建支付订单
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FCYcmPF8-1631150365502)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/创建支付订单.png)]
代码语言：javascript复制public Result createPayment(TradePay tradePay) {
//查询订单支付状态
try {
TradePayExample payExample = new TradePayExample();
TradePayExample.Criteria criteria = payExample.createCriteria();
criteria.andOrderIdEqualTo(tradePay.getOrderId());
criteria.andIsPaidEqualTo(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());
int count = tradePayMapper.countByExample(payExample);
if (count > 0) {
CastException.cast(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY);
}
long payId = idWorker.nextId();
tradePay.setPayId(payId);
tradePay.setIsPaid(ShopCode.SHOP_ORDER_PAY_STATUS_NO_PAY.getCode());
tradePayMapper.insert(tradePay);
log.info("创建支付订单成功:" payId);
} catch (Exception e) {
return new Result(ShopCode.SHOP_FAIL.getSuccess(), ShopCode.SHOP_FAIL.getMessage());
}
return new Result(ShopCode.SHOP_SUCCESS.getSuccess(), ShopCode.SHOP_SUCCESS.getMessage());
}5.2 支付回调
5.2.1 流程分析
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-WLdsTeg3-1631150365503)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/12.支付后回调.png)]
5.2.2 代码实现
代码语言：javascript复制public Result callbackPayment(TradePay tradePay) {
if (tradePay.getIsPaid().equals(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode())) {
tradePay = tradePayMapper.selectByPrimaryKey(tradePay.getPayId());
if (tradePay == null) {
CastException.cast(ShopCode.SHOP_PAYMENT_NOT_FOUND);
}
tradePay.setIsPaid(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());
int i = tradePayMapper.updateByPrimaryKeySelective(tradePay);
//更新成功代表支付成功
if (i == 1) {
TradeMqProducerTemp mqProducerTemp = new TradeMqProducerTemp();
mqProducerTemp.setId(String.valueOf(idWorker.nextId()));
mqProducerTemp.setGroupName("payProducerGroup");
mqProducerTemp.setMsgKey(String.valueOf(tradePay.getPayId()));
mqProducerTemp.setMsgTag(topic);
mqProducerTemp.setMsgBody(JSON.toJSONString(tradePay));
mqProducerTemp.setCreateTime(new Date());
mqProducerTempMapper.insert(mqProducerTemp);
TradePay finalTradePay = tradePay;
executorService.submit(new Runnable() {
@Override
public void run() {
try {
SendResult sendResult = sendMessage(topic,
tag,
finalTradePay.getPayId(),
JSON.toJSONString(finalTradePay));
log.info(JSON.toJSONString(sendResult));
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
mqProducerTempMapper.deleteByPrimaryKey(mqProducerTemp.getId());
System.out.println("删除消息表成功");
}
} catch (Exception e) {
e.printStackTrace();
}
}
});
} else {
CastException.cast(ShopCode.SHOP_PAYMENT_IS_PAID);
}
}
return new Result(ShopCode.SHOP_SUCCESS.getSuccess(), ShopCode.SHOP_SUCCESS.getMessage());
}线程池优化消息发送逻辑
- 创建线程池对象
@Bean
public ThreadPoolTaskExecutor getThreadPool() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(4);
executor.setMaxPoolSize(8);
executor.setQueueCapacity(100);
executor.setKeepAliveSeconds(60);
executor.setThreadNamePrefix("Pool-A");
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}- 使用线程池
@Autowired
private ThreadPoolTaskExecutor executorService;
executorService.submit(new Runnable() {
@Override
public void run() {
try {
SendResult sendResult = sendMessage(topic, tag, finalTradePay.getPayId(), JSON.toJSONString(finalTradePay));
log.info(JSON.toJSONString(sendResult));
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
mqProducerTempMapper.deleteByPrimaryKey(mqProducerTemp.getId());
System.out.println("删除消息表成功");
}
} catch (Exception e) {
e.printStackTrace();
}
}
});5.2.3
处理消息
支付成功后，支付服务payService发送MQ消息，订单服务、用户服务、日志服务需要订阅消息进行处理
- 订单服务修改订单状态为已支付
- 日志服务记录支付日志
- 用户服务负责给用户增加积分
以下用订单服务为例说明消息的处理情况
1）配置RocketMQ属性值
代码语言：javascript复制mq.pay.topic=payTopic
mq.pay.consumer.group.name=pay_payTopic_group2）消费消息
- 在订单服务中，配置公共的消息处理类
public class BaseConsumer {
public TradeOrder handleMessage(IOrderService
orderService,
MessageExt messageExt,Integer code) throws Exception {
//解析消息内容
String body = new String(messageExt.getBody(), "UTF-8");
String msgId = messageExt.getMsgId();
String tags = messageExt.getTags();
String keys = messageExt.getKeys();
OrderMQ orderMq = JSON.parseObject(body, OrderMQ.class);
//查询
TradeOrder order = orderService.findOne(orderMq.getOrderId());
if(ShopCode.SHOP_ORDER_MESSAGE_STATUS_CANCEL.getCode().equals(code)){
order.setOrderStatus(ShopCode.SHOP_ORDER_CANCEL.getCode());
}
if(ShopCode.SHOP_ORDER_MESSAGE_STATUS_ISPAID.getCode().equals(code)){
order.setPayStatus(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());
}
orderService.changeOrderStatus(order);
return order;
}
}- 接受订单支付成功消息
@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.pay.topic}",
consumerGroup = "${mq.pay.consumer.group.name}")
public class PayConsumer extends BaseConsumer implements RocketMQListener<MessageExt> {
@Autowired
private IOrderService orderService;
@Override
public void onMessage(MessageExt messageExt) {
try {
log.info("CancelOrderProcessor receive message:" messageExt);
TradeOrder order = handleMessage(orderService,
messageExt,
ShopCode.SHOP_ORDER_MESSAGE_STATUS_ISPAID.getCode());
log.info("订单:[" order.getOrderId() "]支付成功");
} catch (Exception e) {
e.printStackTrace();
log.error("订单支付失败");
}
}
}6. 整体联调
通过Rest客户端请求shop-order-web和shop-pay-web完成下单和支付操作
6.1 准备工作
1）配置RestTemplate类
代码语言：javascript复制@Configuration
public class RestTemplateConfig {
@Bean
@ConditionalOnMissingBean({ RestOperations.class, RestTemplate.class })
public RestTemplate restTemplate(ClientHttpRequestFactory factory) {
RestTemplate restTemplate = new RestTemplate(factory);
// 使用 utf-8 编码集的 conver 替换默认的 conver（默认的 string conver 的编码集为"ISO-8859-1"）
List<HttpMessageConverter<?>> messageConverters = restTemplate.getMessageConverters();
Iterator<HttpMessageConverter<?>> iterator = messageConverters.iterator();
while (iterator.hasNext()) {
HttpMessageConverter<?> converter = iterator.next();
if (converter instanceof StringHttpMessageConverter) {
iterator.remove();
}
}
messageConverters.add(new StringHttpMessageConverter(Charset.forName("UTF-8")));
return restTemplate;
}
@Bean
@ConditionalOnMissingBean({ClientHttpRequestFactory.class})
public ClientHttpRequestFactory simpleClientHttpRequestFactory() {
SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
// ms
factory.setReadTimeout(15000);
// ms
factory.setConnectTimeout(15000);
return factory;
}
}2）配置请求地址
- 订单系统
server.host=http://localhost
server.servlet.path=/order-web
server.port=8080
shop.order.baseURI=${server.host}:${server.port}${server.servlet.path}
shop.order.confirm=/order/confirm- 支付系统
server.host=http://localhost
server.servlet.path=/pay-web
server.port=9090
shop.pay.baseURI=${server.host}:${server.port}${server.servlet.path}
shop.pay.createPayment=/pay/createPayment
shop.pay.callbackPayment=/pay/callbackPayment6.2 下单测试
代码语言：javascript复制@RunWith(SpringRunner.class)
@ContextConfiguration(classes = ShopOrderWebApplication.class)
@TestPropertySource("classpath:application.properties")
public class OrderTest {
@Autowired
private RestTemplate restTemplate;
@Value("${shop.order.baseURI}")
private String baseURI;
@Value("${shop.order.confirm}")
private String confirmOrderPath;
@Autowired
private IDWorker idWorker;
/**
* 下单
*/
@Test
public void confirmOrder(){
Long goodsId=XXXL;
Long userId=XXXL;
Long couponId=XXXL;
TradeOrder order = new TradeOrder();
order.setGoodsId(goodsId);
order.setUserId(userId);
order.setGoodsNumber(1);
order.setAddress("北京");
order.setGoodsPrice(new BigDecimal("5000"));
order.setOrderAmount(new BigDecimal("5000"));
order.setMoneyPaid(new BigDecimal("100"));
order.setCouponId(couponId);
order.setShippingFee(new BigDecimal(0));
Result result = restTemplate.postForEntity(baseURI confirmOrderPath, order, Result.class).getBody();
System.out.println(result);
}
}6.3 支付测试
代码语言：javascript复制@RunWith(SpringRunner.class)
@ContextConfiguration(classes = ShopPayWebApplication.class)
@TestPropertySource("classpath:application.properties")
public class PayTest {
@Autowired
private RestTemplate restTemplate;
@Value("${shop.pay.baseURI}")
private String baseURI;
@Value("${shop.pay.createPayment}")
private String createPaymentPath;
@Value("${shop.pay.callbackPayment}")
private String callbackPaymentPath;
@Autowired
private IDWorker idWorker;
/**
* 创建支付订单
*/
@Test
public void createPayment(){
Long orderId = 346321587315814400L;
TradePay pay = new TradePay();
pay.setOrderId(orderId);
pay.setPayAmount(new BigDecimal(4800));
Result result = restTemplate.postForEntity(baseURI createPaymentPath, pay, Result.class).getBody();
System.out.println(result);
}
/**
* 支付回调
*/
@Test
public void callbackPayment(){
Long payId = 346321891507720192L;
TradePay pay = new TradePay();
pay.setPayId(payId);
pay.setIsPaid(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());
Result result = restTemplate.postForEntity(baseURI callbackPaymentPath, pay, Result.class).getBody();
System.out.println(result);
}
}1. 高级功能
1.1 消息存储
分布式队列因为有高可靠性的要求，所以数据要进行持久化存储。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-kdAzkxPz-1631150365503)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/消息存储方式.png)]
- 消息生成者发送消息
- MQ收到消息，将消息进行持久化，在存储中新增一条记录
- 返回ACK给生产者
- MQ push 消息给对应的消费者，然后等待消费者返回ACK
- 如果消息消费者在指定时间内成功返回ack，那么MQ认为消息消费成功，在存储中删除消息，即执行第6步；如果MQ在指定时间内没有收到ACK，则认为消息消费失败，会尝试重新push消息,重复执行4、5、6步骤
- MQ删除消息
1.1.1 存储介质
- 关系型数据库DB
Apache下开源的另外一款MQ—ActiveMQ（默认采用的KahaDB做消息存储）可选用JDBC的方式来做消息持久化，通过简单的xml配置信息即可实现JDBC消息存储。由于，普通关系型数据库（如Mysql）在单表数据量达到千万级别的情况下，其IO读写性能往往会出现瓶颈。在可靠性方面，该种方案非常依赖DB，如果一旦DB出现故障，则MQ的消息就无法落盘存储会导致线上故障
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-9NnnoM3S-1631150365504)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/MySQL.png)]
- 文件系统 目前业界较为常用的几款产品（RocketMQ/Kafka/RabbitMQ）均采用的是消息刷盘至所部署虚拟机/物理机的文件系统来做持久化（刷盘一般可以分为异步刷盘和同步刷盘两种模式）。消息刷盘为消息存储提供了一种高效率、高可靠性和高性能的数据持久化方式。除非部署MQ机器本身或是本地磁盘挂了，否则一般是不会出现无法持久化的故障问题。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-BUuffMHD-1631150365504)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/磁盘.png)]
###1.1.2 性能对比
文件系统>关系型数据库DB
1.1.3 消息的存储和发送
1）消息存储
磁盘如果使用得当，磁盘的速度完全可以匹配上网络 的数据传输速度。目前的高性能磁盘，顺序写速度可以达到600MB/s， 超过了一般网卡的传输速度。但是磁盘随机写的速度只有大概100KB/s，和顺序写的性能相差6000倍！因为有如此巨大的速度差别，好的消息队列系统会比普通的消息队列系统速度快多个数量级。RocketMQ的消息用顺序写,保证了消息存储的速度。
####2）消息发送
Linux操作系统分为【用户态】和【内核态】，文件操作、网络操作需要涉及这两种形态的切换，免不了进行数据复制。
一台服务器 把本机磁盘文件的内容发送到客户端，一般分为两个步骤：
1）read；读取本地文件内容；
2）write；将读取的内容通过网络发送出去。
这两个看似简单的操作，实际进行了4 次数据复制，分别是：
- 从磁盘复制数据到内核态内存；
- 从内核态内存复 制到用户态内存；
- 然后从用户态 内存复制到网络驱动的内核态内存；
- 最后是从网络驱动的内核态内存复 制到网卡中进行传输。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eUoQRMkr-1631150365504)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/文件操作和网络操作.png)]通过使用mmap的方式，可以省去向用户态的内存复制，提高速度。这种机制在Java中是通过MappedByteBuffer实现的
RocketMQ充分利用了上述特性，也就是所谓的“零拷贝”技术，提高消息存盘和网络发送的速度。
这里需要注意的是，采用MappedByteBuffer这种内存映射的方式有几个限制，其中之一是一次只能映射1.5~2G 的文件至用户态的虚拟内存，这也是为何RocketMQ默认设置单个CommitLog日志数据文件为1G的原因了
1.1.4 消息存储结构
RocketMQ消息的存储是由ConsumeQueue和CommitLog配合完成 的，消息真正的物理存储文件是CommitLog，ConsumeQueue是消息的逻辑队列，类似数据库的索引文件，存储的是指向物理存储的地址。每 个Topic下的每个Message Queue都有一个对应的ConsumeQueue文件。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-BUDS0uYo-1631150365505)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/消息存储结构.png)]
- CommitLog：存储消息的元数据
- ConsumerQueue：存储消息在CommitLog的索引
- IndexFile：为了消息查询提供了一种通过key或时间区间来查询消息的方法，这种通过IndexFile来查找消息的方法不影响发送与消费消息的主流程
1.1.5 刷盘机制
RocketMQ的消息是存储到磁盘上的，这样既能保证断电后恢复， 又可以让存储的消息量超出内存的限制。RocketMQ为了提高性能，会尽可能地保证磁盘的顺序写。消息在通过Producer写入RocketMQ的时 候，有两种写磁盘方式，分布式同步刷盘和异步刷盘。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-st0Dbq0Q-1631150365505)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/同步刷盘和异步刷盘.png)]
1）同步刷盘
在返回写成功状态时，消息已经被写入磁盘。具体流程是，消息写入内存的PAGECACHE后，立刻通知刷盘线程刷盘， 然后等待刷盘完成，刷盘线程执行完成后唤醒等待的线程，返回消息写 成功的状态。
2）异步刷盘
在返回写成功状态时，消息可能只是被写入了内存的PAGECACHE，写操作的返回快，吞吐量大；当内存里的消息量积累到一定程度时，统一触发写磁盘动作，快速写入。
####3）配置
同步刷盘还是异步刷盘，都是通过Broker配置文件里的flushDiskType 参数设置的，这个参数被配置成SYNC_FLUSH、ASYNC_FLUSH中的 一个。
1.2 高可用性机制
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VnUDArdC-1631150365505)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/RocketMQ角色.jpg)]
RocketMQ分布式集群是通过Master和Slave的配合达到高可用性的。
Master和Slave的区别：在Broker的配置文件中，参数 brokerId的值为0表明这个Broker是Master，大于0表明这个Broker是 Slave，同时brokerRole参数也会说明这个Broker是Master还是Slave。
Master角色的Broker支持读和写，Slave角色的Broker仅支持读，也就是 Producer只能和Master角色的Broker连接写入消息；Consumer可以连接 Master角色的Broker，也可以连接Slave角色的Broker来读取消息。
1.2.1 消息消费高可用
在Consumer的配置文件中，并不需要设置是从Master读还是从Slave 读，当Master不可用或者繁忙的时候，Consumer会被自动切换到从Slave 读。有了自动切换Consumer这种机制，当一个Master角色的机器出现故障后，Consumer仍然可以从Slave读取消息，不影响Consumer程序。这就达到了消费端的高可用性。
1.2.2 消息发送高可用
在创建Topic的时候，把Topic的多个Message Queue创建在多个Broker组上（相同Broker名称，不同 brokerId的机器组成一个Broker组），这样当一个Broker组的Master不可 用后，其他组的Master仍然可用，Producer仍然可以发送消息。 RocketMQ目前还不支持把Slave自动转成Master，如果机器资源不足， 需要把Slave转成Master，则要手动停止Slave角色的Broker，更改配置文 件，用新的配置文件启动Broker。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rVgyBlhI-1631150365506)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/消息发送高可用设计.jpg)]
1.2.3 消息主从复制
如果一个Broker组有Master和Slave，消息需要从Master复制到Slave 上，有同步和异步两种复制方式。
####1）同步复制
同步复制方式是等Master和Slave均写 成功后才反馈给客户端写成功状态；
在同步复制方式下，如果Master出故障， Slave上有全部的备份数据，容易恢复，但是同步复制会增大数据写入 延迟，降低系统吞吐量。
####2）异步复制
异步复制方式是只要Master写成功 即可反馈给客户端写成功状态。
在异步复制方式下，系统拥有较低的延迟和较高的吞吐量，但是如果Master出了故障，有些数据因为没有被写 入Slave，有可能会丢失；
####3）配置
同步复制和异步复制是通过Broker配置文件里的brokerRole参数进行设置的，这个参数可以被设置成ASYNC_MASTER、 SYNC_MASTER、SLAVE三个值中的一个。
####4）总结
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HVSSmPYd-1631150365506)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/复制刷盘.png)]
实际应用中要结合业务场景，合理设置刷盘方式和主从复制方式， 尤其是SYNC_FLUSH方式，由于频繁地触发磁盘写动作，会明显降低 性能。通常情况下，应该把Master和Save配置成ASYNC_FLUSH的刷盘 方式，主从之间配置成SYNC_MASTER的复制方式，这样即使有一台 机器出故障，仍然能保证数据不丢，是个不错的选择。
1.3 负载均衡
1.3.1 Producer负载均衡
Producer端，每个实例在发消息的时候，默认会轮询所有的message queue发送，以达到让消息平均落在不同的queue上。而由于queue可以散落在不同的broker，所以消息就发送到不同的broker下，如下图：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LsPM6NjU-1631150365506)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/producer负载均衡.png)]
图中箭头线条上的标号代表顺序，发布方会把第一条消息发送至 Queue 0，然后第二条消息发送至 Queue 1，以此类推。
1.3.2 Consumer负载均衡
1）集群模式
在集群消费模式下，每条消息只需要投递到订阅这个topic的Consumer Group下的一个实例即可。RocketMQ采用主动拉取的方式拉取并消费消息，在拉取的时候需要明确指定拉取哪一条message queue。
而每当实例的数量有变更，都会触发一次所有实例的负载均衡，这时候会按照queue的数量和实例的数量平均分配queue给每个实例。
默认的分配算法是AllocateMessageQueueAveragely，如下图：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5LeRf9Dp-1631150365507)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/consumer负载均衡.png)]
还有另外一种平均的算法是AllocateMessageQueueAveragelyByCircle，也是平均分摊每一条queue，只是以环状轮流分queue的形式，如下图：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Hh3zm99R-1631150365507)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/consumer负载均衡2.png)]
需要注意的是，集群模式下，queue都是只允许分配只一个实例，这是由于如果多个实例同时消费一个queue的消息，由于拉取哪些消息是consumer主动控制的，那样会导致同一个消息在不同的实例下被消费多次，所以算法上都是一个queue只分给一个consumer实例，一个consumer实例可以允许同时分到不同的queue。
通过增加consumer实例去分摊queue的消费，可以起到水平扩展的消费能力的作用。而有实例下线的时候，会重新触发负载均衡，这时候原来分配到的queue将分配到其他实例上继续消费。
但是如果consumer实例的数量比message queue的总数量还多的话，多出来的consumer实例将无法分到queue，也就无法消费到消息，也就无法起到分摊负载的作用了。所以需要控制让queue的总数量大于等于consumer的数量。
####2）广播模式
由于广播模式下要求一条消息需要投递到一个消费组下面所有的消费者实例，所以也就没有消息被分摊消费的说法。
在实现上，其中一个不同就是在consumer分配queue的时候，所有consumer都分到所有的queue。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-B1r6r6R9-1631150365507)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/consumer负载均衡3.png)]
1.4 消息重试
1.4.1 顺序消息的重试
对于顺序消息，当消费者消费消息失败后，消息队列 RocketMQ 会自动不断进行消息重试（每次间隔时间为 1 秒），这时，应用会出现消息消费被阻塞的情况。因此，在使用顺序消息时，务必保证应用能够及时监控并处理消费失败的情况，避免阻塞现象的发生。
1.4.2 无序消息的重试
对于无序消息（普通、定时、延时、事务消息），当消费者消费消息失败时，您可以通过设置返回状态达到消息重试的结果。
无序消息的重试只针对集群消费方式生效；广播方式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息。
1）重试次数
消息队列 RocketMQ 默认允许每条消息最多重试 16 次，每次重试的间隔时间如下：
第几次重试 | 与上次重试的间隔时间 | 第几次重试 | 与上次重试的间隔时间 |
|---|---|---|---|
1 | 10 秒 | 9 | 7 分钟 |
2 | 30 秒 | 10 | 8 分钟 |
3 | 1 分钟 | 11 | 9 分钟 |
4 | 2 分钟 | 12 | 10 分钟 |
5 | 3 分钟 | 13 | 20 分钟 |
6 | 4 分钟 | 14 | 30 分钟 |
7 | 5 分钟 | 15 | 1 小时 |
8 | 6 分钟 | 16 | 2 小时 |
如果消息重试 16 次后仍然失败，消息将不再投递。如果严格按照上述重试时间间隔计算，某条消息在一直消费失败的前提下，将会在接下来的 4 小时 46 分钟之内进行 16 次重试，超过这个时间范围消息将不再重试投递。
注意： 一条消息无论重试多少次，这些重试消息的 Message ID 不会改变。
2）配置方式
消费失败后，重试配置方式
集群消费方式下，消息消费失败后期望消息重试，需要在消息监听器接口的实现中明确进行配置（三种方式任选一种）：
- 返回 Action.ReconsumeLater （推荐）
- 返回 Null
- 抛出异常
public class MessageListenerImpl implements MessageListener {
@Override
public Action consume(Message message, ConsumeContext context) {
//处理消息
doConsumeMessage(message);
//方式1：返回 Action.ReconsumeLater，消息将重试
return Action.ReconsumeLater;
//方式2：返回 null，消息将重试
return null;
//方式3：直接抛出异常， 消息将重试
throw new RuntimeException("Consumer Message exceotion");
}
}消费失败后，不重试配置方式
集群消费方式下，消息失败后期望消息不重试，需要捕获消费逻辑中可能抛出的异常，最终返回 Action.CommitMessage，此后这条消息将不会再重试。
代码语言：javascript复制public class MessageListenerImpl implements MessageListener {
@Override
public Action consume(Message message, ConsumeContext context) {
try {
doConsumeMessage(message);
} catch (Throwable e) {
//捕获消费逻辑中的所有异常，并返回 Action.CommitMessage;
return Action.CommitMessage;
}
//消息处理正常，直接返回 Action.CommitMessage;
return Action.CommitMessage;
}
}自定义消息最大重试次数
消息队列 RocketMQ 允许 Consumer 启动的时候设置最大重试次数，重试时间间隔将按照如下策略：
- 最大重试次数小于等于 16 次，则重试时间间隔同上表描述。
- 最大重试次数大于 16 次，超过 16 次的重试时间间隔均为每次 2 小时。
Properties properties = new Properties();
//配置对应 Group ID 的最大消息重试次数为 20 次
properties.put(PropertyKeyConst.MaxReconsumeTimes,"20");
Consumer consumer =ONSFactory.createConsumer(properties);注意：
- 消息最大重试次数的设置对相同 Group ID 下的所有 Consumer 实例有效。
- 如果只对相同 Group ID 下两个 Consumer 实例中的其中一个设置了 MaxReconsumeTimes，那么该配置对两个 Consumer 实例均生效。
- 配置采用覆盖的方式生效，即最后启动的 Consumer 实例会覆盖之前的启动实例的配置
获取消息重试次数
消费者收到消息后，可按照如下方式获取消息的重试次数：
代码语言：javascript复制public class MessageListenerImpl implements MessageListener {
@Override
public Action consume(Message message, ConsumeContext context) {
//获取消息的重试次数
System.out.println(message.getReconsumeTimes());
return Action.CommitMessage;
}
}1.5 死信队列
当一条消息初次消费失败，消息队列 RocketMQ 会自动进行消息重试；达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列 RocketMQ 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。
在消息队列 RocketMQ 中，这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。
1.5.1 死信特性
死信消息具有以下特性
- 不会再被消费者正常消费。
- 有效期与正常消息相同，均为 3 天，3 天后会被自动删除。因此，请在死信消息产生后的 3 天内及时处理。
死信队列具有以下特性：
- 一个死信队列对应一个 Group ID， 而不是对应单个消费者实例。
- 如果一个 Group ID 未产生死信消息，消息队列 RocketMQ 不会为其创建相应的死信队列。
- 一个死信队列包含了对应 Group ID 产生的所有死信消息，不论该消息属于哪个 Topic。
1.5.2 查看死信信息
- 在控制台查询出现死信队列的主题信息
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-iyrleBrQ-1631150365508)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/死信队列主题.png)]
- 在消息界面根据主题查询死信消息
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VXUVzcGu-1631150365508)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/死信队列主题2.png)]
- 选择重新发送消息
一条消息进入死信队列，意味着某些因素导致消费者无法正常消费该消息，因此，通常需要您对其进行特殊处理。排查可疑因素并解决问题后，可以在消息队列 RocketMQ 控制台重新发送该消息，让消费者重新消费一次。
1.6 消费幂等
消息队列 RocketMQ 消费者在接收到消息以后，有必要根据业务上的唯一 Key 对消息做幂等处理的必要性。
1.6.1 消费幂等的必要性
在互联网应用中，尤其在网络不稳定的情况下，消息队列 RocketMQ 的消息有可能会出现重复，这个重复简单可以概括为以下情况：
- 发送时消息重复 当一条消息已被成功发送到服务端并完成持久化，此时出现了网络闪断或者客户端宕机，导致服务端对客户端应答失败。 如果此时生产者意识到消息发送失败并尝试再次发送消息，消费者后续会收到两条内容相同并且 Message ID 也相同的消息。
- 投递时消息重复 消息消费的场景下，消息已投递到消费者并完成业务处理，当客户端给服务端反馈应答的时候网络闪断。 为了保证消息至少被消费一次，消息队列 RocketMQ 的服务端将在网络恢复后再次尝试投递之前已被处理过的消息，消费者后续会收到两条内容相同并且 Message ID 也相同的消息。
- 负载均衡时消息重复（包括但不限于网络抖动、Broker 重启以及订阅方应用重启） 当消息队列 RocketMQ 的 Broker 或客户端重启、扩容或缩容时，会触发 Rebalance，此时消费者可能会收到重复消息。
1.6.2 处理方式
因为 Message ID 有可能出现冲突（重复）的情况，所以真正安全的幂等处理，不建议以 Message ID 作为处理依据。 最好的方式是以业务唯一标识作为幂等处理的关键依据，而业务的唯一标识可以通过消息 Key 进行设置：
代码语言：javascript复制Message message = new Message();
message.setKey("ORDERID_100");
SendResult sendResult = producer.send(message);订阅方收到消息时可以根据消息的 Key 进行幂等处理：
代码语言：javascript复制consumer.subscribe("ons_test", "*", new MessageListener() {
public Action consume(Message message, ConsumeContext context) {
String key = message.getKey()
// 根据业务唯一标识的 key 做幂等处理
}
});2. 源码分析
2.1 环境搭建
依赖工具
- JDK ：1.8
- Maven
- IntelliJ IDEA
2.1.1 源码拉取
从官方仓库 https://github.com/apache/rocketmq clone或者download源码。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-JSQE25IN-1631150365508)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/源码1.png)]
源码目录结构：
- broker: broker 模块（broke 启动进程）
- client ：消息客户端，包含消息生产者、消息消费者相关类
- common ：公共包
- dev ：开发者信息（非源代码）
- distribution ：部署实例文件夹（非源代码）
- example: RocketMQ 例代码
- filter ：消息过滤相关基础类
- filtersrv：消息过滤服务器实现相关类（Filter启动进程）
- logappender：日志实现相关类
- namesrv：NameServer实现相关类（NameServer启动进程）
- openmessageing：消息开放标准
- remoting：远程通信模块，给予Netty
- srcutil：服务工具类
- store：消息存储实现相关类
- style：checkstyle相关实现
- test：测试相关类
- tools：工具类，监控命令相关实现类
###2.1.2 导入IDEA
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-p7hZ9YGQ-1631150365509)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/源码2.png)]
执行安装
代码语言：javascript复制clean install -Dmaven.test.skip=true2.1.3 调试
创建conf配置文件夹,从distribution拷贝broker.conf和logback_broker.xml和logback_namesrv.xml
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7dt7EWqg-1631150365509)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/源码6.png)]
1）启动NameServer
- 展开namesrv模块，右键NamesrvStartup.java
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6klTJLoL-1631150365510)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/源码3.png)]
- 配置ROCKETMQ_HOME
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-upNpYjfh-1631150365510)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/源码4.png)]
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-oAdsmfjS-1631150365510)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/源码5.png)]
- 重新启动 控制台打印结果
The Name Server boot success. serializeType=JSON2）启动Broker
broker.conf配置文件内容
brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
# namesrvAddr地址
namesrvAddr=127.0.0.1:9876
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable=true
# 存储路径
storePathRootDir=E:\RocketMQ\data\rocketmq\dataDir
# commitLog路径
storePathCommitLog=E:\RocketMQ\data\rocketmq\dataDir\commitlog
# 消息队列存储路径
storePathConsumeQueue=E:\RocketMQ\data\rocketmq\dataDir\consumequeue
# 消息索引存储路径
storePathIndex=E:\RocketMQ\data\rocketmq\dataDir\index
# checkpoint文件路径
storeCheckpoint=E:\RocketMQ\data\rocketmq\dataDir\checkpoint
# abort文件存储路径
abortFile=E:\RocketMQ\data\rocketmq\dataDir\abort- 创建数据文件夹
dataDir - 启动
BrokerStartup,配置broker.conf和ROCKETMQ_HOME
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VP9SkUT2-1631150365511)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/源码7.png)]
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VTw6DJGZ-1631150365511)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/源码8.png)]
####3）发送消息
- 进入example模块的
org.apache.rocketmq.example.quickstart - 指定Namesrv地址
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
producer.setNamesrvAddr("127.0.0.1:9876");- 运行
main方法，发送消息
4）消费消息
- 进入example模块的
org.apache.rocketmq.example.quickstart - 指定Namesrv地址
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
consumer.setNamesrvAddr("127.0.0.1:9876");- 运行
main方法，消费消息
2.2 NameServer
2.2.1 架构设计
消息中间件的设计思路一般是基于主题订阅发布的机制，消息生产者（Producer）发送某一个主题到消息服务器，消息服务器负责将消息持久化存储，消息消费者（Consumer）订阅该兴趣的主题，消息服务器根据订阅信息（路由信息）将消息推送到消费者（Push模式）或者消费者主动向消息服务器拉去（Pull模式），从而实现消息生产者与消息消费者解耦。为了避免消息服务器的单点故障导致的整个系统瘫痪，通常会部署多台消息服务器共同承担消息的存储。那消息生产者如何知道消息要发送到哪台消息服务器呢？如果某一台消息服务器宕机了，那么消息生产者如何在不重启服务情况下感知呢？
NameServer就是为了解决以上问题设计的。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-uJeUszxD-1631150365511)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/RocketMQ角色.jpg)]
Broker消息服务器在启动的时向所有NameServer注册，消息生产者（Producer）在发送消息时之前先从NameServer获取Broker服务器地址列表，然后根据负载均衡算法从列表中选择一台服务器进行发送。NameServer与每台Broker保持长连接，并间隔30S检测Broker是否存活，如果检测到Broker宕机，则从路由注册表中删除。但是路由变化不会马上通知消息生产者。这样设计的目的是为了降低NameServer实现的复杂度，在消息发送端提供容错机制保证消息发送的可用性。
NameServer本身的高可用是通过部署多台NameServer来实现，但彼此之间不通讯，也就是NameServer服务器之间在某一个时刻的数据并不完全相同，但这对消息发送并不会造成任何影响，这也是NameServer设计的一个亮点，总之，RocketMQ设计追求简单高效。
2.2.2 启动流程
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-uLTusZO7-1631150365512)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/NameServer启动流程.png)]
启动类：org.apache.rocketmq.namesrv.NamesrvStartup
####步骤一
解析配置文件，填充NameServerConfig、NettyServerConfig属性值，并创建NamesrvController
代码：NamesrvController#createNamesrvController
代码语言：javascript复制//创建NamesrvConfig
final NamesrvConfig namesrvConfig = new NamesrvConfig();
//创建NettyServerConfig
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
//设置启动端口号
nettyServerConfig.setListenPort(9876);
//解析启动-c参数
if (commandLine.hasOption('c')) {
String file = commandLine.getOptionValue('c');
if (file != null) {
InputStream in = new BufferedInputStream(new FileInputStream(file));
properties = new Properties();
properties.load(in);
MixAll.properties2Object(properties, namesrvConfig);
MixAll.properties2Object(properties, nettyServerConfig);
namesrvConfig.setConfigStorePath(file);
System.out.printf("load config properties file OK, %s%n", file);
in.close();
}
}
//解析启动-p参数
if (commandLine.hasOption('p')) {
InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
MixAll.printObjectProperties(console, namesrvConfig);
MixAll.printObjectProperties(console, nettyServerConfig);
System.exit(0);
}
//将启动参数填充到namesrvConfig,nettyServerConfig
MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
//创建NameServerController
final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);NamesrvConfig属性
代码语言：javascript复制private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
private String kvConfigPath = System.getProperty("user.home") File.separator "namesrv" File.separator "kvConfig.json";
private String configStorePath = System.getProperty("user.home") File.separator "namesrv" File.separator "namesrv.properties";
private String productEnvName = "center";
private boolean clusterTest = false;
private boolean orderMessageEnable = false;**rocketmqHome：**rocketmq主目录
**kvConfig：**NameServer存储KV配置属性的持久化路径
**configStorePath：**nameServer默认配置文件路径
**orderMessageEnable：**是否支持顺序消息
NettyServerConfig属性
代码语言：javascript复制private int listenPort = 8888;
private int serverWorkerThreads = 8;
private int serverCallbackExecutorThreads = 0;
private int serverSelectorThreads = 3;
private int serverOnewaySemaphoreValue = 256;
private int serverAsyncSemaphoreValue = 64;
private int serverChannelMaxIdleTimeSeconds = 120;
private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
private boolean serverPooledByteBufAllocatorEnable = true;
private boolean useEpollNativeSelector = false;**listenPort：**NameServer监听端口，该值默认会被初始化为9876 **serverWorkerThreads：**Netty业务线程池线程个数 **serverCallbackExecutorThreads：**Netty public任务线程池线程个数，Netty网络设计，根据业务类型会创建不同的线程池，比如处理消息发送、消息消费、心跳检测等。如果该业务类型未注册线程池，则由public线程池执行。 **serverSelectorThreads：**IO线程池个数，主要是NameServer、Broker端解析请求、返回相应的线程个数，这类线程主要是处理网路请求的，解析请求包，然后转发到各个业务线程池完成具体的操作，然后将结果返回给调用方; **serverOnewaySemaphoreValue：**send oneway消息请求并发读（Broker端参数）; **serverAsyncSemaphoreValue：**异步消息发送最大并发度; **serverChannelMaxIdleTimeSeconds ：**网络连接最大的空闲时间，默认120s。 **serverSocketSndBufSize：**网络socket发送缓冲区大小。 serverSocketRcvBufSize： 网络接收端缓存区大小。 **serverPooledByteBufAllocatorEnable：**ByteBuffer是否开启缓存; **useEpollNativeSelector：**是否启用Epoll IO模型。
步骤二
根据启动属性创建NamesrvController实例，并初始化该实例。NameServerController实例为NameServer核心控制器
代码：NamesrvController#initialize
代码语言：javascript复制public boolean initialize() {
//加载KV配置
this.kvConfigManager.load();
//创建NettyServer网络处理对象
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
//开启定时任务:每隔10s扫描一次Broker,移除不活跃的Broker
this.remotingExecutor =
Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
this.registerProcessor();
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.routeInfoManager.scanNotActiveBroker();
}
}, 5, 10, TimeUnit.SECONDS);
//开启定时任务:每隔10min打印一次KV配置
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.kvConfigManager.printAllPeriodically();
}
}, 1, 10, TimeUnit.MINUTES);
return true;
}步骤三
在JVM进程关闭之前，先将线程池关闭，及时释放资源
代码：NamesrvStartup#start
代码语言：javascript复制//注册JVM钩子函数代码
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
@Override
public Void call() throws Exception {
//释放资源
controller.shutdown();
return null;
}
}));2.2.3 路由管理
NameServer的主要作用是为消息的生产者和消息消费者提供关于主题Topic的路由信息，那么NameServer需要存储路由的基础信息，还要管理Broker节点，包括路由注册、路由删除等。
2.2.3.1 路由元信息
代码：RouteInfoManager
代码语言：javascript复制private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eFKAW3CA-1631150365512)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/路由实体图.png)]
**topicQueueTable：**Topic消息队列路由信息，消息发送时根据路由表进行负载均衡
**brokerAddrTable：**Broker基础信息，包括brokerName、所属集群名称、主备Broker地址
**clusterAddrTable：**Broker集群信息，存储集群中所有Broker名称
**brokerLiveTable：**Broker状态信息，NameServer每次收到心跳包是会替换该信息
**filterServerTable：**Broker上的FilterServer列表，用于类模式消息过滤。
RocketMQ基于定于发布机制，一个Topic拥有多个消息队列，一个Broker为每一个主题创建4个读队列和4个写队列。多个Broker组成一个集群，集群由相同的多台Broker组成Master-Slave架构，brokerId为0代表Master，大于0为Slave。BrokerLiveInfo中的lastUpdateTimestamp存储上次收到Broker心跳包的时间。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-uEZ70ei2-1631150365512)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/实体数据实例.png)]
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xDjQkIq5-1631150365513)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/实体数据实例2.png)]
2.2.3.2 路由注册
#####1）发送心跳包
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FYw2NQP8-1631150365513)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/路由注册.png)]
RocketMQ路由注册是通过Broker与NameServer的心跳功能实现的。Broker启动时向集群中所有的NameServer发送心跳信息，每隔30s向集群中所有NameServer发送心跳包，NameServer收到心跳包时会更新brokerLiveTable缓存中BrokerLiveInfo的lastUpdataTimeStamp信息，然后NameServer每隔10s扫描brokerLiveTable，如果连续120S没有收到心跳包，NameServer将移除Broker的路由信息同时关闭Socket连接。
代码：BrokerController#start
代码语言：javascript复制//注册Broker信息
this.registerBrokerAll(true, false, true);
//每隔30s上报Broker信息到NameServer
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
} catch (Throwable e) {
log.error("registerBrokerAll Exception", e);
}
}
}, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)),
TimeUnit.MILLISECONDS);代码：BrokerOuterAPI#registerBrokerAll
代码语言：javascript复制//获得nameServer地址信息
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
//遍历所有nameserver列表
if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
//封装请求头
final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
requestHeader.setBrokerAddr(brokerAddr);
requestHeader.setBrokerId(brokerId);
requestHeader.setBrokerName(brokerName);
requestHeader.setClusterName(clusterName);
requestHeader.setHaServerAddr(haServerAddr);
requestHeader.setCompressed(compressed);
//封装请求体
RegisterBrokerBody requestBody = new RegisterBrokerBody();
requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
requestBody.setFilterServerList(filterServerList);
final byte[] body = requestBody.encode(compressed);
final int bodyCrc32 = UtilAll.crc32(body);
requestHeader.setBodyCrc32(bodyCrc32);
final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run() {
try {
//分别向NameServer注册
RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
if (result != null) {
registerBrokerResultList.add(result);
}
log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
} catch (Exception e) {
log.warn("registerBroker Exception, {}", namesrvAddr, e);
} finally {
countDownLatch.countDown();
}
}
});
}
try {
countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
}
}代码：BrokerOutAPI#registerBroker
代码语言：javascript复制if (oneway) {
try {
this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
} catch (RemotingTooMuchRequestException e) {
// Ignore
}
return null;
}
RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);2）处理心跳包
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-c1x4cSbW-1631150365513)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/NameServer处理路由注册.png)]
org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor网路处理类解析请求类型，如果请求类型是为***REGISTER_BROKER***，则将请求转发到RouteInfoManager#regiesterBroker
代码：DefaultRequestProcessor#processRequest
代码语言：javascript复制//判断是注册Broker信息
case RequestCode.REGISTER_BROKER:
Version brokerVersion = MQVersion.value2Version(request.getVersion());
if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
return this.registerBrokerWithFilterServer(ctx, request);
} else {
//注册Broker信息
return this.registerBroker(ctx, request);
}代码：DefaultRequestProcessor#registerBroker
代码语言：javascript复制RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
requestHeader.getClusterName(),
requestHeader.getBrokerAddr(),
requestHeader.getBrokerName(),
requestHeader.getBrokerId(),
requestHeader.getHaServerAddr(),
topicConfigWrapper,
null,
ctx.channel()
);代码：RouteInfoManager#registerBroker
维护路由信息
代码语言：javascript复制//加锁
this.lock.writeLock().lockInterruptibly();
//维护clusterAddrTable
Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
if (null == brokerNames) {
brokerNames = new HashSet<String>();
this.clusterAddrTable.put(clusterName, brokerNames);
}
brokerNames.add(brokerName);//维护brokerAddrTable
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
//第一次注册,则创建brokerData
if (null == brokerData) {
registerFirst = true;
brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
this.brokerAddrTable.put(brokerName, brokerData);
}
//非第一次注册,更新Broker
Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();
Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();
while (it.hasNext()) {
Entry<Long, String> item = it.next();
if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {
it.remove();
}
}
String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
registerFirst = registerFirst || (null == oldAddr);//维护topicQueueTable
if (null != topicConfigWrapper && MixAll.MASTER_ID == brokerId) {
if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion()) ||
registerFirst) {
ConcurrentMap<String, TopicConfig> tcTable = topicConfigWrapper.getTopicConfigTable();
if (tcTable != null) {
for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
this.createAndUpdateQueueData(brokerName, entry.getValue());
}
}
}
}代码：RouteInfoManager#createAndUpdateQueueData
代码语言：javascript复制private void createAndUpdateQueueData(final String brokerName, final TopicConfig topicConfig) {
//创建QueueData
QueueData queueData = new QueueData();
queueData.setBrokerName(brokerName);
queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
queueData.setReadQueueNums(topicConfig.getReadQueueNums());
queueData.setPerm(topicConfig.getPerm());
queueData.setTopicSynFlag(topicConfig.getTopicSysFlag());
//获得topicQueueTable中队列集合
List<QueueData> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName());
//topicQueueTable为空,则直接添加queueData到队列集合
if (null == queueDataList) {
queueDataList = new LinkedList<QueueData>();
queueDataList.add(queueData);
this.topicQueueTable.put(topicConfig.getTopicName(), queueDataList);
log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);
} else {
//判断是否是新的队列
boolean addNewOne = true;
Iterator<QueueData> it = queueDataList.iterator();
while (it.hasNext()) {
QueueData qd = it.next();
//如果brokerName相同,代表不是新的队列
if (qd.getBrokerName().equals(brokerName)) {
if (qd.equals(queueData)) {
addNewOne = false;
} else {
log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), qd,
queueData);
it.remove();
}
}
}
//如果是新的队列,则添加队列到queueDataList
if (addNewOne) {
queueDataList.add(queueData);
}
}
}//维护brokerLiveTable
BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,new BrokerLiveInfo(
System.currentTimeMillis(),
topicConfigWrapper.getDataVersion(),
channel,
haServerAddr));//维护filterServerList
if (filterServerList != null) {
if (filterServerList.isEmpty()) {
this.filterServerTable.remove(brokerAddr);
} else {
this.filterServerTable.put(brokerAddr, filterServerList);
}
}
if (MixAll.MASTER_ID != brokerId) {
String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
if (masterAddr != null) {
BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
if (brokerLiveInfo != null) {
result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
result.setMasterAddr(masterAddr);
}
}
}2.2.3.3 路由删除
Broker每隔30s向NameServer发送一个心跳包，心跳包包含BrokerId，Broker地址，Broker名称，Broker所属集群名称、Broker关联的FilterServer列表。但是如果Broker宕机，NameServer无法收到心跳包，此时NameServer如何来剔除这些失效的Broker呢？NameServer会每隔10s扫描brokerLiveTable状态表，如果BrokerLive的lastUpdateTimestamp的时间戳距当前时间超过120s，则认为Broker失效，移除该Broker，关闭与Broker连接，同时更新topicQueueTable、brokerAddrTable、brokerLiveTable、filterServerTable。
RocketMQ有两个触发点来删除路由信息：
- NameServer定期扫描brokerLiveTable检测上次心跳包与当前系统的时间差，如果时间超过120s，则需要移除broker。
- Broker在正常关闭的情况下，会执行unregisterBroker指令
这两种方式路由删除的方法都是一样的，就是从相关路由表中删除与该broker相关的信息。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-snH72sVx-1631150365514)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/路由删除.png)]
代码：NamesrvController#initialize
代码语言：javascript复制//每隔10s扫描一次为活跃Broker
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.routeInfoManager.scanNotActiveBroker();
}
}, 5, 10, TimeUnit.SECONDS);代码：RouteInfoManager#scanNotActiveBroker
代码语言：javascript复制public void scanNotActiveBroker() {
//获得brokerLiveTable
Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
//遍历brokerLiveTable
while (it.hasNext()) {
Entry<String, BrokerLiveInfo> next = it.next();
long last = next.getValue().getLastUpdateTimestamp();
//如果收到心跳包的时间距当时时间是否超过120s
if ((last BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
//关闭连接
RemotingUtil.closeChannel(next.getValue().getChannel());
//移除broker
it.remove();
//维护路由表
this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
}
}
}代码：RouteInfoManager#onChannelDestroy
代码语言：javascript复制//申请写锁,根据brokerAddress从brokerLiveTable和filterServerTable移除
this.lock.writeLock().lockInterruptibly();
this.brokerLiveTable.remove(brokerAddrFound);
this.filterServerTable.remove(brokerAddrFound);//维护brokerAddrTable
String brokerNameFound = null;
boolean removeBrokerName = false;
Iterator<Entry<String, BrokerData>> itBrokerAddrTable =this.brokerAddrTable.entrySet().iterator();
//遍历brokerAddrTable
while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
BrokerData brokerData = itBrokerAddrTable.next().getValue();
//遍历broker地址
Iterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();
while (it.hasNext()) {
Entry<Long, String> entry = it.next();
Long brokerId = entry.getKey();
String brokerAddr = entry.getValue();
//根据broker地址移除brokerAddr
if (brokerAddr.equals(brokerAddrFound)) {
brokerNameFound = brokerData.getBrokerName();
it.remove();
log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
brokerId, brokerAddr);
break;
}
}
//如果当前主题只包含待移除的broker,则移除该topic
if (brokerData.getBrokerAddrs().isEmpty()) {
removeBrokerName = true;
itBrokerAddrTable.remove();
log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
brokerData.getBrokerName());
}
}//维护clusterAddrTable
if (brokerNameFound != null && removeBrokerName) {
Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
//遍历clusterAddrTable
while (it.hasNext()) {
Entry<String, Set<String>> entry = it.next();
//获得集群名称
String clusterName = entry.getKey();
//获得集群中brokerName集合
Set<String> brokerNames = entry.getValue();
//从brokerNames中移除brokerNameFound
boolean removed = brokerNames.remove(brokerNameFound);
if (removed) {
log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
brokerNameFound, clusterName);
if (brokerNames.isEmpty()) {
log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
clusterName);
//如果集群中不包含任何broker,则移除该集群
it.remove();
}
break;
}
}
}//维护topicQueueTable队列
if (removeBrokerName) {
//遍历topicQueueTable
Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =
this.topicQueueTable.entrySet().iterator();
while (itTopicQueueTable.hasNext()) {
Entry<String, List<QueueData>> entry = itTopicQueueTable.next();
//主题名称
String topic = entry.getKey();
//队列集合
List<QueueData> queueDataList = entry.getValue();
//遍历该主题队列
Iterator<QueueData> itQueueData = queueDataList.iterator();
while (itQueueData.hasNext()) {
//从队列中移除为活跃broker信息
QueueData queueData = itQueueData.next();
if (queueData.getBrokerName().equals(brokerNameFound)) {
itQueueData.remove();
log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
topic, queueData);
}
}
//如果该topic的队列为空,则移除该topic
if (queueDataList.isEmpty()) {
itTopicQueueTable.remove();
log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
topic);
}
}
}//释放写锁
finally {
this.lock.writeLock().unlock();
}2.2.3.4 路由发现
RocketMQ路由发现是非实时的，当Topic路由出现变化后，NameServer不会主动推送给客户端，而是由客户端定时拉取主题最新的路由。
代码：DefaultRequestProcessor#getRouteInfoByTopic
代码语言：javascript复制public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
final GetRouteInfoRequestHeader requestHeader =
(GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
//调用RouteInfoManager的方法,从路由表topicQueueTable、brokerAddrTable、filterServerTable中分别填充TopicRouteData的List<QueueData>、List<BrokerData>、filterServer
TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());
//如果找到主题对应你的路由信息并且该主题为顺序消息，则从NameServer KVConfig中获取关于顺序消息相关的配置填充路由信息
if (topicRouteData != null) {
if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
String orderTopicConf =
this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,
requestHeader.getTopic());
topicRouteData.setOrderTopicConf(orderTopicConf);
}
byte[] content = topicRouteData.encode();
response.setBody(content);
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
return response;
}
response.setCode(ResponseCode.TOPIC_NOT_EXIST);
response.setRemark("No topic route info in name server for the topic: " requestHeader.getTopic()
FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
return response;
}2.2.4 小结
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-PWk9zio3-1631150365514)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/NameServer小结.png)]
2.3 Producer
消息生产者的代码都在client模块中，相对于RocketMQ来讲，消息生产者就是客户端，也是消息的提供者。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-itmU75Iy-1631150365514)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/DefaultMQProducer类图.png)]
###2.3.1 方法和属性
####1）主要方法介绍
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-aINzhBAM-1631150365515)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/MQAdmin.png)]
代码语言：javascript复制//创建主题
void createTopic(final String key, final String newTopic, final int queueNum) throws MQClientException;//根据时间戳从队列中查找消息偏移量
long searchOffset(final MessageQueue mq, final long timestamp)//查找消息队列中最大的偏移量
long maxOffset(final MessageQueue mq) throws MQClientException;//查找消息队列中最小的偏移量
long minOffset(final MessageQueue mq) //根据偏移量查找消息
MessageExt viewMessage(final String offsetMsgId) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;//根据条件查找消息
QueryResult queryMessage(final String topic, final String key, final int maxNum, final long begin,
final long end) throws MQClientException, InterruptedException;//根据消息ID和主题查找消息
MessageExt viewMessage(String topic,String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException;[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KQm2uWzF-1631150365515)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/MQProducer.png)]
代码语言：javascript复制//启动
void start() throws MQClientException;//关闭
void shutdown();//查找该主题下所有消息
List<MessageQueue> fetchPublishMessageQueues(final String topic) throws MQClientException;//同步发送消息
SendResult send(final Message msg) throws MQClientException, RemotingException, MQBrokerException,
InterruptedException;//同步超时发送消息
SendResult send(final Message msg, final long timeout) throws MQClientException,
RemotingException, MQBrokerException, InterruptedException;//异步发送消息
void send(final Message msg, final SendCallback sendCallback) throws MQClientException,
RemotingException, InterruptedException;//异步超时发送消息
void send(final Message msg, final SendCallback sendCallback, final long timeout)
throws MQClientException, RemotingException, InterruptedException;//发送单向消息
void sendOneway(final Message msg) throws MQClientException, RemotingException,
InterruptedException;//选择指定队列同步发送消息
SendResult send(final Message msg, final MessageQueue mq) throws MQClientException,
RemotingException, MQBrokerException, InterruptedException;//选择指定队列异步发送消息
void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback)
throws MQClientException, RemotingException, InterruptedException;//选择指定队列单项发送消息
void sendOneway(final Message msg, final MessageQueue mq) throws MQClientException,
RemotingException, InterruptedException;//批量发送消息
SendResult send(final Collection<Message> msgs) throws MQClientException, RemotingException, MQBrokerException,InterruptedException;####2）属性介绍
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bSLhWkIu-1631150365516)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/DefaultMQProducer属性.png)]
代码语言：javascript复制producerGroup：生产者所属组
createTopicKey：默认Topic
defaultTopicQueueNums：默认主题在每一个Broker队列数量
sendMsgTimeout：发送消息默认超时时间，默认3s
compressMsgBodyOverHowmuch：消息体超过该值则启用压缩，默认4k
retryTimesWhenSendFailed：同步方式发送消息重试次数，默认为2，总共执行3次
retryTimesWhenSendAsyncFailed：异步方法发送消息重试次数，默认为2
retryAnotherBrokerWhenNotStoreOK：消息重试时选择另外一个Broker时，是否不等待存储结果就返回，默认为false
maxMessageSize：允许发送的最大消息长度，默认为4M2.3.2 启动流程
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zzkhEUrk-1631150365516)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/生产者启动流程.png)]
代码：DefaultMQProducerImpl#start
代码语言：javascript复制//检查生产者组是否满足要求
this.checkConfig();
//更改当前instanceName为进程ID
if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
this.defaultMQProducer.changeInstanceNameToPID();
}
//获得MQ客户端实例
this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);整个JVM中只存在一个MQClientManager实例，维护一个MQClientInstance缓存表 ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable = new ConcurrentHashMap<String,MQClientInstance>(); 同一个clientId只会创建一个MQClientInstance。 MQClientInstance封装了RocketMQ网络处理API，是消息生产者和消息消费者与NameServer、Broker打交道的网络通道
代码：MQClientManager#getAndCreateMQClientInstance
代码语言：javascript复制public MQClientInstance getAndCreateMQClientInstance(final ClientConfig clientConfig,
RPCHook rpcHook) {
//构建客户端ID
String clientId = clientConfig.buildMQClientId();
//根据客户端ID或者客户端实例
MQClientInstance instance = this.factoryTable.get(clientId);
//实例如果为空就创建新的实例,并添加到实例表中
if (null == instance) {
instance =
new MQClientInstance(clientConfig.cloneClientConfig(),
this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
if (prev != null) {
instance = prev;
log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
} else {
log.info("Created new MQClientInstance for clientId:[{}]", clientId);
}
}
return instance;
}代码：DefaultMQProducerImpl#start
代码语言：javascript复制//注册当前生产者到到MQClientInstance管理中,方便后续调用网路请求
boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
if (!registerOK) {
this.serviceState = ServiceState.CREATE_JUST;
throw new MQClientException("The producer group[" this.defaultMQProducer.getProducerGroup()
"] has been created before, specify another name please." FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
null);
}
//启动生产者
if (startFactory) {
mQClientFactory.start();
}2.3.3 消息发送
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qTL3Rtzr-1631150365517)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/消息发送.png)]
代码：DefaultMQProducerImpl#send(Message msg)
代码语言：javascript复制//发送消息
public SendResult send(Message msg) {
return send(msg, this.defaultMQProducer.getSendMsgTimeout());
}代码：DefaultMQProducerImpl#send(Message msg,long timeout)
代码语言：javascript复制//发送消息,默认超时时间为3s
public SendResult send(Message msg,long timeout){
return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
}代码：DefaultMQProducerImpl#sendDefaultImpl
代码语言：javascript复制//校验消息
Validators.checkMessage(msg, this.defaultMQProducer);####1）验证消息
代码：Validators#checkMessage
代码语言：javascript复制public static void checkMessage(Message msg, DefaultMQProducer defaultMQProducer)
throws MQClientException {
//判断是否为空
if (null == msg) {
throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message is null");
}
// 校验主题
Validators.checkTopic(msg.getTopic());
// 校验消息体
if (null == msg.getBody()) {
throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body is null");
}
if (0 == msg.getBody().length) {
throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body length is zero");
}
if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) {
throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL,
"the message body size over max value, MAX: " defaultMQProducer.getMaxMessageSize());
}
}####2）查找路由
代码：DefaultMQProducerImpl#tryToFindTopicPublishInfo
代码语言：javascript复制private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
//从缓存中获得主题的路由信息
TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
//路由信息为空,则从NameServer获取路由
if (null == topicPublishInfo || !topicPublishInfo.ok()) {
this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
topicPublishInfo = this.topicPublishInfoTable.get(topic);
}
if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
return topicPublishInfo;
} else {
//如果未找到当前主题的路由信息,则用默认主题继续查找
this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
topicPublishInfo = this.topicPublishInfoTable.get(topic);
return topicPublishInfo;
}
}[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KFbb8pD5-1631150365517)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/Topic路由信息.png)]
代码：TopicPublishInfo
代码语言：javascript复制public class TopicPublishInfo {
private boolean orderTopic = false; //是否是顺序消息
private boolean haveTopicRouterInfo = false;
private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>(); //该主题消息队列
private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();//每选择一次消息队列,该值 1
private TopicRouteData topicRouteData;//关联Topic路由元信息
}代码：MQClientInstance#updateTopicRouteInfoFromNameServer
代码语言：javascript复制TopicRouteData topicRouteData;
//使用默认主题从NameServer获取路由信息
if (isDefault && defaultMQProducer != null) {
topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),
1000 * 3);
if (topicRouteData != null) {
for (QueueData data : topicRouteData.getQueueDatas()) {
int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
data.setReadQueueNums(queueNums);
data.setWriteQueueNums(queueNums);
}
}
} else {
//使用指定主题从NameServer获取路由信息
topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
}代码：MQClientInstance#updateTopicRouteInfoFromNameServer
代码语言：javascript复制//判断路由是否需要更改
TopicRouteData old = this.topicRouteTable.get(topic);
boolean changed = topicRouteDataIsChange(old, topicRouteData);
if (!changed) {
changed = this.isNeedUpdateTopicRouteInfo(topic);
} else {
log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
}代码：MQClientInstance#updateTopicRouteInfoFromNameServer
代码语言：javascript复制if (changed) {
//将topicRouteData转换为发布队列
TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
publishInfo.setHaveTopicRouterInfo(true);
//遍历生产
Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, MQProducerInner> entry = it.next();
MQProducerInner impl = entry.getValue();
if (impl != null) {
//生产者不为空时,更新publishInfo信息
impl.updateTopicPublishInfo(topic, publishInfo);
}
}
}代码：MQClientInstance#topicRouteData2TopicPublishInfo
代码语言：javascript复制public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) {
//创建TopicPublishInfo对象
TopicPublishInfo info = new TopicPublishInfo();
//关联topicRoute
info.setTopicRouteData(route);
//顺序消息,更新TopicPublishInfo
if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) {
String[] brokers = route.getOrderTopicConf().split(";");
for (String broker : brokers) {
String[] item = broker.split(":");
int nums = Integer.parseInt(item[1]);
for (int i = 0; i < nums; i ) {
MessageQueue mq = new MessageQueue(topic, item[0], i);
info.getMessageQueueList().add(mq);
}
}
info.setOrderTopic(true);
} else {
//非顺序消息更新TopicPublishInfo
List<QueueData> qds = route.getQueueDatas();
Collections.sort(qds);
//遍历topic队列信息
for (QueueData qd : qds) {
//是否是写队列
if (PermName.isWriteable(qd.getPerm())) {
BrokerData brokerData = null;
//遍历写队列Broker
for (BrokerData bd : route.getBrokerDatas()) {
//根据名称获得读队列对应的Broker
if (bd.getBrokerName().equals(qd.getBrokerName())) {
brokerData = bd;
break;
}
}
if (null == brokerData) {
continue;
}
if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) {
continue;
}
//封装TopicPublishInfo写队列
for (int i = 0; i < qd.getWriteQueueNums(); i ) {
MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i);
info.getMessageQueueList().add(mq);
}
}
}
info.setOrderTopic(false);
}
//返回TopicPublishInfo对象
return info;
}3）选择队列
- 默认不启用Broker故障延迟机制
代码：TopicPublishInfo#selectOneMessageQueue(lastBrokerName)
代码语言：javascript复制public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
//第一次选择队列
if (lastBrokerName == null) {
return selectOneMessageQueue();
} else {
//sendWhichQueue
int index = this.sendWhichQueue.getAndIncrement();
//遍历消息队列集合
for (int i = 0; i < this.messageQueueList.size(); i ) {
//sendWhichQueue自增后取模
int pos = Math.abs(index ) % this.messageQueueList.size();
if (pos < 0)
pos = 0;
//规避上次Broker队列
MessageQueue mq = this.messageQueueList.get(pos);
if (!mq.getBrokerName().equals(lastBrokerName)) {
return mq;
}
}
//如果以上情况都不满足,返回sendWhichQueue取模后的队列
return selectOneMessageQueue();
}
}代码：TopicPublishInfo#selectOneMessageQueue()
代码语言：javascript复制//第一次选择队列
public MessageQueue selectOneMessageQueue() {
//sendWhichQueue自增
int index = this.sendWhichQueue.getAndIncrement();
//对队列大小取模
int pos = Math.abs(index) % this.messageQueueList.size();
if (pos < 0)
pos = 0;
//返回对应的队列
return this.messageQueueList.get(pos);
}- 启用Broker故障延迟机制
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
//Broker故障延迟机制
if (this.sendLatencyFaultEnable) {
try {
//对sendWhichQueue自增
int index = tpInfo.getSendWhichQueue().getAndIncrement();
//对消息队列轮询获取一个队列
for (int i = 0; i < tpInfo.getMessageQueueList().size(); i ) {
int pos = Math.abs(index ) % tpInfo.getMessageQueueList().size();
if (pos < 0)
pos = 0;
MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
//验证该队列是否可用
if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
//可用
if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
return mq;
}
}
//从规避的Broker中选择一个可用的Broker
final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
//获得Broker的写队列集合
int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
if (writeQueueNums > 0) {
//获得一个队列,指定broker和队列ID并返回
final MessageQueue mq = tpInfo.selectOneMessageQueue();
if (notBestBroker != null) {
mq.setBrokerName(notBestBroker);
mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
}
return mq;
} else {
latencyFaultTolerance.remove(notBestBroker);
}
} catch (Exception e) {
log.error("Error occurred when selecting message queue", e);
}
return tpInfo.selectOneMessageQueue();
}
return tpInfo.selectOneMessageQueue(lastBrokerName);
}[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-tnadQoZh-1631150365517)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/Broker故障延迟机制核心类.png)]
- 延迟机制接口规范
public interface LatencyFaultTolerance<T> {
//更新失败条目
void updateFaultItem(final T name, final long currentLatency, final long notAvailableDuration);
//判断Broker是否可用
boolean isAvailable(final T name);
//移除Fault条目
void remove(final T name);
//尝试从规避的Broker中选择一个可用的Broker
T pickOneAtLeast();
}- FaultItem：失败条目
class FaultItem implements Comparable<FaultItem> {
//条目唯一键,这里为brokerName
private final String name;
//本次消息发送延迟
private volatile long currentLatency;
//故障规避开始时间
private volatile long startTimestamp;
}- 消息失败策略
public class MQFaultStrategy {
//根据currentLatency本地消息发送延迟,从latencyMax尾部向前找到第一个比currentLatency小的索引,如果没有找到,返回0
private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
//根据这个索引从notAvailableDuration取出对应的时间,在该时长内,Broker设置为不可用
private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};
}原理分析
代码：DefaultMQProducerImpl#sendDefaultImpl
代码语言：javascript复制sendResult = this.sendKernelImpl(msg,
mq,
communicationMode,
sendCallback,
topicPublishInfo,
timeout - costTime);
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);如果上述发送过程出现异常，则调用DefaultMQProducerImpl#updateFaultItem
public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
//参数一：broker名称
//参数二:本次消息发送延迟时间
//参数三:是否隔离
this.mqFaultStrategy.updateFaultItem(brokerName, currentLatency, isolation);
}代码：MQFaultStrategy#updateFaultItem
代码语言：javascript复制public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
if (this.sendLatencyFaultEnable) {
//计算broker规避的时长
long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
//更新该FaultItem规避时长
this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
}
}代码：MQFaultStrategy#computeNotAvailableDuration
代码语言：javascript复制private long computeNotAvailableDuration(final long currentLatency) {
//遍历latencyMax
for (int i = latencyMax.length - 1; i >= 0; i--) {
//找到第一个比currentLatency的latencyMax值
if (currentLatency >= latencyMax[i])
return this.notAvailableDuration[i];
}
//没有找到则返回0
return 0;
}代码：LatencyFaultToleranceImpl#updateFaultItem
代码语言：javascript复制public void updateFaultItem(final String name, final long currentLatency, final long notAvailableDuration) {
//获得原FaultItem
FaultItem old = this.faultItemTable.get(name);
//为空新建faultItem对象,设置规避时长和开始时间
if (null == old) {
final FaultItem faultItem = new FaultItem(name);
faultItem.setCurrentLatency(currentLatency);
faultItem.setStartTimestamp(System.currentTimeMillis() notAvailableDuration);
old = this.faultItemTable.putIfAbsent(name, faultItem);
if (old != null) {
old.setCurrentLatency(currentLatency);
old.setStartTimestamp(System.currentTimeMillis() notAvailableDuration);
}
} else {
//更新规避时长和开始时间
old.setCurrentLatency(currentLatency);
old.setStartTimestamp(System.currentTimeMillis() notAvailableDuration);
}
}####4）发送消息
消息发送API核心入口***DefaultMQProducerImpl#sendKernelImpl***
代码语言：javascript复制private SendResult sendKernelImpl(
final Message msg, //待发送消息
final MessageQueue mq, //消息发送队列
final CommunicationMode communicationMode, //消息发送内模式
final SendCallback sendCallback, pp //异步消息回调函数
final TopicPublishInfo topicPublishInfo, //主题路由信息
final long timeout //超时时间
)代码：DefaultMQProducerImpl#sendKernelImpl
代码语言：javascript复制//获得broker网络地址信息
String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
if (null == brokerAddr) {
//没有找到从NameServer更新broker网络地址信息
tryToFindTopicPublishInfo(mq.getTopic());
brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
}//为消息分类唯一ID
if (!(msg instanceof MessageBatch)) {
MessageClientIDSetter.setUniqID(msg);
}
boolean topicWithNamespace = false;
if (null != this.mQClientFactory.getClientConfig().getNamespace()) {
msg.setInstanceId(this.mQClientFactory.getClientConfig().getNamespace());
topicWithNamespace = true;
}
//消息大小超过4K,启用消息压缩
int sysFlag = 0;
boolean msgBodyCompressed = false;
if (this.tryToCompressMessage(msg)) {
sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
msgBodyCompressed = true;
}
//如果是事务消息,设置消息标记MessageSysFlag.TRANSACTION_PREPARED_TYPE
final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
}//如果注册了消息发送钩子函数,在执行消息发送前的增强逻辑
if (this.hasSendMessageHook()) {
context = new SendMessageContext();
context.setProducer(this);
context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
context.setCommunicationMode(communicationMode);
context.setBornHost(this.defaultMQProducer.getClientIP());
context.setBrokerAddr(brokerAddr);
context.setMessage(msg);
context.setMq(mq);
context.setNamespace(this.defaultMQProducer.getNamespace());
String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (isTrans != null && isTrans.equals("true")) {
context.setMsgType(MessageType.Trans_Msg_Half);
}
if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
context.setMsgType(MessageType.Delay_Msg);
}
this.executeSendMessageHookBefore(context);
}代码：SendMessageHook
代码语言：javascript复制public interface SendMessageHook {
String hookName();
void sendMessageBefore(final SendMessageContext context);
void sendMessageAfter(final SendMessageContext context);
}代码：DefaultMQProducerImpl#sendKernelImpl
代码语言：javascript复制//构建消息发送请求包
SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
//生产者组
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
//主题
requestHeader.setTopic(msg.getTopic());
//默认创建主题Key
requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
//该主题在单个Broker默认队列树
requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
//队列ID
requestHeader.setQueueId(mq.getQueueId());
//消息系统标记
requestHeader.setSysFlag(sysFlag);
//消息发送时间
requestHeader.setBornTimestamp(System.currentTimeMillis());
//消息标记
requestHeader.setFlag(msg.getFlag());
//消息扩展信息
requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
//消息重试次数
requestHeader.setReconsumeTimes(0);
requestHeader.setUnitMode(this.isUnitMode());
//是否是批量消息等
requestHeader.setBatch(msg instanceof MessageBatch);
if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
if (reconsumeTimes != null) {
requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
}
String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
if (maxReconsumeTimes != null) {
requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
}
}case ASYNC: //异步发送
Message tmpMessage = msg;
boolean messageCloned = false;
if (msgBodyCompressed) {
//If msg body was compressed, msgbody should be reset using prevBody.
//Clone new message using commpressed message body and recover origin massage.
//Fix bug:https://github.com/apache/rocketmq-externals/issues/66
tmpMessage = MessageAccessor.cloneMessage(msg);
messageCloned = true;
msg.setBody(prevBody);
}
if (topicWithNamespace) {
if (!messageCloned) {
tmpMessage = MessageAccessor.cloneMessage(msg);
messageCloned = true;
}
msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(),
this.defaultMQProducer.getNamespace()));
}
long costTimeAsync = System.currentTimeMillis() - beginStartTime;
if (timeout < costTimeAsync) {
throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
}
sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
mq.getBrokerName(),
tmpMessage,
requestHeader,
timeout - costTimeAsync,
communicationMode,
sendCallback,
topicPublishInfo,
this.mQClientFactory,
this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),
context,
this);
break;
case ONEWAY:
case SYNC: //同步发送
long costTimeSync = System.currentTimeMillis() - beginStartTime;
if (timeout < costTimeSync) {
throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
}
sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
mq.getBrokerName(),
msg,
requestHeader,
timeout - costTimeSync,
communicationMode,
context,
this);
break;
default:
assert false;
break;
}//如果注册了钩子函数,则发送完毕后执行钩子函数
if (this.hasSendMessageHook()) {
context.setSendResult(sendResult);
this.executeSendMessageHookAfter(context);
}2.3.4 批量消息发送
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-sf0mQ6mI-1631150365518)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/发送批量消息.png)]
批量消息发送是将同一个主题的多条消息一起打包发送到消息服务端，减少网络调用次数，提高网络传输效率。当然，并不是在同一批次中发送的消息数量越多越好，其判断依据是单条消息的长度，如果单条消息内容比较长，则打包多条消息发送会影响其他线程发送消息的响应时间，并且单批次消息总长度不能超过DefaultMQProducer#maxMessageSize。
批量消息发送要解决的问题是如何将这些消息编码以便服务端能够正确解码出每条消息的消息内容。
代码：DefaultMQProducer#send
代码语言：javascript复制public SendResult send(Collection<Message> msgs)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
//压缩消息集合成一条消息,然后发送出去
return this.defaultMQProducerImpl.send(batch(msgs));
}代码：DefaultMQProducer#batch
代码语言：javascript复制private MessageBatch batch(Collection<Message> msgs) throws MQClientException {
MessageBatch msgBatch;
try {
//将集合消息封装到MessageBatch
msgBatch = MessageBatch.generateFromList(msgs);
//遍历消息集合,检查消息合法性,设置消息ID,设置Topic
for (Message message : msgBatch) {
Validators.checkMessage(message, this);
MessageClientIDSetter.setUniqID(message);
message.setTopic(withNamespace(message.getTopic()));
}
//压缩消息,设置消息body
msgBatch.setBody(msgBatch.encode());
} catch (Exception e) {
throw new MQClientException("Failed to initiate the MessageBatch", e);
}
//设置msgBatch的topic
msgBatch.setTopic(withNamespace(msgBatch.getTopic()));
return msgBatch;
}2.4 消息存储
###2.4.1 消息存储核心类
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mrPl4TzJ-1631150365518)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/DefaultMessageStore.png)]
代码语言：javascript复制private final MessageStoreConfig messageStoreConfig; //消息配置属性
private final CommitLog commitLog; //CommitLog文件存储的实现类
private final ConcurrentMap<String/* topic */, ConcurrentMap<Integer/* queueId */, ConsumeQueue>> consumeQueueTable; //消息队列存储缓存表,按照消息主题分组
private final FlushConsumeQueueService flushConsumeQueueService; //消息队列文件刷盘线程
private final CleanCommitLogService cleanCommitLogService; //清除CommitLog文件服务
private final CleanConsumeQueueService cleanConsumeQueueService; //清除ConsumerQueue队列文件服务
private final IndexService indexService; //索引实现类
private final AllocateMappedFileService allocateMappedFileService; //MappedFile分配服务
private final ReputMessageService reputMessageService;//CommitLog消息分发,根据CommitLog文件构建ConsumerQueue、IndexFile文件
private final HAService haService; //存储HA机制
private final ScheduleMessageService scheduleMessageService; //消息服务调度线程
private final StoreStatsService storeStatsService; //消息存储服务
private final TransientStorePool transientStorePool; //消息堆外内存缓存
private final BrokerStatsManager brokerStatsManager; //Broker状态管理器
private final MessageArrivingListener messageArrivingListener; //消息拉取长轮询模式消息达到监听器
private final BrokerConfig brokerConfig; //Broker配置类
private StoreCheckpoint storeCheckpoint; //文件刷盘监测点
private final LinkedList<CommitLogDispatcher> dispatcherList; //CommitLog文件转发请求2.4.2 消息存储流程
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Pjwm9GWU-1631150365519)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/消息存储流程.png)]
消息存储入口：DefaultMessageStore#putMessage
代码语言：javascript复制//判断Broker角色如果是从节点,则无需写入
if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
long value = this.printTimes.getAndIncrement();
if ((value % 50000) == 0) {
log.warn("message store is slave mode, so putMessage is forbidden ");
}
return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
}
//判断当前写入状态如果是正在写入,则不能继续
if (!this.runningFlags.isWriteable()) {
long value = this.printTimes.getAndIncrement();
return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
} else {
this.printTimes.set(0);
}
//判断消息主题长度是否超过最大限制
if (msg.getTopic().length() > Byte.MAX_VALUE) {
log.warn("putMessage message topic length too long " msg.getTopic().length());
return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
}
//判断消息属性长度是否超过限制
if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) {
log.warn("putMessage message properties length too long " msg.getPropertiesString().length());
return new PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, null);
}
//判断系统PageCache缓存去是否占用
if (this.isOSPageCacheBusy()) {
return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, null);
}
//将消息写入CommitLog文件
PutMessageResult result = this.commitLog.putMessage(msg);代码：CommitLog#putMessage
代码语言：javascript复制//记录消息存储时间
msg.setStoreTimestamp(beginLockTimestamp);
//判断如果mappedFile如果为空或者已满,创建新的mappedFile文件
if (null == mappedFile || mappedFile.isFull()) {
mappedFile = this.mappedFileQueue.getLastMappedFile(0);
}
//如果创建失败,直接返回
if (null == mappedFile) {
log.error("create mapped file1 error, topic: " msg.getTopic() " clientAddr: " msg.getBornHostString());
beginTimeInLock = 0;
return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
}
//写入消息到mappedFile中
result = mappedFile.appendMessage(msg, this.appendMessageCallback);代码：MappedFile#appendMessagesInner
代码语言：javascript复制//获得文件的写入指针
int currentPos = this.wrotePosition.get();
//如果指针大于文件大小则直接返回
if (currentPos < this.fileSize) {
//通过writeBuffer.slice()创建一个与MappedFile共享的内存区,并设置position为当前指针
ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
byteBuffer.position(currentPos);
AppendMessageResult result = null;
if (messageExt instanceof MessageExtBrokerInner) {
//通过回调方法写入
result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt);
} else if (messageExt instanceof MessageExtBatch) {
result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt);
} else {
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}
this.wrotePosition.addAndGet(result.getWroteBytes());
this.storeTimestamp = result.getStoreTimestamp();
return result;
}代码：CommitLog#doAppend
代码语言：javascript复制//文件写入位置
long wroteOffset = fileFromOffset byteBuffer.position();
//设置消息ID
this.resetByteBuffer(hostHolder, 8);
String msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(hostHolder), wroteOffset);
//获得该消息在消息队列中的偏移量
keyBuilder.setLength(0);
keyBuilder.append(msgInner.getTopic());
keyBuilder.append('-');
keyBuilder.append(msgInner.getQueueId());
String key = keyBuilder.toString();
Long queueOffset = CommitLog.this.topicQueueTable.get(key);
if (null == queueOffset) {
queueOffset = 0L;
CommitLog.this.topicQueueTable.put(key, queueOffset);
}
//获得消息属性长度
final byte[] propertiesData =msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
final int propertiesLength = propertiesData == null ? 0 : propertiesData.length;
if (propertiesLength > Short.MAX_VALUE) {
log.warn("putMessage message properties length too long. length={}", propertiesData.length);
return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
}
//获得消息主题大小
final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
final int topicLength = topicData.length;
//获得消息体大小
final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
//计算消息总长度
final int msgLen = calMsgLength(bodyLength, topicLength, propertiesLength);代码：CommitLog#calMsgLength
代码语言：javascript复制protected static int calMsgLength(int bodyLength, int topicLength, int propertiesLength) {
final int msgLen = 4 //TOTALSIZE
4 //MAGICCODE
4 //BODYCRC
4 //QUEUEID
4 //FLAG
8 //QUEUEOFFSET
8 //PHYSICALOFFSET
4 //SYSFLAG
8 //BORNTIMESTAMP
8 //BORNHOST
8 //STORETIMESTAMP
8 //STOREHOSTADDRESS
4 //RECONSUMETIMES
8 //Prepared Transaction Offset
4 (bodyLength > 0 ? bodyLength : 0) //BODY
1 topicLength //TOPIC
2 (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength
0;
return msgLen;
}代码：CommitLog#doAppend
代码语言：javascript复制//消息长度不能超过4M
if (msgLen > this.maxMessageSize) {
CommitLog.log.warn("message size exceeded, msg total size: " msgLen ", msg body size: " bodyLength
", maxMessageSize: " this.maxMessageSize);
return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
}
//消息是如果没有足够的存储空间则新创建CommitLog文件
if ((msgLen END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
this.resetByteBuffer(this.msgStoreItemMemory, maxBlank);
// 1 TOTALSIZE
this.msgStoreItemMemory.putInt(maxBlank);
// 2 MAGICCODE
this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
// 3 The remaining space may be any value
// Here the length of the specially set maxBlank
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank);
return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(),
queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
}
//将消息存储到ByteBuffer中,返回AppendMessageResult
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
// Write messages to the queue buffer
byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);
AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset,
msgLen, msgId,msgInner.getStoreTimestamp(),
queueOffset,
CommitLog.this.defaultMessageStore.now()
-beginTimeMills);
switch (tranType) {
case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
break;
case MessageSysFlag.TRANSACTION_NOT_TYPE:
case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
//更新消息队列偏移量
CommitLog.this.topicQueueTable.put(key, queueOffset);
break;
default:
break;
}代码：CommitLog#putMessage
代码语言：javascript复制//释放锁
putMessageLock.unlock();
//刷盘
handleDiskFlush(result, putMessageResult, msg);
//执行HA主从同步
handleHA(result, putMessageResult, msg);2.4.3 存储文件
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fceNVEli-1631150365519)(…/文档/ https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/存储文件.png)]
- commitLog：消息存储目录
- config：运行期间一些配置信息
- consumerqueue：消息消费队列存储目录
- index：消息索引文件存储目录
- abort：如果存在改文件寿命Broker非正常关闭
- checkpoint：文件检查点，存储CommitLog文件最后一次刷盘时间戳、consumerquueue最后一次刷盘时间，index索引文件最后一次刷盘时间戳。
2.4.4 存储文件内存映射
RocketMQ通过使用内存映射文件提高IO访问性能，无论是CommitLog、ConsumerQueue还是IndexFile，单个文件都被设计为固定长度，如果一个文件写满以后再创建一个新文件，文件名就为该文件第一条消息对应的全局物理偏移量。
####1）MappedFileQueue
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LFvqHA2n-1631150365520)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/MappedFileQueue.png)]
代码语言：javascript复制String storePath; //存储目录
int mappedFileSize; // 单个文件大小
CopyOnWriteArrayList<MappedFile> mappedFiles; //MappedFile文件集合
AllocateMappedFileService allocateMappedFileService; //创建MapFile服务类
long flushedWhere = 0; //当前刷盘指针
long committedWhere = 0; //当前数据提交指针,内存中ByteBuffer当前的写指针,该值大于等于flushWhere- 根据存储时间查询MappedFile
public MappedFile getMappedFileByTime(final long timestamp) {
Object[] mfs = this.copyMappedFiles(0);
if (null == mfs)
return null;
//遍历MappedFile文件数组
for (int i = 0; i < mfs.length; i ) {
MappedFile mappedFile = (MappedFile) mfs[i];
//MappedFile文件的最后修改时间大于指定时间戳则返回该文件
if (mappedFile.getLastModifiedTimestamp() >= timestamp) {
return mappedFile;
}
}
return (MappedFile) mfs[mfs.length - 1];
}- 根据消息偏移量offset查找MappedFile
public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
try {
//获得第一个MappedFile文件
MappedFile firstMappedFile = this.getFirstMappedFile();
//获得最后一个MappedFile文件
MappedFile lastMappedFile = this.getLastMappedFile();
//第一个文件和最后一个文件均不为空,则进行处理
if (firstMappedFile != null && lastMappedFile != null) {
if (offset < firstMappedFile.getFileFromOffset() ||
offset >= lastMappedFile.getFileFromOffset() this.mappedFileSize) {
} else {
//获得文件索引
int index = (int) ((offset / this.mappedFileSize)
- (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
MappedFile targetFile = null;
try {
//根据索引返回目标文件
targetFile = this.mappedFiles.get(index);
} catch (Exception ignored) {
}
if (targetFile != null && offset >= targetFile.getFileFromOffset()
&& offset < targetFile.getFileFromOffset() this.mappedFileSize) {
return targetFile;
}
for (MappedFile tmpMappedFile : this.mappedFiles) {
if (offset >= tmpMappedFile.getFileFromOffset()
&& offset < tmpMappedFile.getFileFromOffset() this.mappedFileSize) {
return tmpMappedFile;
}
}
}
if (returnFirstOnNotFound) {
return firstMappedFile;
}
}
} catch (Exception e) {
log.error("findMappedFileByOffset Exception", e);
}
return null;
}- 获取存储文件最小偏移量
public long getMinOffset() {
if (!this.mappedFiles.isEmpty()) {
try {
return this.mappedFiles.get(0).getFileFromOffset();
} catch (IndexOutOfBoundsException e) {
//continue;
} catch (Exception e) {
log.error("getMinOffset has exception.", e);
}
}
return -1;
}- 获取存储文件最大偏移量
public long getMaxOffset() {
MappedFile mappedFile = getLastMappedFile();
if (mappedFile != null) {
return mappedFile.getFileFromOffset() mappedFile.getReadPosition();
}
return 0;
}- 返回存储文件当前写指针
public long getMaxWrotePosition() {
MappedFile mappedFile = getLastMappedFile();
if (mappedFile != null) {
return mappedFile.getFileFromOffset() mappedFile.getWrotePosition();
}
return 0;
}####2）MappedFile
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KecCgDH7-1631150365520)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/MappedFile.png)]
代码语言：javascript复制int OS_PAGE_SIZE = 1024 * 4; //操作系统每页大小,默认4K
AtomicLong TOTAL_MAPPED_VIRTUAL_MEMORY = new AtomicLong(0); //当前JVM实例中MappedFile虚拟内存
AtomicInteger TOTAL_MAPPED_FILES = new AtomicInteger(0); //当前JVM实例中MappedFile对象个数
AtomicInteger wrotePosition = new AtomicInteger(0); //当前文件的写指针
AtomicInteger committedPosition = new AtomicInteger(0); //当前文件的提交指针
AtomicInteger flushedPosition = new AtomicInteger(0); //刷写到磁盘指针
int fileSize; //文件大小
FileChannel fileChannel; //文件通道
ByteBuffer writeBuffer = null; //堆外内存ByteBuffer
TransientStorePool transientStorePool = null; //堆外内存池
String fileName; //文件名称
long fileFromOffset; //该文件的处理偏移量
File file; //物理文件
MappedByteBuffer mappedByteBuffer; //物理文件对应的内存映射Buffer
volatile long storeTimestamp = 0; //文件最后一次内容写入时间
boolean firstCreateInQueue = false; //是否是MappedFileQueue队列中第一个文件MappedFile初始化
- 未开启
transientStorePoolEnable。transientStorePoolEnable=true为true表示数据先存储到堆外内存，然后通过Commit线程将数据提交到内存映射Buffer中，再通过Flush线程将内存映射Buffer中数据持久化磁盘。
private void init(final String fileName, final int fileSize) throws IOException {
this.fileName = fileName;
this.fileSize = fileSize;
this.file = new File(fileName);
this.fileFromOffset = Long.parseLong(this.file.getName());
boolean ok = false;
ensureDirOK(this.file.getParent());
try {
this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);
TOTAL_MAPPED_FILES.incrementAndGet();
ok = true;
} catch (FileNotFoundException e) {
log.error("create file channel " this.fileName " Failed. ", e);
throw e;
} catch (IOException e) {
log.error("map file " this.fileName " Failed. ", e);
throw e;
} finally {
if (!ok && this.fileChannel != null) {
this.fileChannel.close();
}
}
}开启transientStorePoolEnable
public void init(final String fileName, final int fileSize,
final TransientStorePool transientStorePool) throws IOException {
init(fileName, fileSize);
this.writeBuffer = transientStorePool.borrowBuffer(); //初始化writeBuffer
this.transientStorePool = transientStorePool;
}MappedFile提交
提交数据到FileChannel，commitLeastPages为本次提交最小的页数，如果待提交数据不满commitLeastPages，则不执行本次提交操作。如果writeBuffer如果为空，直接返回writePosition指针，无需执行commit操作，表名commit操作主体是writeBuffer。
代码语言：javascript复制public int commit(final int commitLeastPages) {
if (writeBuffer == null) {
//no need to commit data to file channel, so just regard wrotePosition as committedPosition.
return this.wrotePosition.get();
}
//判断是否满足提交条件
if (this.isAbleToCommit(commitLeastPages)) {
if (this.hold()) {
commit0(commitLeastPages);
this.release();
} else {
log.warn("in commit, hold failed, commit offset = " this.committedPosition.get());
}
}
// 所有数据提交后,清空缓冲区
if (writeBuffer != null && this.transientStorePool != null && this.fileSize == this.committedPosition.get()) {
this.transientStorePool.returnBuffer(writeBuffer);
this.writeBuffer = null;
}
return this.committedPosition.get();
}MappedFile#isAbleToCommit
判断是否执行commit操作，如果文件已满返回true；如果commitLeastpages大于0，则比较writePosition与上一次提交的指针commitPosition的差值，除以OS_PAGE_SIZE得到当前脏页的数量，如果大于commitLeastPages则返回true，如果commitLeastpages小于0表示只要存在脏页就提交。
代码语言：javascript复制protected boolean isAbleToCommit(final int commitLeastPages) {
//已经刷盘指针
int flush = this.committedPosition.get();
//文件写指针
int write = this.wrotePosition.get();
//写满刷盘
if (this.isFull()) {
return true;
}
if (commitLeastPages > 0) {
//文件内容达到commitLeastPages页数,则刷盘
return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= commitLeastPages;
}
return write > flush;
}MappedFile#commit0
具体提交的实现，首先创建WriteBuffer区共享缓存区，然后将新创建的position回退到上一次提交的位置（commitPosition），设置limit为wrotePosition（当前最大有效数据指针），然后把commitPosition到wrotePosition的数据写入到FileChannel中，然后更新committedPosition指针为wrotePosition。commit的作用就是将MappedFile的writeBuffer中数据提交到文件通道FileChannel中。
代码语言：javascript复制protected void commit0(final int commitLeastPages) {
//写指针
int writePos = this.wrotePosition.get();
//上次提交指针
int lastCommittedPosition = this.committedPosition.get();
if (writePos - this.committedPosition.get() > 0) {
try {
//复制共享内存区域
ByteBuffer byteBuffer = writeBuffer.slice();
//设置提交位置是上次提交位置
byteBuffer.position(lastCommittedPosition);
//最大提交数量
byteBuffer.limit(writePos);
//设置fileChannel位置为上次提交位置
this.fileChannel.position(lastCommittedPosition);
//将lastCommittedPosition到writePos的数据复制到FileChannel中
this.fileChannel.write(byteBuffer);
//重置提交位置
this.committedPosition.set(writePos);
} catch (Throwable e) {
log.error("Error occurred when commit data to FileChannel.", e);
}
}
}MappedFile#flush
刷写磁盘，直接调用MappedByteBuffer或fileChannel的force方法将内存中的数据持久化到磁盘，那么flushedPosition应该等于MappedByteBuffer中的写指针；如果writeBuffer不为空，则flushPosition应该等于上一次的commit指针；因为上一次提交的数据就是进入到MappedByteBuffer中的数据；如果writeBuffer为空，数据时直接进入到MappedByteBuffer，wrotePosition代表的是MappedByteBuffer中的指针，故设置flushPosition为wrotePosition。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6pUyKhzD-1631150365520)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/flush.jpg)]
代码语言：javascript复制public int flush(final int flushLeastPages) {
//数据达到刷盘条件
if (this.isAbleToFlush(flushLeastPages)) {
//加锁，同步刷盘
if (this.hold()) {
//获得读指针
int value = getReadPosition();
try {
//数据从writeBuffer提交数据到fileChannel再刷新到磁盘
if (writeBuffer != null || this.fileChannel.position() != 0) {
this.fileChannel.force(false);
} else {
//从mmap刷新数据到磁盘
this.mappedByteBuffer.force();
}
} catch (Throwable e) {
log.error("Error occurred when force data to disk.", e);
}
//更新刷盘位置
this.flushedPosition.set(value);
this.release();
} else {
log.warn("in flush, hold failed, flush offset = " this.flushedPosition.get());
this.flushedPosition.set(getReadPosition());
}
}
return this.getFlushedPosition();
}MappedFile#getReadPosition
获取当前文件最大可读指针。如果writeBuffer为空，则直接返回当前的写指针；如果writeBuffer不为空，则返回上一次提交的指针。在MappedFile设置中,只有提交了的数据（写入到MappedByteBuffer或FileChannel中的数据）才是安全的数据
代码语言：javascript复制public int getReadPosition() {
//如果writeBuffer为空,刷盘的位置就是应该等于上次commit的位置,如果为空则为mmap的写指针
return this.writeBuffer == null ? this.wrotePosition.get() : this.committedPosition.get();
}MappedFile#selectMappedBuffer
查找pos到当前最大可读之间的数据，由于在整个写入期间都未曾改MappedByteBuffer的指针，如果mappedByteBuffer.slice()方法返回的共享缓存区空间为整个MappedFile，然后通过设置ByteBuffer的position为待查找的值，读取字节长度当前可读最大长度，最终返回的ByteBuffer的limit为size。整个共享缓存区的容量为（MappedFile#fileSize-pos）。故在操作SelectMappedBufferResult不能对包含在里面的ByteBuffer调用filp方法。
代码语言：javascript复制public SelectMappedBufferResult selectMappedBuffer(int pos) {
//获得最大可读指针
int readPosition = getReadPosition();
//pos小于最大可读指针,并且大于0
if (pos < readPosition && pos >= 0) {
if (this.hold()) {
//复制mappedByteBuffer读共享区
ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
//设置读指针位置
byteBuffer.position(pos);
//获得可读范围
int size = readPosition - pos;
//设置最大刻度范围
ByteBuffer byteBufferNew = byteBuffer.slice();
byteBufferNew.limit(size);
return new SelectMappedBufferResult(this.fileFromOffset pos, byteBufferNew, size, this);
}
}
return null;
}MappedFile#shutdown
MappedFile文件销毁的实现方法为public boolean destory(long intervalForcibly)，intervalForcibly表示拒绝被销毁的最大存活时间。
代码语言：javascript复制public void shutdown(final long intervalForcibly) {
if (this.available) {
//关闭MapedFile
this.available = false;
//设置当前关闭时间戳
this.firstShutdownTimestamp = System.currentTimeMillis();
//释放资源
this.release();
} else if (this.getRefCount() > 0) {
if ((System.currentTimeMillis() - this.firstShutdownTimestamp) >= intervalForcibly) {
this.refCount.set(-1000 - this.getRefCount());
this.release();
}
}
}3）TransientStorePool
短暂的存储池。RocketMQ单独创建一个MappedByteBuffer内存缓存池，用来临时存储数据，数据先写入该内存映射中，然后由commit线程定时将数据从该内存复制到与目标物理文件对应的内存映射中。RocketMQ引入该机制主要的原因是提供一种内存锁定，将当前堆外内存一直锁定在内存中，避免被进程将内存交换到磁盘。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-45ywNKXY-1631150365521)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/TransientStorePool.png)]
代码语言：javascript复制private final int poolSize; //availableBuffers个数
private final int fileSize; //每隔ByteBuffer大小
private final Deque<ByteBuffer> availableBuffers; //ByteBuffer容器。双端队列初始化
代码语言：javascript复制public void init() {
//创建poolSize个堆外内存
for (int i = 0; i < poolSize; i ) {
ByteBuffer byteBuffer = ByteBuffer.allocateDirect(fileSize);
final long address = ((DirectBuffer) byteBuffer).address();
Pointer pointer = new Pointer(address);
//使用com.sun.jna.Library类库将该批内存锁定,避免被置换到交换区,提高存储性能
LibC.INSTANCE.mlock(pointer, new NativeLong(fileSize));
availableBuffers.offer(byteBuffer);
}
}2.4.5 实时更新消息消费队列与索引文件
消息消费队文件、消息属性索引文件都是基于CommitLog文件构建的，当消息生产者提交的消息存储在CommitLog文件中，ConsumerQueue、IndexFile需要及时更新，否则消息无法及时被消费，根据消息属性查找消息也会出现较大延迟。RocketMQ通过开启一个线程ReputMessageService来准实时转发CommitLog文件更新事件，相应的任务处理器根据转发的消息及时更新ConsumerQueue、IndexFile文件。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fiaCqAQ4-1631150365521)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/消息存储结构.png)]
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-G46Hv8V5-1631150365522)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/构建消息消费队列和索引文件.png)]
代码：DefaultMessageStore：start
代码语言：javascript复制//设置CommitLog内存中最大偏移量
this.reputMessageService.setReputFromOffset(maxPhysicalPosInLogicQueue);
//启动
this.reputMessageService.start();代码：DefaultMessageStore：run
代码语言：javascript复制public void run() {
DefaultMessageStore.log.info(this.getServiceName() " service started");
//每隔1毫秒就继续尝试推送消息到消息消费队列和索引文件
while (!this.isStopped()) {
try {
Thread.sleep(1);
this.doReput();
} catch (Exception e) {
DefaultMessageStore.log.warn(this.getServiceName() " service has exception. ", e);
}
}
DefaultMessageStore.log.info(this.getServiceName() " service end");
}代码：DefaultMessageStore：deReput
代码语言：javascript复制//从result中循环遍历消息,一次读一条,创建DispatherRequest对象。
for (int readSize = 0; readSize < result.getSize() && doNext; ) {
DispatchRequest dispatchRequest = DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false);
int size = dispatchRequest.getBufferSize() == -1 ? dispatchRequest.getMsgSize() : dispatchRequest.getBufferSize();
if (dispatchRequest.isSuccess()) {
if (size > 0) {
DefaultMessageStore.this.doDispatch(dispatchRequest);
}
}
}DispatchRequest
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-M7A3JMtx-1631150365522)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/DispatchRequest.png)]
代码语言：javascript复制String topic; //消息主题名称
int queueId; //消息队列ID
long commitLogOffset; //消息物理偏移量
int msgSize; //消息长度
long tagsCode; //消息过滤tag hashCode
long storeTimestamp; //消息存储时间戳
long consumeQueueOffset; //消息队列偏移量
String keys; //消息索引key
boolean success; //是否成功解析到完整的消息
String uniqKey; //消息唯一键
int sysFlag; //消息系统标记
long preparedTransactionOffset; //消息预处理事务偏移量
Map<String, String> propertiesMap; //消息属性
byte[] bitMap; //位图1）转发到ConsumerQueue
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-hsYhH8du-1631150365522)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/消息分发到消息消费队列.png)]
代码语言：javascript复制class CommitLogDispatcherBuildConsumeQueue implements CommitLogDispatcher {
@Override
public void dispatch(DispatchRequest request) {
final int tranType = MessageSysFlag.getTransactionValue(request.getSysFlag());
switch (tranType) {
case MessageSysFlag.TRANSACTION_NOT_TYPE:
case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
//消息分发
DefaultMessageStore.this.putMessagePositionInfo(request);
break;
case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
break;
}
}
}代码：DefaultMessageStore#putMessagePositionInfo
代码语言：javascript复制public void putMessagePositionInfo(DispatchRequest dispatchRequest) {
//获得消费队列
ConsumeQueue cq = this.findConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId());
//消费队列分发消息
cq.putMessagePositionInfoWrapper(dispatchRequest);
}代码：DefaultMessageStore#putMessagePositionInfo
代码语言：javascript复制//依次将消息偏移量、消息长度、tag写入到ByteBuffer中
this.byteBufferIndex.flip();
this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE);
this.byteBufferIndex.putLong(offset);
this.byteBufferIndex.putInt(size);
this.byteBufferIndex.putLong(tagsCode);
//获得内存映射文件
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(expectLogicOffset);
if (mappedFile != null) {
//将消息追加到内存映射文件,异步输盘
return mappedFile.appendMessage(this.byteBufferIndex.array());
}2）转发到Index
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZlJvakmG-1631150365523)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/消息分发到索引文件.png)]
代码语言：javascript复制class CommitLogDispatcherBuildIndex implements CommitLogDispatcher {
@Override
public void dispatch(DispatchRequest request) {
if (DefaultMessageStore.this.messageStoreConfig.isMessageIndexEnable()) {
DefaultMessageStore.this.indexService.buildIndex(request);
}
}
}代码：DefaultMessageStore#buildIndex
代码语言：javascript复制public void buildIndex(DispatchRequest req) {
//获得索引文件
IndexFile indexFile = retryGetAndCreateIndexFile();
if (indexFile != null) {
//获得文件最大物理偏移量
long endPhyOffset = indexFile.getEndPhyOffset();
DispatchRequest msg = req;
String topic = msg.getTopic();
String keys = msg.getKeys();
//如果该消息的物理偏移量小于索引文件中的最大物理偏移量,则说明是重复数据,忽略本次索引构建
if (msg.getCommitLogOffset() < endPhyOffset) {
return;
}
final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
switch (tranType) {
case MessageSysFlag.TRANSACTION_NOT_TYPE:
case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
break;
case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
return;
}
//如果消息ID不为空,则添加到Hash索引中
if (req.getUniqKey() != null) {
indexFile = putKey(indexFile, msg, buildKey(topic, req.getUniqKey()));
if (indexFile == null) {
return;
}
}
//构建索引key,RocketMQ支持为同一个消息建立多个索引,多个索引键空格隔开.
if (keys != null && keys.length() > 0) {
String[] keyset = keys.split(MessageConst.KEY_SEPARATOR);
for (int i = 0; i < keyset.length; i ) {
String key = keyset[i];
if (key.length() > 0) {
indexFile = putKey(indexFile, msg, buildKey(topic, key));
if (indexFile == null) {
return;
}
}
}
}
} else {
log.error("build index error, stop building index");
}
}2.4.6 消息队列和索引文件恢复
由于RocketMQ存储首先将消息全量存储在CommitLog文件中，然后异步生成转发任务更新ConsumerQueue和Index文件。如果消息成功存储到CommitLog文件中，转发任务未成功执行，此时消息服务器Broker由于某个愿意宕机，导致CommitLog、ConsumerQueue、IndexFile文件数据不一致。如果不加以人工修复的话，会有一部分消息即便在CommitLog中文件中存在，但由于没有转发到ConsumerQueue，这部分消息将永远复发被消费者消费。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-piDATzgJ-1631150365523)( https://jokerdj-blog-images.oss-cn-shanghai.aliyuncs.com/RocketMQ/img/文件恢复总体流程.png)]
####1）存储文件加载
代码：DefaultMessageStore#load
判断上一次是否异常退出。实现机制是Broker在启动时创建abort文件，在退出时通过JVM钩子函数删除abort文件。如果下次启动时存在abort文件。说明Broker时异常退出的，CommitLog与ConsumerQueue数据有可能不一致，需要进行修复。
代码语言：javascript复制//判断临时文件是否存在
boolean lastExitOK = !this.isTempFileExist();
//根据临时文件判断当前Broker是否异常退出
private boolean isTempFileExist() {
String fileName = StorePathConfigHelper
.getAbortFile(this.messageStoreConfig.getStorePathRootDir());
File file = new File(fileName);
return file.exists();
}代码：DefaultMessageStore#load
代码语言：javascript复制//加载延时队列
if (null != scheduleMessageService) {
result = result && this.scheduleMessageService.load();
}
// 加载CommitLog文件
result = result && this.commitLog.load();
// 加载消费队列文件
result = result && this.loadConsumeQueue();
if (result) {
//加载存储监测点,监测点主要记录CommitLog文件、ConsumerQueue文件、Index索引文件的刷盘点
this.storeCheckpoint =new StoreCheckpoint(StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
//加载index文件
this.indexService.load(lastExitOK);
//根据Broker是否异常退出,执行不同的恢复策略
this.recover(lastExitOK);
}代码：MappedFileQueue#load
加载CommitLog到映射文件
代码语言：javascript复制//指向CommitLog文件目录
File dir = new File(this.storePath);
//获得文件数组
File[] files = dir.listFiles();
if (files != null) {
// 文件排序
Arrays.sort(files);
//遍历文件
for (File file : files) {
//如果文件大小和配置文件不一致,退出
if (file.length() != this.mappedFileSize) {
return false;
}
try {
//创建映射文件
MappedFile mappedFile = new MappedFile(file.getPath(), mappedFileSize);
mappedFile.setWrotePosition(this.mappedFileSize);
mappedFile.setFlushedPosition(this.mappedFileSize);
mappedFile.setCommittedPosition(this.mappedFileSize);
//将映射文件添加到队列
this.mappedFiles.add(mappedFile);
log.info("load " file.getPath() " OK");
} catch (IOException e) {
log.error("load file " file " error", e);
return false;
}
}
}
return true;代码：DefaultMessageStore#loadConsumeQueue
加载消息消费队列
代码语言：javascript复制//执行消费队列目录
File dirLogic = new File(StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()));
//遍历消费队列目录
File[] fileTopicList = dirLogic.listFiles();
if (fileTopicList != null) {
for (File fileTopic : fileTopicList) {
//获得子目录名称,即topic名称
String topic = fileTopic.getName();
//遍历子目录下的消费队列文件
File[] fileQueueIdList = fileTopic.listFiles();
if (fileQueueIdList != null) {
//遍历文件
for (File fileQueueId : fileQueueIdList) {
//文件名称即队列ID
int queueId;
try {
queueId = Integer.parseInt(fileQueueId.getName());
} catch (NumberFormatException e) {
continue;
}
//创建消费队列并加载到内存
ConsumeQueue logic = new ConsumeQueue(
topic,
queueId,
StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
this.getMessageStoreConfig().getMapedFileSizeConsumeQueue(),
this);
this.putConsumeQueue(topic, queueId, logic);
if (!logic.load()) {
return false;
}
}
}
}
}
log.info("load logics queue all over, OK");
return true;代码：IndexService#load
加载索引文件
代码语言：javascript复制public boolean load(final boolean lastExitOK) {
//索引文件目录
File dir = new File(this.storePath);
//遍历索引文件
File[] files = dir.listFiles();
if (files != null) {
//文件排序
Arrays.sort(files);
//遍历文件
for (File file : files) {
try {
//加载索引文件
IndexFile f = new IndexFile(file.getPath(), this.hashSlotNum, this.indexNum, 0, 0);
f.load();
if (!lastExitOK) {
//索引文件上次的刷盘时间小于该索引文件的消息时间戳,该文件将立即删除
if (f.getEndTimestamp() > this.defaultMessageStore.getStoreCheckpoint()
.getIndexMsgTimestamp()) {
f.destroy(0);
continue;
}
}
//将索引文件添加到队列
log.info("load index file OK, " f.getFileName());
this.indexFileList.add(f);
} catch (IOException e) {
log.error("load file {} error", file, e);
return false;
} catch (NumberFormatException e) {
log.error("load file {} error", file, e);
}
}
}
return true;
}代码：DefaultMessageStore#recover
文件恢复，根据Broker是否正常退出执行不同的恢复策略
代码语言：javascript复制private void recover(final boolean lastExitOK) {
//获得最大的物理便宜消费队列
long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue();
if (lastExitOK) {
//正常恢复
this.commitLog.recoverNormally(maxPhyOffsetOfConsumeQueue);
} else {
//异常恢复
this.commitLog.recoverAbnormally(maxPhyOffsetOfConsumeQueue);
