一、RocketMQ简介
Apache RocketMQ是一个采用Java语言开发的分布式的消息系统，由阿里巴巴团队开发，与2016年底贡献给Apache，成为了Apache的一个顶级项目。
在阿里内部，RocketMQ 很好地服务了 集 团大大小小上千个应用，在每年的双十一当天，更有不可思议的万亿级消息通过 RocketMQ 流转(在 2017 年的双十一当天，整个阿里巴巴集团通过 RocketMQ 流转的线上消息达到了 万亿级，峰值 TPS 达到 5600 万)，在阿里大中台策略上发挥着举足轻重的作用 。
地址：http://rocketmq.apache.org/
RocketMQ的历史发展
阿里巴巴消息中间件起源 于 2001 年的五彩石项目， Notify 在这期间应运而生，用于交易核心消息的流转 。
2010 年， B2B 开始大规模使用 ActiveMQ 作为消息内核，随着阿里业务 的快速发展，急需一款支持顺序消息，拥有海量消息堆积能力的消息中间件， MetaQ 1.0 在 2011 年诞生 。
2012年， MetaQ已经发展到了3.0版本，并抽象出了通用的消息引擎 RocketMQ。 随后，对 RocketMQ 进行了开源 ， 阿里的消息中间件正式走人了 公众视野 。
2015年， RocketMQ已经经历了多年双十一的洗礼，在可用性、 可靠性以 及稳定性等方面都有出色的表现。与此同时 ，云计算大行其道， 阿里消息中间 件基于 RocketMQ推出了 Aliware MQ 1.0，开始为阿里云上成千上万家企业提 供消息服务
2016 年， MetaQ 在双十一期间承载了万亿级消息的流转，跨越了一个新的里程碑 ，同时 RocketMQ 进入Apache 孵化 。
核心概念说明
broker接收生产者发的消息，存储消息数据。broker会注册到nameserver中，生产者和消费者通过nameserver找到broker，这样生产者就可以给broker发消息，消费者订阅并获取消息。
broker可以做集群，nameserver也可以做集群，当然，生产者和消费者也可以做集群。
broker的topic中的messageQueue来存储数据。broker中会有多个topic，topic中会有多个messageQueue。
Producer
消息生产者，负责产生消息，一般由业务系统负责产生消息。
Producer Group
一类 Producer 的集合名称，这类 Producer 通常发送一类消息，且发送逻辑一致。
Consumer
消息费者，负责消费消息，一般是后台系统负责异步消费。消费者分为两种：push Consumer和Pull Consumer。
Push Consumer
服务端向消费者端推送消息
Pull Consumer
消费者端向服务定时拉取消息
Consumer Group
一类 Consumer 的集合名称，这类 Consumer 通常消费一类消息，且消费逻辑一致。
NameServer
集群架构中的组织协调员，收集broker的工作情况，不负责消息的处理
Broker
是RocketMQ的核心负责消息的发送、接收、高可用等（真正干活的）
需要定时发送自身情况到NameServer，默认10秒发送一次，超时2分钟会认为该broker失效。
Topic
不同类型的消息以不同的Topic名称进行区分，如User、Order等，是逻辑概念
Message Queue
消息队列，用于存储消息，是物理概念，是真实存在的。
二、部署安装
1、非docker安装
点击download
4.9.3版本安装
我们下载4.9.3版本，下载后如下所示：
rocketmq要开放四个端口：9876，10911，10909，10912，11011
（1）、在/usr/local目录下新建rocketmq文件夹
mkdir rocketmq
cd rocketmq
（2）、将下载的rocketmq-all-4.9.3-bin-release.zip上传到rocketmq目录下
（3）、解压缩
unzip rocketmq-all-4.9.3-bin-release.zip
（4）、启动nameserver
cd rocketmq-4.9.3/bin
./mqnamesrv
The Name Server boot success. serializeType=JSON 看到这个表示已经提供成功
[root@hecs-xxx bin]# ./mqnamesrv
Java HotSpot(TM) 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release
Java HotSpot(TM) 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release.
The Name Server boot success. serializeType=JSON
另外，停止nameserver命令为：
./mqshutdown namesrv
（5）、启动broker
./mqbroker -n localhost:9876
-n 指定nameserver地址和端口，9876为nameserver默认的端口，IP根据你自己的IP填写。
另外，停止broker命令为
./mqshutdown broker
启动出错
Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000005c0000000, 8589934592, 0) failed; error='Cannot allocate memory' (errno=12) # # There is insufficient memory for the Java Runtime Environment to continue. # Native memory allocation (mmap) failed to map 8589934592 bytes for committing reserved memory. # An error report file with more information is saved as: # /usr/local/rocketmq/rocketmq-all-4.3.2-bin-release/bin/hs_err_pid32159.log
启动错误，是因为内存不够，导致启动失败，原因：RocketMQ的配置默认是生产环境的配置，设置的jvm的内存大小值比较大，
runserver.sh中的默认参数
JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
runbroker.sh中的默认参数
JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"
对于学习而言没有必要设置这么大，测试环境的内存往往都不是很大，所以需要调整默认值。
（6）、调整默认的内存大小参数
cd bin/
vim runserver.sh
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=128m"
cd bin/
vim runbroker.sh
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m"
如下所示：
（7）、重新启动
./mqbroker -n localhost:9876
如下：
[root@hecs-xxx bin]# ./mqbroker -n localhost:9876 The broker[hecs-xxx, 192.168.0.150:10911] boot success. serializeType=JSON and name server is localhost:9876
说明已经成功的启动了。注意：10911是broker的端口。9876是nameserver的端口。生产者需要通过nameServer找到broker，broker的地址为192.168.0.150:10911。
下面通过rocketmq中bin目录下的工具包tools.sh进行发送消息测试：
先设置环境变量（本机的9876）
export NAMESRV_ADDR=localhost:9876
如果不执行会报：
执行脚本
sh tools.sh org.apache.rocketmq.example.quickstart.Producer
结果：
SendResult [sendStatus=SEND_OK, msgId=C0AC00013CA9135FBAA411A9C19D03E4, offsetMsgId=C0A8009600002A9F00000000000F0A8E, messageQueue=MessageQueue [topic=TopicTest, brokerName=hecs-140905, queueId=7], queueOffset=124] SendResult [sendStatus=SEND_OK, msgId=C0AC00013CA9135FBAA411A9C19E03E5, offsetMsgId=C0A8009600002A9F00000000000F0B4E, messageQueue=MessageQueue [topic=TopicTest, brokerName=hecs-140905, queueId=0], queueOffset=1229] SendResult [sendStatus=SEND_OK, msgId=C0AC00013CA9135FBAA411A9C19E03E6, offsetMsgId=C0A8009600002A9F00000000000F0C0E, messageQueue=MessageQueue [topic=TopicTest, brokerName=hecs-140905, queueId=1], queueOffset=1228] SendResult [sendStatus=SEND_OK, msgId=C0AC00013CA9135FBAA411A9C19F03E7, offsetMsgId=C0A8009600002A9F00000000000F0CCE, messageQueue=MessageQueue [topic=TopicTest, brokerName=hecs-140905, queueId=2], queueOffset=1228] 10:18:57.832 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[127.0.0.1:9876] result: true 10:18:57.836 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[127.0.0.1:9876] result: true 10:18:57.836 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[192.168.0.150:10911] result: true
可以正常发送消息
测试接收消息：
export NAMESRV_ADDR=localhost:9876
sh tools.sh org.apache.rocketmq.example.quickstart.Consumer
注意：要先加入环境变量再执行。
结果如下：
ConsumeMessageThread_please_rename_unique_group_name_4_9 Receive New Messages: [MessageExt [brokerName=hecs-140905, queueId=3, storeSize=192, queueOffset=1469, sysFlag=0, bornTimestamp=1680578753981, bornHost=/192.168.0.150:60844, storeTimestamp=1680578753981, storeHost=/192.168.0.150:10911, msgId=C0A8009600002A9F000000000014AFB2, commitLogOffset=1355698, bodyCRC=994843245, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1479, CONSUME_START_TIME=1680578798444, UNIQ_KEY=C0AC000151AE135FBAA411E709BD039C, CLUSTER=DefaultCluster, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 57, 50, 52], transactionId='null'}]] ConsumeMessageThread_please_rename_unique_group_name_4_20 Receive New Messages: [MessageExt [brokerName=hecs-140905, queueId=3, storeSize=192, queueOffset=1468, sysFlag=0, bornTimestamp=1680578753972, bornHost=/192.168.0.150:60844, storeTimestamp=1680578753973, storeHost=/192.168.0.150:10911, msgId=C0A8009600002A9F000000000014A9B2, commitLogOffset=1354162, bodyCRC=2121214082, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1479, CONSUME_START_TIME=1680578798444, UNIQ_KEY=C0AC000151AE135FBAA411E709B40394, CLUSTER=DefaultCluster, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 57, 49, 54], transactionId='null'}]]
可查看日志
tail -f ~/logs/rocketmqlogs/namesrv.log
tail -f ~/logs/rocketmqlogs/broker.log
查看brokerip注册到nameservier的ip情况
[root@hecs-xxx bin]# ./mqadmin clusterList -n localhost:9876 RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.InternalThreadLocalMap). RocketMQLog:WARN Please initialize the logger system properly. #Cluster Name #Broker Name #BID #Addr #Version #InTPS(LOAD) #OutTPS(LOAD) #PCWait(ms) #Hour #SPACE DefaultCluster hecs-xxx 0 192.168.0.150:10911 V4_9_3 0.00(0,0ms) 0.00(0,0ms) 0 466827.48 0.1900
编写Java代码进行测试
（1）、创建maven工程
（2）、添加依赖
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.3.2</version>
</dependency>
</dependencies>
（3）、编写测试类
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
public class SyncProducer {
public static void main(String[] args) throws Exception {
//Instantiate with a producer group name.
DefaultMQProducer producer = new DefaultMQProducer("test-group");
// Specify name server addresses.
producer.setNamesrvAddr("云服务器IP:9876");
//Launch the instance.
producer.start();
for (int i = 0; i < 100; i++) {
//Create a message instance, specifying topic, tag and message body.
Message msg = new Message("TopicTest11" /* Topic */,
"TagA" /* Tag */,
("Hello RocketMQ " +
i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
//Call send message to deliver message to one of brokers.
SendResult sendResult = producer.send(msg);
System.out.printf("%s%n", sendResult);
}
//Shut down once the producer instance is not longer in use.
producer.shutdown();
}
}
执行后报错如下：
测试结果会发现，发送消息会报错。原因是什么呢？通过工具包发送消息没有问题，为什么编码就不行呢？
仔细观察broker启动的信息：
[root@hecs-xxx bin]# ./mqbroker -n localhost:9876 The broker[hecs-140905, 192.168.0.150:10911] boot success. serializeType=JSON and name server is localhost:9876
会发现，broker的ip地址是192.172.0.1，那么在开发机上是不可能访问到的。
所以，需要指定broker的ip地址。
在/root目录下创建broker.conf（注意：要另外创建一个broker.conf，不要在../conf/broker.cnf上面添加下面的配置）
mkdir /root/rocketmq/rmqbroker/conf/ -p
创建broker.conf
touch broker.conf
vim broker.conf
添加内容如下：
brokerIP1=云服务器IP
namesrvAddr=云服务器IP:9876
brokerName=broker_haoke_im
先停止broker，再启动broker，通过 -c 指定配置文件（此时不需要指定-n了，因为broker.conf中有）
./mqshutdown broker
./mqbroker -c /root/rocketmq/rmqbroker/conf/broker.conf
启动结果如下：
[root@hecs-xxx bin]# ./mqbroker -c /root/rocketmq/rmqbroker/conf/broker.conf The broker[broker_haoke_im, 114.xxx.xxx.xxx:10911] boot success. serializeType=JSON and name server is localhost:9876
查看所有broker列表：
./mqadmin clusterList -n localhost:9876
结果：
[root@hecs-140905 bin]# ./mqadmin clusterList -n localhost:9876 RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.InternalThreadLocalMap). RocketMQLog:WARN Please initialize the logger system properly. #Cluster Name #Broker Name #BID #Addr #Version #InTPS(LOAD) #OutTPS(LOAD) #PCWait(ms) #Hour #SPACE org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to 114.xxx.xxx.xxx:10911 failed at org.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeSync(NettyRemotingClient.java:407) at org.apache.rocketmq.client.impl.MQClientAPIImpl.getBrokerRuntimeInfo(MQClientAPIImpl.java:1288) at org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl.fetchBrokerRuntimeStats(DefaultMQAdminExtImpl.java:287) at org.apache.rocketmq.tools.admin.DefaultMQAdminExt.fetchBrokerRuntimeStats(DefaultMQAdminExt.java:238) at org.apache.rocketmq.tools.command.cluster.ClusterListSubCommand.printClusterBaseInfo(ClusterListSubCommand.java:212) at org.apache.rocketmq.tools.command.cluster.ClusterListSubCommand.execute(ClusterListSubCommand.java:88) at org.apache.rocketmq.tools.command.MQAdminStartup.main0(MQAdminStartup.java:146) at org.apache.rocketmq.tools.command.MQAdminStartup.main(MQAdminStartup.java:97) DefaultCluster broker_haoke_im 0 114.xxx.xxx.xxx:10911 0.00(,ms) 0.00(,ms) 0.00 0.0000
再次执行测试类，控制台打印如下：
初次安装完成后，后面再次启动命令：
./mqnamesrv
./mqbroker -c /root/rocketmq/rmqbroker/conf/broker.conf
4.3.2版本安装
我们下载4.3.2版本，下载后如下所示：
rocketmq要开放四个端口：9876，10911，10909，10912，11011
（1）、在/usr/local目录下新建rocketmq文件夹
mkdir rocketmq
cd rocketmq
（2）、将下载的rocketmq-all-4.3.2-bin-release.zip上传到rocketmq目录下
（3）、解压缩
unzip rocketmq-all-4.3.2-bin-release.zip
（4）、启动nameserver
cd rocketmq-all-4.3.2-bin-release/bin
./mqnamesrv
The Name Server boot success. serializeType=JSON 看到这个表示已经提供成功
[root@hecs-xxx bin]# ./mqnamesrv Java HotSpot(TM) 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release Java HotSpot(TM) 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release. The Name Server boot success. serializeType=JSON
另外，停止nameserver命令为：
./mqshutdown namesrv
（5）、启动broker
./mqbroker -n localhost:9876
-n 指定nameserver地址和端口，9876为nameserver默认的端口，IP根据你自己的IP填写。
另外，停止broker命令为
./mqshutdown broker
启动出错
Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000005c0000000, 8589934592, 0) failed; error='Cannot allocate memory' (errno=12) # # There is insufficient memory for the Java Runtime Environment to continue. # Native memory allocation (mmap) failed to map 8589934592 bytes for committing reserved memory. # An error report file with more information is saved as: # /usr/local/rocketmq/rocketmq-all-4.3.2-bin-release/bin/hs_err_pid32159.log
启动错误，是因为内存不够，导致启动失败，原因：RocketMQ的配置默认是生产环境的配置，设置的jvm的内存大小值比较大，
runserver.sh中的默认参数
JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
runbroker.sh中的默认参数
JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"
对于学习而言没有必要设置这么大，测试环境的内存往往都不是很大，所以需要调整默认值。
（6）、调整默认的内存大小参数
cd bin/ vim runserver.sh JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx128m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=128m" cd bin/ vim runbroker.sh JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx128m -Xmn128m"
（7）、重新启动
./mqbroker -n localhost:9876
如下：
[root@hecs-xxx bin]# ./mqbroker -n localhost:9876 The broker[hecs-xxx, 192.172.0.1:10911] boot success. serializeType=JSON and name server is 172.16.55.185:9876
说明已经成功的启动了。
下面通过rocketmq中bin目录下的工具包tools.sh进行发送消息测试：
先设置环境变量（本机的9876）
export NAMESRV_ADDR=localhost:9876
如果不执行会报：
执行脚本
cd bin
sh tools.sh org.apache.rocketmq.example.quickstart.Producer
发现报错：no route info of this topic，TopicTest
说明没有手动创建topic，且没有自动创建topic
先关闭broker
[root@hecs-xxx conf]# jps 32513 BrokerStartup 32641 Jps 32466 NamesrvStartup 915 WrapperSimpleApp [root@hecs-xxx conf]# kill -9 32523
进入rocketmq配置文件目录conf,找到broker.conf文件
cd ../conf
vim broker.conf
在文件末尾添加
autoCreateTopicEnable=true
保存退出
重新启动
./mqbroker -n localhost:9876
结果如下：
[root@hecs-xxx bin]# ./mqbroker -n localhost:9876 The broker[hecs-xxx, 192.172.0.1:10911] boot success. serializeType=JSON and name server is localhost:9876
注意：10911是broker的端口。9876是nameserver的端口。生产者需要通过nameServer找到broker，broker的地址为192.172.0.1:10911。
再执行
export NAMESRV_ADDR=localhost:9876 sh tools.sh org.apache.rocketmq.example.quickstart.Producer
结果：
SendResult [sendStatus=SEND_OK, msgId=AC11000101466D6F6E280C9FE3E903E3, offsetMsgId=C0AC000100002A9F000000000002BB2E, messageQueue=MessageQueue [topic=TopicTest, brokerName=hecs-140905, queueId=3], queueOffset=248] SendResult [sendStatus=SEND_OK, msgId=AC11000101466D6F6E280C9FE3EA03E4, offsetMsgId=C0AC000100002A9F000000000002BBE2, messageQueue=MessageQueue [topic=TopicTest, brokerName=hecs-140905, queueId=0], queueOffset=249] SendResult [sendStatus=SEND_OK, msgId=AC11000101466D6F6E280C9FE3EA03E5, offsetMsgId=C0AC000100002A9F000000000002BC96, messageQueue=MessageQueue [topic=TopicTest, brokerName=hecs-140905, queueId=1], queueOffset=249] SendResult [sendStatus=SEND_OK, msgId=AC11000101466D6F6E280C9FE3EB03E6, offsetMsgId=C0AC000100002A9F000000000002BD4A, messageQueue=MessageQueue [topic=TopicTest, brokerName=hecs-140905, queueId=2], queueOffset=249] SendResult [sendStatus=SEND_OK, msgId=AC11000101466D6F6E280C9FE3EB03E7, offsetMsgId=C0AC000100002A9F000000000002BDFE, messageQueue=MessageQueue [topic=TopicTest, brokerName=hecs-140905, queueId=3], queueOffset=249] 10:50:05.170 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[127.0.0.1:9876] result: true 10:50:05.171 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[192.172.0.1:10911] result: true 10:50:05.174 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[192.172.0.1:10909] result: true
可以正常发送消息
测试接收消息：
export NAMESRV_ADDR=localhost:9876
sh tools.sh org.apache.rocketmq.example.quickstart.Consumer
结果：
ConsumeMessageThread_18 Receive New Messages: [MessageExt [queueId=0, storeSize=180, queueOffset=241, sysFlag=0, bornTimestamp=1680490205143, bornHost=/192.168.0.150:42858, storeTimestamp=1680490205144, storeHost=/192.172.0.1:10911, msgId=C0AC000100002A9F000000000002A562, commitLogOffset=173410, bodyCRC=1595989865, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=250, CONSUME_START_TIME=1680490338152, UNIQ_KEY=AC11000101466D6F6E280C9FE3D703C4, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 57, 54, 52], transactionId='null'}]] ConsumeMessageThread_20 Receive New Messages: [MessageExt [queueId=0, storeSize=180, queueOffset=240, sysFlag=0, bornTimestamp=1680490205140, bornHost=/192.168.0.150:42858, storeTimestamp=1680490205140, storeHost=/192.172.0.1:10911, msgId=C0AC000100002A9F000000000002A292, commitLogOffset=172690, bodyCRC=1481448304, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=250, CONSUME_START_TIME=1680490338152, UNIQ_KEY=AC11000101466D6F6E280C9FE3D403C0, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 57, 54, 48], transactionId='null'}]]
从结果中，可以看出，接收消息正常
查看brokerip注册到nameservier的ip情况
[root@hecs-xxx bin]# ./mqadmin clusterList -n localhost:9876 Java HotSpot(TM) 64-Bit Server VM warning: ignoring option PermSize=128m; support was removed in 8.0 Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=128m; support was removed in 8.0 #Cluster Name #Broker Name #BID #Addr #Version #InTPS(LOAD) #OutTPS(LOAD) #PCWait(ms) #Hour #SPACE DefaultCluster hecs-xxx 0 192.172.0.1:10911 V4_3_2 0.00(0,0ms) 0.00(0,0ms) 0 466824.75 0.1726
编写Java代码进行测试
（1）、创建maven工程
（2）、添加依赖
<dependencies> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.3.2</version> </dependency> </dependencies>
（3）、编写测试类
import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper; public class SyncProducer { public static void main(String[] args) throws Exception { //Instantiate with a producer group name. DefaultMQProducer producer = new DefaultMQProducer("test-group"); // Specify name server addresses. producer.setNamesrvAddr("云服务器IP:9876"); //Launch the instance. producer.start(); for (int i = 0; i < 100; i++) { //Create a message instance, specifying topic, tag and message body. Message msg = new Message("TopicTest11" /* Topic */, "TagA" /* Tag */, ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */ ); //Call send message to deliver message to one of brokers. SendResult sendResult = producer.send(msg); System.out.printf("%s%n", sendResult); } //Shut down once the producer instance is not longer in use. producer.shutdown(); } }
执行后报错如下：
测试结果会发现，发送消息会报错。原因是什么呢？通过工具包发送消息没有问题，为什么编码就不行呢？
仔细观察broker启动的信息：
[root@hecs-xxx bin]# ./mqbroker -n localhost:9876 The broker[hecs-140905, 192.172.0.1:10911] boot success. serializeType=JSON and name server is localhost:9876
会发现，broker的ip地址是192.172.0.1，那么在开发机上是不可能访问到的。
所以，需要指定broker的ip地址。
在/root目录下创建broker.conf（注意：要另外创建一个broker.conf，不要在../conf/broker.cnf上面添加下面的配置）
mkdir /root/rocketmq/rmqbroker/conf/ -p
创建broker.conf
touch broker.conf
vim broker.conf
添加内容如下：
brokerIP1=云服务器IP namesrvAddr=localhost:9876 brokerName=broker_haoke_im
关闭broker再启动broker，通过 -c 指定配置文件（此时不需要指定-n了，因为broker.conf中有）
./mqshutdown broker
./mqbroker -c /root/rocketmq/rmqbroker/conf/broker.conf
结果如下：
The broker[broker_haoke_im, 云服务器IP:10911] boot success. serializeType=JSON and name server is localhost:9876
查看所有broker列表：
./mqadmin clusterList -n localhost:9876
结果：
[root@hecs-xxx bin]# ./mqadmin clusterList -n localhost:9876 Java HotSpot(TM) 64-Bit Server VM warning: ignoring option PermSize=128m; support was removed in 8.0 Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=128m; support was removed in 8.0 #Cluster Name #Broker Name #BID #Addr #Version #InTPS(LOAD) #OutTPS(LOAD) #PCWait(ms) #Hour #SPACE org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to <xxx:10909> failed at org.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeSync(NettyRemotingClient.java:393) at org.apache.rocketmq.client.impl.MQClientAPIImpl.getBrokerRuntimeInfo(MQClientAPIImpl.java:1121) at org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl.fetchBrokerRuntimeStats(DefaultMQAdminExtImpl.java:235) at org.apache.rocketmq.tools.admin.DefaultMQAdminExt.fetchBrokerRuntimeStats(DefaultMQAdminExt.java:202) at org.apache.rocketmq.tools.command.cluster.ClusterListSubCommand.printClusterBaseInfo(ClusterListSubCommand.java:212) at org.apache.rocketmq.tools.command.cluster.ClusterListSubCommand.execute(ClusterListSubCommand.java:88) at org.apache.rocketmq.tools.command.MQAdminStartup.main0(MQAdminStartup.java:132) at org.apache.rocketmq.tools.command.MQAdminStartup.main(MQAdminStartup.java:83) DefaultCluster broker_haoke_im 0 114.xxx.xxx.xxx:10911 0.00(,ms) 0.00(,ms) 0.00 0.0000
执行测试代码，结果如下:
RocketMQ的管理工具安装
是一个开源项目，拉取到本地以后打包即可。地址为：https://gitcode.net/mirrors/apache/rocketmq-dashboard?utm_source=csdn_github_accelerator
使用git下载下来：git clone https://gitcode.net/mirrors/apache/rocketmq-dashboard.git
修改配置文件application.yml
server: port: 8981 servlet: encoding: charset: UTF-8 enabled: true force: true ## SSL setting # ssl: # key-store: classpath:rmqcngkeystore.jks # key-store-password: rocketmq # key-store-type: PKCS12 # key-alias: rmqcngkey spring: application: name: rocketmq-dashboard logging: config: classpath:logback.xml rocketmq: config: # if this value is empty,use env value rocketmq.config.namesrvAddr NAMESRV_ADDR | now, default localhost:9876 # configure multiple namesrv addresses to manage multiple different clusters namesrvAddrs: - 114.xxx.xxx.xxx:9876 # if you use rocketmq version < 3.5.8, rocketmq.config.isVIPChannel should be false.default true isVIPChannel: # timeout for mqadminExt, default 5000ms timeoutMillis: # rocketmq-console's data path:dashboard/monitor dataPath: /tmp/rocketmq-console/data # set it false if you don't want use dashboard.default true enableDashBoardCollect: true # set the message track trace topic if you don't want use the default one msgTrackTopicName: ticketKey: ticket # must create userInfo file: ${rocketmq.config.dataPath}/users.properties if the login is required loginRequired: false useTLS: false # set the accessKey and secretKey if you used acl accessKey: # if version > 4.4.0 secretKey: # if version > 4.4.0 threadpool: config: coreSize: 10 maxSize: 10 keepAliveTime: 3000 queueSize: 5000
修改端口和namesrvAddrs
进入项目根目录rocketmq-dashboard，打包命令如下：
mvn clean package -Dmaven.test.skip=true
打包后如下所示:
cmd进入D:\install\rockermq-dashboard\rocketmq-dashboard\target目录下，命令如下：
java -jar rocketmq-dashboard-1.0.1-SNAPSHOT.jar
结果如下：
浏览器访问：http://localhost:8981/，页面如下：
如果出现报错：查看云服务器的端口是否都放开：
在右上角可以切换成中文。
三、Rocketmq的API
1、创建topic
前面我们没有创建topic也能发消息，原因是我们设置了发送消息时如果topic不存在则自动创建，而有些时候需要我们手动创建，比如生产环境会将自动创建topic的功能禁用掉，如果不禁用则会导致topic创建比较多，
import org.apache.rocketmq.client.producer.DefaultMQProducer; public class TopicDemo { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("haoke"); //设置nameserver的地址 producer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); // 启动生产者 producer.start(); /** * 创建topic，参数分别是：broker的名称（与实际保持一致），topic的名称，messagequeue的数量(默认4) * */ producer.createTopic("broker_haoke_im", "my-topic", 8); System.out.println("topic创建成功!");
// 关闭生产者 producer.shutdown(); } }
broker的名字与broker.conf中的名字一致
brokerName=broker_haoke_im
控制台打印：topic创建成功!
查看管理控制台如下：
2、发送消息（同步）
import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; public class SyncProducer { public static void main(String[] args) throws Exception {
// 使用生产者组名实例化一个生产者 DefaultMQProducer producer = new DefaultMQProducer("haoke");
// 指定nameserver地址 producer.setNamesrvAddr("114.xxx.xxx.xxx:9876");
// 启动生产者 producer.start(); //发送消息 String msg = "我的第一个消息6!";
// 创建message实例 Message message = new Message("my-topic", "delete", msg.getBytes("UTF-8"));
// 同步发送消息 SendResult sendResult = producer.send(message); System.out.println("消息id：" + sendResult.getMsgId()); System.out.println("消息队列：" + sendResult.getMessageQueue()); System.out.println("消息offset值：" + sendResult.getQueueOffset()); System.out.println(sendResult); producer.shutdown(); } }
Message是RocketMQ对消息的封装，我们也只能将消息封装为Message实例，才能通过RocketMQ发送出去。
控制台打印
消息id：C0A80A720BD818B4AAC2223955220000 消息队列：MessageQueue [topic=my-topic, brokerName=broker_haoke_im, queueId=6] 消息offset值：2 SendResult [sendStatus=SEND_OK, msgId=C0A80A720BD818B4AAC2223955220000, offsetMsgId=7273F69800002A9F000000000025ACB7, messageQueue=MessageQueue [topic=my-topic, brokerName=broker_haoke_im, queueId=6], queueOffset=2]
我们可以根据返回的消息ID查找消息。
queueId为队列的ID，前面我们设置了messageQueue的数量为8，
管理控制台：
点击message detail
Message数据结构
前面四个用得比较多。
3、发送消息（异步）
import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendCallback; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; public class AsyncProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("haoke"); producer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); producer.start(); // 发送消息 String msg = "我的第一个异步发送消息!"; Message message = new Message("my-topic", msg.getBytes("UTF-8"));
// 异步发送消息 producer.send(message, new SendCallback() { public void onSuccess(SendResult sendResult) { System.out.println("发送成功了!" + sendResult); System.out.println("消息id：" + sendResult.getMsgId()); System.out.println("消息队列：" + sendResult.getMessageQueue()); System.out.println("消息offset值：" + sendResult.getQueueOffset()); } public void onException(Throwable e) { System.out.println("消息发送失败!" + e); } }); // producer.shutdown(); } }
注意：由于是异步发送消息，故不能在发送消息之后立即shutdown，否则会导致消息发送不出去。
控制台打印如下：
发送成功了!SendResult [sendStatus=SEND_OK, msgId=C0A80A723C2818B4AAC22243949E0000, offsetMsgId=7273F69800002A9F000000000025AD7D, messageQueue=MessageQueue [topic=my-topic, brokerName=broker_haoke_im, queueId=2], queueOffset=0] 消息id：C0A80A723C2818B4AAC22243949E0000 消息队列：MessageQueue [topic=my-topic, brokerName=broker_haoke_im, queueId=2] 消息offset值：0
管理控制台
点击message detail
注意： producer.shutdown()要注释掉，否则发送失败。原因是，异步发送，还未来得及发送就被关闭了。
此时my-topic的状态如下：
4、消费消息
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; import java.io.UnsupportedEncodingException; import java.util.List; public class ConsumerDemo { public static void main(String[] args) throws Exception { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("haoke-consumer"); consumer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); // 订阅消息，接收的是所有消息 consumer.subscribe("my-topic", "*"); // consumer.subscribe("my-topic", "add || update"); consumer.setMessageModel(MessageModel.CLUSTERING);
// 注册消息监听 consumer.registerMessageListener(new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { try { for (MessageExt msg : msgs) { System.out.println("消息:" + new String(msg.getBody(), "UTF-8")); } } catch (UnsupportedEncodingException e) { e.printStackTrace(); } System.out.println("接收到消息 -> " + msgs);
// 给服务端响应消息消费成功 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); // 启动消费者 consumer.start(); } }
注意：* 表示只要是my-topic中的消息，都会接收，
控制台打印如下：
消息:我的第一个异步发送消息! 消息:我的第一个消息6! 接收到消息 -> [MessageExt [queueId=2, storeSize=197, queueOffset=0, sysFlag=0, bornTimestamp=1680853254303, bornHost=/49.222.85.128:56472, storeTimestamp=1680853255056, storeHost=/114.115.xxx.xxx:10911, msgId=7273F69800002A9F000000000025AD7D, commitLogOffset=2469245, bodyCRC=46384065, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='my-topic', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, CONSUME_START_TIME=1680854004566, UNIQ_KEY=C0A80A723C2818B4AAC22243949E0000, CLUSTER=DefaultCluster}, body=[-26, -120, -111, -25, -102, -124, -25, -84, -84, -28, -72, -128, -28, -72, -86, -27, -68, -126, -26, -83, -91, -27, -113, -111, -23, -128, -127, -26, -74, -120, -26, -127, -81, 33], transactionId='null'}]] 接收到消息 -> [MessageExt [queueId=6, storeSize=198, queueOffset=2, sysFlag=0, bornTimestamp=1680852582691, bornHost=/49.222.85.128:55201, storeTimestamp=1680852583446, storeHost=/114.115.xxx.xxx:10911, msgId=7273F69800002A9F000000000025ACB7, commitLogOffset=2469047, bodyCRC=270612399, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='my-topic', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=3, CONSUME_START_TIME=1680854004566, UNIQ_KEY=C0A80A720BD818B4AAC2223955220000, CLUSTER=DefaultCluster, TAGS=delete}, body=[-26, -120, -111, -25, -102, -124, -25, -84, -84, -28, -72, -128, -28, -72, -86, -26, -74, -120, -26, -127, -81, 54, 33], transactionId='null'}]]
如果我们想选择性的接收消息，我们可以根据tag进行筛选过滤，
public class ConsumerDemo { public static void main(String[] args) throws Exception { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("haoke-consumer"); consumer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); // 订阅消息，接收的是所有消息 // consumer.subscribe("my-topic", "*"); consumer.subscribe("my-topic", "add || update"); consumer.setMessageModel(MessageModel.CLUSTERING); consumer.registerMessageListener(new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { try { for (MessageExt msg : msgs) { System.out.println("消息:" + new String(msg.getBody(), "UTF-8")); } } catch (UnsupportedEncodingException e) { e.printStackTrace(); } System.out.println("接收到消息 -> " + msgs); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); // 启动消费者 consumer.start(); } }
此时，如果消费者发送的消息的tag为add或者update，消费者都可以接收到
//发送消息 String msg = "我的第一个消息7!"; Message message = new Message("my-topic", "add", msg.getBytes("UTF-8"));
启动消费者，生产者发消息，此时消费者控制台打印如下：
消息:我的第一个消息7! 接收到消息 -> [MessageExt [queueId=4, storeSize=195, queueOffset=0, sysFlag=0, bornTimestamp=1680854472081, bornHost=/49.222.85.128:58779, storeTimestamp=1680854472831, storeHost=/114.115.xxx.xxx:10911, msgId=7273F69800002A9F000000000025AF05, commitLogOffset=2469637, bodyCRC=154797806, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='my-topic', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, CONSUME_START_TIME=1680854472143, UNIQ_KEY=C0A80A726FB018B4AAC2225629910000, CLUSTER=DefaultCluster, TAGS=add}, body=[-26, -120, -111, -25, -102, -124, -25, -84, -84, -28, -72, -128, -28, -72, -86, -26, -74, -120, -26, -127, -81, 55, 33], transactionId='null'}]]
我们一般通过tag设置操作类型。
除了* ，其它订阅方式：
//完整匹配 consumer.subscribe("haoke_im_topic", "SEND_MSG"); //或匹配 consumer.subscribe("haoke_im_topic", "SEND_MSG || SEND_MSG1");
5、消息过滤器
RocketMQ支持根据用户自定义属性进行过滤，过滤表达式类似于SQL的where，如：a> 5 AND b ='abc'
发送消息：
import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; public class SyncProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("haoke"); producer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); producer.start(); //发送消息 String msg = "这是一个用户的消息, id = 1003"; Message message = new Message("my-topic-filter", "delete", msg.getBytes("UTF-8")); message.putUserProperty("sex","女"); message.putUserProperty("age","20"); SendResult sendResult = producer.send(message); System.out.println("消息id：" + sendResult.getMsgId()); System.out.println("消息队列：" + sendResult.getMessageQueue()); System.out.println("消息offset值：" + sendResult.getQueueOffset()); System.out.println(sendResult); producer.shutdown(); } }
接收消息：
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.MessageSelector; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.common.message.MessageExt; import java.io.UnsupportedEncodingException; import java.util.List; public class ConsumerFilter { public static void main(String[] args) throws Exception { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("haoke-consumer"); consumer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); // 订阅消息,筛选消息 consumer.subscribe("my-topic-filter", MessageSelector.bySql("sex='女' AND age>=18")); consumer.registerMessageListener(new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { try { for (MessageExt msg : msgs) { System.out.println("消息:" + new String(msg.getBody(), "UTF-8")); } } catch (UnsupportedEncodingException e) { e.printStackTrace(); } System.out.println("接收到消息 -> " + msgs); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); // 启动消费者 consumer.start(); } }
启动消费者，生产者发送消息，结果发现并没有接收到消息。重启消费者，报错如下：
Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: CODE: 1 DESC: The broker does not support consumer to filter message by SQL92 For more information, please visit the url, http://rocketmq.apache.org/docs/faq/ at org.apache.rocketmq.client.impl.MQClientAPIImpl.checkClientInBroker(MQClientAPIImpl.java:2089) at org.apache.rocketmq.client.impl.factory.MQClientInstance.checkClientInBroker(MQClientInstance.java:432) at org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl.start(DefaultMQPushConsumerImpl.java:633) at org.apache.rocketmq.client.consumer.DefaultMQPushConsumer.start(DefaultMQPushConsumer.java:520) at com.zwh.filter.ConsumerFilter.main(ConsumerFilter.java:34)
原因是默认配置下，不支持自定义属性，需要设置开启：
#加入到broker的配置文件中 enablePropertyFilter=true
停止broker，添加如下配置，即：
brokerIP1=云服务器IP namesrvAddr=云服务器IP:9876 brokerName=broker_haoke_im enablePropertyFilter=true
启动broker
./mqbroker -c /root/rocketmq/rmqbroker/conf/broker.conf
发现启动消费者就不报错了。查看管理控制台的集群，点击config
生产者发送消息，此时消费者消费消息，结果如下：
消息:这是一个用户的消息, id = 1003 接收到消息 -> [MessageExt [queueId=0, storeSize=235, queueOffset=0, sysFlag=0, bornTimestamp=1680858450933, bornHost=/49.222.85.128:52007, storeTimestamp=1680858451678, storeHost=/114.115.xxx.xxx:10911, msgId=7273F69800002A9F000000000025B374, commitLogOffset=2470772, bodyCRC=1222781001, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='my-topic-filter', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, sex=女, CONSUME_START_TIME=1680858451001, UNIQ_KEY=C0A80A72652818B4AAC22292DFF40000, CLUSTER=DefaultCluster, TAGS=delete, age=20}, body=[-24, -65, -103, -26, -104, -81, -28, -72, -128, -28, -72, -86, -25, -108, -88, -26, -120, -73, -25, -102, -124, -26, -74, -120, -26, -127, -81, 44, 32, 105, 100, 32, 61, 32, 49, 48, 48, 51], transactionId='null'}]]
如果将生产者中的女改为男，再次发送消息，此时消费者不会接收到消息。
三、producer详解
1、顺序消息
在某些业务中，consumer在消费消息时，是需要按照生产者发送消息的顺序进行消费的，比如在电商系统中，订单的消息，会有创建订单、订单支付、订单完成，如果消息的顺序发生改变，那么这样的消息就没有意义了。
如何保证发出去的消息是有顺序的呢?生产者发送的消息都要进入同一个队列，这是对生产者的要求。而之前发送的消息不在同一个队列中。消费者要在同一个队列中消费消息。
生产者
import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper; public class OrderProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("HAOKE_ORDER_PRODUCER"); producer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); producer.start(); for (int i = 0; i < 100; i++) { int orderId = i % 10; // 模拟生成订单id，值为0，1，2，3，4，5，6，7，8，9 String msgStr = "order --> " + i +", id = "+ orderId; Message message = new Message("haoke_order_topic", "ORDER_MSG", msgStr.getBytes(RemotingHelper.DEFAULT_CHARSET)); SendResult sendResult = producer.send(message, (mqs, msg, arg) -> { Integer id = (Integer) arg;
// mqs为消息队列数量，一个topic中的消息队列数默认为4 int index = id % mqs.size(); // 值为0，1，2，3 return mqs.get(index); // 返回消息队列 }, orderId); System.out.println(sendResult); } producer.shutdown(); } }
send方法如下：
public SendResult send(Message msg, MessageQueueSelector selector, Object arg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.defaultMQProducerImpl.send(msg, selector, arg); }
第二个参数是MessageQueueSelector，双击MessageQueueSelector，它是一个接口，该接口返回消息队列，通过该接口来选择消息队列。
public interface MessageQueueSelector { MessageQueue select(List<MessageQueue> var1, Message var2, Object var3); }
注意：send方法的第三个参数arg会进入select方法的第三个参数var3，即var3的值等于arg。
消费者
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly; import org.apache.rocketmq.common.message.MessageExt; import java.io.UnsupportedEncodingException; import java.util.List; public class OrderConsumer { public static void main(String[] args) throws Exception { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("HAOKE_ORDER_CONSUMER"); consumer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); consumer.subscribe("haoke_order_topic", "*"); consumer.registerMessageListener(new MessageListenerOrderly() { @Override public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { for (MessageExt msg : msgs) { try { System.out.println(Thread.currentThread().getName() + " " + msg.getQueueId() + " " + new String(msg.getBody(),"UTF-8")); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } } // System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs); return ConsumeOrderlyStatus.SUCCESS; } }); consumer.start(); } }
注意：设置监听器的时候，用的是MessageListenerOrderly，进行有序监听，就会通过同一个队列接收消息。
启动消费者，生产者发送消息
SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B205CC0000, offsetMsgId=7273F69800002A9F000000000025B54A, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=0], queueOffset=0] SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B206120001, offsetMsgId=7273F69800002A9F000000000025B618, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=1], queueOffset=0] SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B206300002, offsetMsgId=7273F69800002A9F000000000025B6E6, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=2], queueOffset=0] SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B2064D0003, offsetMsgId=7273F69800002A9F000000000025B7B4, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=3], queueOffset=0] SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B206650004, offsetMsgId=7273F69800002A9F000000000025B882, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=0], queueOffset=1] SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B2067D0005, offsetMsgId=7273F69800002A9F000000000025B950, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=1], queueOffset=1] SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B206950006, offsetMsgId=7273F69800002A9F000000000025BA1E, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=2], queueOffset=1] SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B206AC0007, offsetMsgId=7273F69800002A9F000000000025BAEC, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=3], queueOffset=1] SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B206C40008, offsetMsgId=7273F69800002A9F000000000025BBBA, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=0], queueOffset=2] SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B206DB0009, offsetMsgId=7273F69800002A9F000000000025BC88, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=1], queueOffset=2] SendResult [sendStatus=SEND_OK, msgId=C0A80A72774418B4AAC222B206F3000A, offsetMsgId=7273F69800002A9F000000000025BD56, messageQueue=MessageQueue [topic=haoke_order_topic, brokerName=broker_haoke_im, queueId=0], queueOffset=3]
...
消费者打印如下：
ConsumeMessageThread_1 2 order --> 2, id = 2 ConsumeMessageThread_1 2 order --> 6, id = 6 ConsumeMessageThread_1 2 order --> 12, id = 2 ConsumeMessageThread_1 2 order --> 16, id = 6 ConsumeMessageThread_1 2 order --> 22, id = 2 ConsumeMessageThread_1 2 order --> 26, id = 6 ConsumeMessageThread_1 2 order --> 32, id = 2 ConsumeMessageThread_1 2 order --> 36, id = 6 ConsumeMessageThread_1 2 order --> 42, id = 2 ConsumeMessageThread_1 2 order --> 46, id = 6 ConsumeMessageThread_1 2 order --> 52, id = 2 ConsumeMessageThread_1 2 order --> 56, id = 6 ConsumeMessageThread_1 2 order --> 62, id = 2 ConsumeMessageThread_1 2 order --> 66, id = 6 ConsumeMessageThread_1 2 order --> 72, id = 2 ConsumeMessageThread_1 2 order --> 76, id = 6 ConsumeMessageThread_1 2 order --> 82, id = 2 ConsumeMessageThread_1 2 order --> 86, id = 6 ConsumeMessageThread_1 2 order --> 92, id = 2 ConsumeMessageThread_1 2 order --> 96, id = 6 ConsumeMessageThread_2 3 order --> 3, id = 3 ConsumeMessageThread_2 3 order --> 7, id = 7 ConsumeMessageThread_2 3 order --> 13, id = 3 ConsumeMessageThread_2 3 order --> 17, id = 7 ConsumeMessageThread_2 3 order --> 23, id = 3 ConsumeMessageThread_2 3 order --> 27, id = 7 ConsumeMessageThread_2 3 order --> 33, id = 3 ConsumeMessageThread_2 3 order --> 37, id = 7 ConsumeMessageThread_2 3 order --> 43, id = 3 ConsumeMessageThread_2 3 order --> 47, id = 7 ConsumeMessageThread_2 3 order --> 53, id = 3 ConsumeMessageThread_2 3 order --> 57, id = 7 ConsumeMessageThread_2 3 order --> 63, id = 3 ConsumeMessageThread_2 3 order --> 67, id = 7 ConsumeMessageThread_2 3 order --> 73, id = 3 ConsumeMessageThread_2 3 order --> 77, id = 7 ConsumeMessageThread_2 3 order --> 83, id = 3 ConsumeMessageThread_2 3 order --> 87, id = 7 ConsumeMessageThread_2 3 order --> 93, id = 3 ConsumeMessageThread_2 3 order --> 97, id = 7 ConsumeMessageThread_3 0 order --> 0, id = 0 ConsumeMessageThread_3 0 order --> 4, id = 4 ConsumeMessageThread_3 0 order --> 8, id = 8 ConsumeMessageThread_3 0 order --> 10, id = 0 ConsumeMessageThread_3 0 order --> 14, id = 4 ConsumeMessageThread_3 0 order --> 18, id = 8 ConsumeMessageThread_3 0 order --> 20, id = 0 ConsumeMessageThread_3 0 order --> 24, id = 4 ConsumeMessageThread_3 0 order --> 28, id = 8 ConsumeMessageThread_3 0 order --> 30, id = 0 ConsumeMessageThread_3 0 order --> 34, id = 4 ConsumeMessageThread_3 0 order --> 38, id = 8 ConsumeMessageThread_3 0 order --> 40, id = 0 ConsumeMessageThread_3 0 order --> 44, id = 4 ConsumeMessageThread_3 0 order --> 48, id = 8 ConsumeMessageThread_3 0 order --> 50, id = 0 ConsumeMessageThread_3 0 order --> 54, id = 4 ConsumeMessageThread_3 0 order --> 58, id = 8 ConsumeMessageThread_3 0 order --> 60, id = 0 ConsumeMessageThread_3 0 order --> 64, id = 4 ConsumeMessageThread_3 0 order --> 68, id = 8 ConsumeMessageThread_3 0 order --> 70, id = 0 ConsumeMessageThread_3 0 order --> 74, id = 4 ConsumeMessageThread_3 0 order --> 78, id = 8 ConsumeMessageThread_3 0 order --> 80, id = 0 ConsumeMessageThread_3 0 order --> 84, id = 4 ConsumeMessageThread_3 0 order --> 88, id = 8 ConsumeMessageThread_3 0 order --> 90, id = 0 ConsumeMessageThread_3 0 order --> 94, id = 4 ConsumeMessageThread_3 0 order --> 98, id = 8 ConsumeMessageThread_4 1 order --> 1, id = 1 ConsumeMessageThread_4 1 order --> 5, id = 5 ConsumeMessageThread_4 1 order --> 9, id = 9 ConsumeMessageThread_4 1 order --> 11, id = 1 ConsumeMessageThread_4 1 order --> 15, id = 5 ConsumeMessageThread_4 1 order --> 19, id = 9 ConsumeMessageThread_4 1 order --> 21, id = 1 ConsumeMessageThread_4 1 order --> 25, id = 5 ConsumeMessageThread_4 1 order --> 29, id = 9 ConsumeMessageThread_4 1 order --> 31, id = 1 ConsumeMessageThread_4 1 order --> 35, id = 5 ConsumeMessageThread_4 1 order --> 39, id = 9 ConsumeMessageThread_4 1 order --> 41, id = 1 ConsumeMessageThread_4 1 order --> 45, id = 5 ConsumeMessageThread_4 1 order --> 49, id = 9 ConsumeMessageThread_4 1 order --> 51, id = 1 ConsumeMessageThread_4 1 order --> 55, id = 5 ConsumeMessageThread_4 1 order --> 59, id = 9 ConsumeMessageThread_4 1 order --> 61, id = 1 ConsumeMessageThread_4 1 order --> 65, id = 5 ConsumeMessageThread_4 1 order --> 69, id = 9 ConsumeMessageThread_4 1 order --> 71, id = 1 ConsumeMessageThread_4 1 order --> 75, id = 5 ConsumeMessageThread_4 1 order --> 79, id = 9 ConsumeMessageThread_4 1 order --> 81, id = 1 ConsumeMessageThread_4 1 order --> 85, id = 5 ConsumeMessageThread_4 1 order --> 89, id = 9 ConsumeMessageThread_4 1 order --> 91, id = 1 ConsumeMessageThread_4 1 order --> 95, id = 5 ConsumeMessageThread_4 1 order --> 99, id = 9
发现同一个线程监听的是同一个队列，拿到同一个订单的消息。
相同订单id的消息会落到同一个queue中，一个消费者线程会顺序消费queue，从而实现顺序消费消息。
2、分布式事务消息解决分布式事务问题
回顾什么事务？聊什么是事务，最经典的例子就是转账操作，用户A转账给用户B1000元的过程如下：
（1）、用户A发起转账请求，用户A账户减去1000元
（2）、用户B的账户增加1000元
如果，用户A账户减去1000元后，出现了故障（如网络故障），那么需要将该操作回滚，用户A账户增加1000元。这就是事务。
分布式事务
随着项目越来越复杂，越来越服务化，就会导致系统间的事务问题，这个就是分布式事务问题。
分布式事务分类有这几种：
（1）、基于单个JVM，数据库分库分表了（跨多个数据库）。
（2）、基于多JVM，服务拆分了（不跨数据库）。
（3）、基于多JVM，服务拆分了 并且数据库分库分表了。
解决分布式事务问题的方案有很多，使用消息实现只是其中的一种。
原理
（1）、Half(Prepare) Message
指的是暂不能投递的消息，发送方已经将消息成功发送到了 MQ 服务端，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态（暂时不能给消费者），处于该种状态下的消息即半消息。
（2）、Message Status Check
由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，MQ 服务端通过扫描发现某条消息长期处于“半消息”时，需要主动向消息生产者询问该消息的最终状态（Commit 或是 Rollback），该过程即消息回查。
执行流程
（1）、发送方向 MQ 服务端发送消息。
（2）、MQ Server 将消息持久化成功之后，向发送方 ACK 确认消息已经发送成功，此时消息为半消息。
（3）、发送方开始执行本地事务逻辑。
（4）、发送方根据本地事务执行结果向 MQ Server 提交二次确认（Commit 或是 Rollback），MQ Server 收到Commit 状态则将半消息标记为可投递，订阅方最终将收到该消息；MQ Server 收到 Rollback 状态则删除半消息，订阅方将不会接受该消息。
（5）、在断网或者是应用重启的特殊情况下，上述步骤4提交的二次确认最终未到达 MQ Server，经过固定时间后MQ Server 将对该消息发起消息回查。
（6）、发送方收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
（7）、发送方根据检查得到的本地事务的最终状态再次提交二次确认，MQ Server 仍按照步骤4对半消息进行操作。
生产者
import org.apache.rocketmq.client.producer.TransactionMQProducer; import org.apache.rocketmq.common.message.Message; public class TransactionProducer { public static void main(String[] args) throws Exception { TransactionMQProducer producer = new TransactionMQProducer("transaction_producer"); producer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); // 设置事务监听器 producer.setTransactionListener(new TransactionListenerImpl()); producer.start(); // 发送消息 Message message = new Message("pay_topic", "用户A给用户B转账500元".getBytes("UTF-8")); producer.sendMessageInTransaction(message, null); Thread.sleep(999999); producer.shutdown(); } }
注意：发送消息使用的是TransactionMQProducer
本地事务处理
import org.apache.rocketmq.client.producer.LocalTransactionState; import org.apache.rocketmq.client.producer.TransactionListener; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageExt; import java.util.HashMap; import java.util.Map; public class TransactionListenerImpl implements TransactionListener { private static Map<String, LocalTransactionState> STATE_MAP = new HashMap<>(); /** * 执行具体的业务逻辑 * * @param msg 发送的消息对象 * @param arg * @return */ @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { try {
// 发送方执行本地事务逻辑 System.out.println("用户A账户减500元."); Thread.sleep(500); //模拟调用服务 // System.out.println(1/0); System.out.println("用户B账户加500元."); Thread.sleep(800); // 模拟调用服务 STATE_MAP.put(msg.getTransactionId(), LocalTransactionState.COMMIT_MESSAGE); // 记录事务状态 // 二次提交确认 // return LocalTransactionState.UNKNOW; return LocalTransactionState.COMMIT_MESSAGE; // MQ server收到Commit状态则将半消息标记为可投递，订阅方最终将收到该消息 } catch (Exception e) { e.printStackTrace(); } STATE_MAP.put(msg.getTransactionId(), LocalTransactionState.ROLLBACK_MESSAGE); // 记录事务状态 // 回滚 return LocalTransactionState.ROLLBACK_MESSAGE; // MQ server收到rollback状态则删除半消息，订阅方将不会收到该消息。 } /** * 消息回查 * * @param msg * @return */ @Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { System.out.println("状态回查 ---> " + msg.getTransactionId() +" " +STATE_MAP.get(msg.getTransactionId()) ); return STATE_MAP.get(msg.getTransactionId()); // 获取事务状态 } }
消费者
import java.io.UnsupportedEncodingException; import java.util.List; public class TransactionConsumer { public static void main(String[] args) throws Exception { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("HAOKE_CONSUMER"); consumer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); // 订阅topic，接收此Topic下的所有消息 consumer.subscribe("pay_topic", "*"); consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { for (MessageExt msg : msgs) { try { System.out.println(new String(msg.getBody(), "UTF-8")); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); consumer.start(); } }
启动消费者，然后让生产者发送消息，生产者控制台打印如下：
用户A账户减500元.
用户B账户加500元.
消费者控制台打印如下：
用户A给用户B转账500元
如果来制造一个异常，来模拟rollback的情况
import org.apache.rocketmq.client.producer.LocalTransactionState; import org.apache.rocketmq.client.producer.TransactionListener; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageExt; import java.util.HashMap; import java.util.Map; public class TransactionListenerImpl implements TransactionListener { private static Map<String, LocalTransactionState> STATE_MAP = new HashMap<>(); /** * 执行具体的业务逻辑 * * @param msg 发送的消息对象 * @param arg * @return */ @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { try { System.out.println("用户A账户减500元."); Thread.sleep(500); //模拟调用服务 System.out.println(1/0); System.out.println("用户B账户加500元."); Thread.sleep(800); STATE_MAP.put(msg.getTransactionId(), LocalTransactionState.COMMIT_MESSAGE); // 二次提交确认 // return LocalTransactionState.UNKNOW; return LocalTransactionState.COMMIT_MESSAGE; } catch (Exception e) { e.printStackTrace(); } STATE_MAP.put(msg.getTransactionId(), LocalTransactionState.ROLLBACK_MESSAGE); // 回滚 return LocalTransactionState.ROLLBACK_MESSAGE; } /** * 消息回查 * * @param msg * @return */ @Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { System.out.println("状态回查 ---> " + msg.getTransactionId() +" " +STATE_MAP.get(msg.getTransactionId()) ); return STATE_MAP.get(msg.getTransactionId()); } }
启动消费者，生产者再发送消息
生产者控制台打印如下：
用户A账户减500元. java.lang.ArithmeticException: / by zero at com.zwh.transaction.TransactionListenerImpl.executeLocalTransaction(TransactionListenerImpl.java:24) at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendMessageInTransaction(DefaultMQProducerImpl.java:1156) at org.apache.rocketmq.client.producer.TransactionMQProducer.sendMessageInTransaction(TransactionMQProducer.java:79) at com.zwh.transaction.TransactionProducer.main(TransactionProducer.java:18)
消费者无法接收到消息。
测试结果：返回commit状态时，消费者能够接收到消息，返回rollback状态时，消费者接受不到消息。
我们通过返回unknown状态来模拟回查
import org.apache.rocketmq.client.producer.LocalTransactionState; import org.apache.rocketmq.client.producer.TransactionListener; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageExt; import java.util.HashMap; import java.util.Map; public class TransactionListenerImpl implements TransactionListener { private static Map<String, LocalTransactionState> STATE_MAP = new HashMap<>(); /** * 执行具体的业务逻辑 * * @param msg 发送的消息对象 * @param arg * @return */ @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { try { System.out.println("用户A账户减500元."); Thread.sleep(500); //模拟调用服务 // System.out.println(1/0); System.out.println("用户B账户加500元."); Thread.sleep(800); STATE_MAP.put(msg.getTransactionId(), LocalTransactionState.COMMIT_MESSAGE); // 二次提交确认 return LocalTransactionState.UNKNOW; // return LocalTransactionState.COMMIT_MESSAGE; } catch (Exception e) { e.printStackTrace(); } STATE_MAP.put(msg.getTransactionId(), LocalTransactionState.ROLLBACK_MESSAGE); // 回滚 return LocalTransactionState.ROLLBACK_MESSAGE; } /** * 消息回查 * * @param msg * @return */ @Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { System.out.println("状态回查 ---> " + msg.getTransactionId() +" " +STATE_MAP.get(msg.getTransactionId()) ); return STATE_MAP.get(msg.getTransactionId()); } }
启动消费者，然后生产者发送消息
生产者控制台打印如下：
用户A账户减500元.
用户B账户加500元.
状态回查 ---> C0A81FA74B9818B4AAC223F21B4F0000 COMMIT_MESSAGE
消费者控制台打印如下：
用户A给用户B转账500元
四、Consumer详解
push和pull模式
在RocketMQ中，消费者有两种模式，一种是push模式，另一种是pull模式。
push模式：客户端与服务端建立连接后，当服务端有消息时，将消息推送到客户端。
pull模式：客户端不断的轮询请求服务端，来获取新的消息。
但在具体实现时，Push和Pull模式都是采用消费端主动拉取的方式，即consumer轮询从broker拉取消息。编码时很少使用pull，使用push比较多。
区别：
Push方式里，consumer把轮询过程封装了，并注册MessageListener监听器，取到消息后，唤醒MessageListener的consumeMessage()来消费，对用户而言，感觉消息是被推送过来的。
Pull方式里，取消息的过程需要用户自己写，首先通过打算消费的Topic拿到MessageQueue的集合，遍历MessageQueue集合，然后针对每个MessageQueue批量取消息，一次取完后，记录该队列下一次要取的开始offset，直到取完了，再换另一个MessageQueue。
疑问：既然是采用pull方式实现，RocketMQ如何保证消息的实时性呢？
长轮询
RocketMQ中采用了长轮询的方式实现（消息的实时性），什么是长轮询呢？
长轮询即是在请求的过程中，若是服务器端数据并没有更新，那么则将这个连接挂起，直到服务器推送新的数据，再返回，然后进入循环周期。避免了频繁的轮询。
客户端像传统轮询一样从服务端请求数据，服务端会阻塞请求不会立刻返回，直到有数据或超时才返回给客户端，然后关闭连接，客户端处理完响应信息后再向服务器发送新的请求。
消息模式
DefaultMQPushConsumer实现了自动保存offset值以及实现多个consumer的负载均衡。
//设置组名 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("HAOKE_IM");
通过groupname将多个consumer组合在一起，那么就会存在一个问题，消息发送到这个组后，消息怎么分配呢？
这个时候，就需要指定消息模式，分别有集群和广播模式。
集群模式（默认是集群模式）
同一个 ConsumerGroup(GroupName相同) 里的每 个 Consumer 只消费所订阅消息的一部分内容， 同一个 ConsumerGroup 里所有的 Consumer消费的内容合起来才是所订阅 Topic 内容的整体， 从而达到负载均衡的目的 。（这样当消息比较多时能被快速消费掉）
广播模式
同一个 ConsumerGroup里的每个 Consumer都 能消费到所订阅 Topic 的全部消息，也就是一个消息会被多次分发，被多个 Consumer消费。
// 集群模式 consumer.setMessageModel(MessageModel.CLUSTERING); // 广播模式 consumer.setMessageModel(MessageModel.BROADCASTING);
重复消息的解决方案
造成消息重复的根本原因是：网络不可达。只要通过网络交换数据，就无法避免这个问题。所以解决这个问题的办法就是绕过这个问题。那么问题就变成了：
如果消费端收到两条一样的消息，应该怎样处理？
（1）、消费端处理消息的业务逻辑保持幂等性（无论执行多少次结果是一样的）
（2）、保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现
第1条很好理解，只要保持幂等性，不管来多少条重复消息，最后处理的结果都一样。第2条原理就是利用一张日志表来记录已经处理成功的消息的ID，如果新到的消息ID已经在日志表中，那么就不再处理这条消息。
第1条解决方案，很明显应该在消费端实现，不属于消息系统要实现的功能。第2条可以消息系统实现，也可以业务端实现。正常情况下出现重复消息的概率其实很小，如果由消息系统来实现的话，肯定会对消息系统的吞吐量和高可用有影响，所以最好还是由业务端自己处理消息重复的问题，这也是RocketMQ不解决消息重复的问题的原因。
RocketMQ不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重。
五、RocketMQ存储
RocketMQ中的消息数据存储，采用了零拷贝技术（使用 mmap + write 方式），文件系统采用 Linux Ext4 文件系统进行存储。
消息数据的存储
在RocketMQ中，消息数据是保存在磁盘文件中，为了保证写入的性能，RocketMQ尽可能保证顺序写入，顺序写入的效率比随机写入的效率高很多。
RocketMQ消息的存储是由ConsumeQueue和CommitLog配合完成的，CommitLog是真正存储数据的文件，ConsumeQueue是索引文件，存储数据指向到物理文件的配置。
我们进入/root/store中
[root@hecs-xxx ~]# cd store [root@hecs-xxx store]# ll total 24 -rw-r--r-- 1 root root 0 Apr 8 07:00 abort -rw-r--r-- 1 root root 4096 Apr 8 07:52 checkpoint drwxr-xr-x 2 root root 4096 Apr 3 11:23 commitlog drwxr-xr-x 2 root root 4096 Apr 8 07:53 config drwxr-xr-x 9 root root 4096 Apr 7 23:24 consumequeue drwxr-xr-x 2 root root 4096 Apr 5 20:12 index -rw-r--r-- 1 root root 4 Apr 8 07:00 lock
进入consumequeue目录
[root@hecs-xxx store]# cd consumequeue [root@hecs-xxx consumequeue]# ll total 28 drwxr-xr-x 6 root root 4096 Apr 7 17:41 haoke_order_topic drwxr-xr-x 9 root root 4096 Apr 7 16:01 my-topic drwxr-xr-x 5 root root 4096 Apr 7 17:07 my-topic-filter drwxr-xr-x 3 root root 4096 Apr 7 23:24 pay_topic drwxr-xr-x 3 root root 4096 Apr 7 23:23 RMQ_SYS_TRANS_HALF_TOPIC drwxr-xr-x 3 root root 4096 Apr 7 23:24 RMQ_SYS_TRANS_OP_HALF_TOPIC drwxr-xr-x 10 root root 4096 Apr 4 10:18 TopicTest
进入my-topic
[root@hecs-xxx consumequeue]# cd my-topic [root@hecs-xxx my-topic]# ll total 28 drwxr-xr-x 2 root root 4096 Apr 7 16:00 0 drwxr-xr-x 2 root root 4096 Apr 6 08:43 1 drwxr-xr-x 2 root root 4096 Apr 7 15:40 2 drwxr-xr-x 2 root root 4096 Apr 6 08:49 3 drwxr-xr-x 2 root root 4096 Apr 7 16:01 4 drwxr-xr-x 2 root root 4096 Apr 5 20:31 6 drwxr-xr-x 2 root root 4096 Apr 5 20:30 7
发现有8个队列
如上图所示：
（1）、消息主体以及元数据都存储在CommitLog当中。
（2）、Consume Queue相当于kafka中的partition，是一个逻辑队列，存储了这个Queue在CommiLog中的起始offset，log大小和MessageTag的hashCode。
（3）、每次读取消息队列先读取consumerQueue,然后再通过consumerQueue去commitLog中拿到消息主体。
文件位置：
同步刷盘与异步刷盘
RocketMQ 为了提高性能，会尽可能地保证 磁盘的顺序写。消息在通过 Producer 写入 RocketMQ 的时候，有两种写磁盘方式，分别是同步刷盘与异步刷盘。
1、同步刷盘
在返回写成功状态时，消息已经被写入磁盘 。
具体流程是：消息写入内存的 PAGECACHE 后，立刻通知刷盘线程刷盘，然后等待刷盘完成，刷盘线程执行完成后唤醒等待的线程，返回消息写成功的状态 。
优点：保证数据不丢失。缺点：速度要慢一些，因为要等待刷盘完成再返回消息写成功的状态。
2、异步刷盘
在返回写成功状态时，消息可能只是被写入了内存的 PAGECACHE，写操作的返回快，吞吐量大。
当内存里的消息量积累到一定程度时，统一触发写磁盘动作，快速写入。
缺点：有可能丢失数据，当我们把数据写入到内存当中的时候，如果突然断电，会导致数据的丢失。
broker配置文件中指定刷盘方式“
flushDiskType=ASYNC_FLUSH -- 异步
flushDiskType=SYNC_FLUSH -- 同步
如果配置文件没有指定，则默认为异步刷盘
如何选择呢？
当我们的场景对数据可靠性要求比较高的时候，选择同步刷盘，宁愿牺牲速度来换取可靠性。当我们的场景是需要快速响应，有些时候丢失一些数据也无所谓的时候，比如说传输日志数据，此时可以选择异步刷盘的方式。
六、重试策略
在消息的发送和消费过程中，都有可能出现错误，如网络异常等，出现了错误就需要进行错误重试，这种消息的重试需要分2种，分别是producer端重试和consumer端重试。
1、producer端重试
生产者端的消息失败，也就是Producer往MQ上发消息没有发送成功，比如网络抖动导致生产者发送消息到MQ失败。
import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper; public class SyncProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("HAOKE_IM"); producer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); //消息发送失败时，重试3次 producer.setRetryTimesWhenSendFailed(3); producer.start(); String msgStr = "用户A发送消息给用户B"; Message msg = new Message("haoke_im_topic", "SEND_MSG", msgStr.getBytes(RemotingHelper.DEFAULT_CHARSET)); // 发送消息,并且指定超时时间 SendResult sendResult = producer.send(msg, 1000); System.out.println("消息状态：" + sendResult.getSendStatus()); System.out.println("消息id：" + sendResult.getMsgId()); System.out.println("消息queue：" + sendResult.getMessageQueue()); System.out.println("消息offset：" + sendResult.getQueueOffset()); System.out.println(sendResult); producer.shutdown(); } }
生产者要做两件事：1、设置重试次数，2、发消息时指定超时时间
由于不太好演示，我们通过源码的方式来讲解：
// 发送消息,并且指定超时时间 SendResult sendResult = producer.send(msg, 1000);
点击send方法：
public SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.defaultMQProducerImpl.send(msg, timeout); }
再点击send方法
public SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.sendDefaultImpl(msg, CommunicationMode.SYNC, (SendCallback)null, timeout); }
再点击sendDefaultImpl
private SendResult sendDefaultImpl(Message msg, CommunicationMode communicationMode, SendCallback sendCallback, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { this.makeSureStateOK(); Validators.checkMessage(msg, this.defaultMQProducer); long invokeID = this.random.nextLong(); long beginTimestampFirst = System.currentTimeMillis(); long beginTimestampPrev = beginTimestampFirst; TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic()); if (topicPublishInfo != null && topicPublishInfo.ok()) { boolean callTimeout = false; MessageQueue mq = null; Exception exception = null; SendResult sendResult = null; int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1; int times = 0; String[] brokersSent = new String[timesTotal]; while(true) { label122: { String info; if (times < timesTotal) { // 如果times小于总次数，就进行重试 info = null == mq ? null : mq.getBrokerName(); MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, info); if (mqSelected != null) { mq = mqSelected; brokersSent[times] = mqSelected.getBrokerName(); long endTimestamp; try { beginTimestampPrev = System.currentTimeMillis(); long costTime = beginTimestampPrev - beginTimestampFirst; if (timeout >= costTime) { sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime); endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false); switch(communicationMode) { case ASYNC: return null; case ONEWAY: return null; case SYNC: if (sendResult.getSendStatus() == SendStatus.SEND_OK || !this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) { return sendResult; } default: break label122; } } callTimeout = true; } catch (RemotingException var26) { // 出现RemotingException就重试 endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true); this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var26); this.log.warn(msg.toString()); exception = var26; break label122; } catch (MQClientException var27) { // 出现MQClientException就重试 endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true); this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var27); this.log.warn(msg.toString()); exception = var27; break label122; } catch (MQBrokerException var28) { // 出现MQBrokerException有可能重试，也有可能抛异常 endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true); this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var28); this.log.warn(msg.toString()); exception = var28; switch(var28.getResponseCode()) { case 1: case 14: case 16: case 17: case 204: case 205: break label122; default: if (sendResult != null) { return sendResult; } throw var28; } } catch (InterruptedException var29) { // 不重试抛异常 endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, false); this.log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var29); this.log.warn(msg.toString()); this.log.warn("sendKernelImpl exception", var29); this.log.warn(msg.toString()); throw var29; } } } if (sendResult != null) { return sendResult; } info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s", times, System.currentTimeMillis() - beginTimestampFirst, msg.getTopic(), Arrays.toString(brokersSent)); info = info + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"); MQClientException mqClientException = new MQClientException(info, (Throwable)exception); if (callTimeout) { throw new RemotingTooMuchRequestException("sendDefaultImpl call timeout"); } if (exception instanceof MQBrokerException) { mqClientException.setResponseCode(((MQBrokerException)exception).getResponseCode()); } else if (exception instanceof RemotingConnectException) { mqClientException.setResponseCode(10001); } else if (exception instanceof RemotingTimeoutException) { mqClientException.setResponseCode(10002); } else if (exception instanceof MQClientException) { mqClientException.setResponseCode(10003); } throw mqClientException; } ++times; } } else { List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList(); if (null != nsList && !nsList.isEmpty()) { throw (new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"), (Throwable)null)).setResponseCode(10005); } else { throw (new MQClientException("No name server address, please set it." + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"), (Throwable)null)).setResponseCode(10004); } } }
timeTotal表示总次数，如果时同步模式，重试次数为1加重试次数，如果是异步模式，则不会重试。
2、consumer端重试
消费者端的失败，分为2种情况，一个是exception，一个是timeout。
（1）、exception
消息正常的到了消费者，结果消费者发生异常，处理失败了。例如反序列化失败，消息数据本身无法处理（例如话费充值，当前消息的手机号被注销，无法充值）等。
返回的消息的状态：
package org.apache.rocketmq.client.consumer.listener; public enum ConsumeConcurrentlyStatus { /** * Success consumption */ CONSUME_SUCCESS, /** * Failure consumption,later try to consume */ RECONSUME_LATER; }
可以看到，消息的状态分为成功或者失败。如果返回的状态为失败会怎么样呢？
在启动broker的日志中可以看到这样的信息：
INFO main - messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
这个表示了，如果消息消费失败，那么消息将会在1s、5s、10s后重试，一直到2h后不再重试。
其实，有些时候并不需要重试这么多次，一般重试3~5次即可。这个时候就可以通过msg.getReconsumeTimes()获取重试次数进行控制。
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.common.message.MessageExt; import java.io.UnsupportedEncodingException; import java.util.List; public class ConsumerDemo { public static void main(String[] args) throws Exception { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("HAOKE_IM"); consumer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); // 订阅topic，接收此Topic下的所有消息 consumer.subscribe("my-test-topic", "*"); consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { for (MessageExt msg : msgs) { try { System.out.println(new String(msg.getBody(), "UTF-8")); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } } System.out.println("收到消息->" + msgs); if(msgs.get(0).getReconsumeTimes() >= 3){ // 重试3次后，不再进行重试 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } return ConsumeConcurrentlyStatus.RECONSUME_LATER; } }); consumer.start(); } }
生产者：
import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; public class SyncProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("haoke"); producer.setNamesrvAddr("114.xxx.xxx.xxx:9876"); producer.start(); //发送消息 String msg = "我的第一个消息7!"; Message message = new Message("my-test-topic", "add", msg.getBytes("UTF-8")); SendResult sendResult = producer.send(message); System.out.println("消息id：" + sendResult.getMsgId()); System.out.println("消息队列：" + sendResult.getMessageQueue()); System.out.println("消息offset值：" + sendResult.getQueueOffset()); System.out.println(sendResult); producer.shutdown(); } }
先启动消费者，生产者再发一条消息：
生产者控制台打印如下：
消息id：C0A81FA7135C18B4AAC226765B1D0000 消息队列：MessageQueue [topic=my-test-topic, brokerName=broker_haoke_im, queueId=0] 消息offset值：0 SendResult [sendStatus=SEND_OK, msgId=C0A81FA7135C18B4AAC226765B1D0000, offsetMsgId=7273F69800002A9F0000000000266045, messageQueue=MessageQueue [topic=my-test-topic, brokerName=broker_haoke_im, queueId=0], queueOffset=0]
消费者控制台如下：
我的第一个消息7! 收到消息->[MessageExt [queueId=0, storeSize=200, queueOffset=0, sysFlag=0, bornTimestamp=1680923690782, bornHost=/171.43.251.60:17995, storeTimestamp=1680923691803, storeHost=/114.xxx.xxx.xxx:10911, msgId=7273F69800002A9F0000000000266045, commitLogOffset=2515013, bodyCRC=154797806, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='my-test-topic', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, CONSUME_START_TIME=1680923723807, UNIQ_KEY=C0A81FA7135C18B4AAC226765B1D0000, CLUSTER=DefaultCluster, TAGS=add}, body=[-26, -120, -111, -25, -102, -124, -25, -84, -84, -28, -72, -128, -28, -72, -86, -26, -74, -120, -26, -127, -81, 55, 33], transactionId='null'}]] 我的第一个消息7! 收到消息->[MessageExt [queueId=0, storeSize=336, queueOffset=0, sysFlag=0, bornTimestamp=1680923690782, bornHost=/171.43.251.60:17995, storeTimestamp=1680923735236, storeHost=/114.xxx.xxx.xxx:10911, msgId=7273F69800002A9F0000000000266261, commitLogOffset=2515553, bodyCRC=154797806, reconsumeTimes=1, preparedTransactionOffset=0, toString()=Message{topic='my-test-topic', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=%RETRY%HAOKE_IM, ORIGIN_MESSAGE_ID=7273F69800002A9F0000000000266045, RETRY_TOPIC=my-test-topic, MAX_OFFSET=1, CONSUME_START_TIME=1680923734356, UNIQ_KEY=C0A81FA7135C18B4AAC226765B1D0000, CLUSTER=DefaultCluster, WAIT=false, DELAY=3, TAGS=add, REAL_QID=0}, body=[-26, -120, -111, -25, -102, -124, -25, -84, -84, -28, -72, -128, -28, -72, -86, -26, -74, -120, -26, -127, -81, 55, 33], transactionId='null'}]] 我的第一个消息7! 收到消息->[MessageExt [queueId=0, storeSize=336, queueOffset=1, sysFlag=0, bornTimestamp=1680923690782, bornHost=/171.43.251.60:17995, storeTimestamp=1680923765363, storeHost=/114.xxx.xxx.xxx:10911, msgId=7273F69800002A9F0000000000266505, commitLogOffset=2516229, bodyCRC=154797806, reconsumeTimes=2, preparedTransactionOffset=0, toString()=Message{topic='my-test-topic', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=%RETRY%HAOKE_IM, ORIGIN_MESSAGE_ID=7273F69800002A9F0000000000266045, RETRY_TOPIC=my-test-topic, MAX_OFFSET=2, CONSUME_START_TIME=1680923764425, UNIQ_KEY=C0A81FA7135C18B4AAC226765B1D0000, CLUSTER=DefaultCluster, WAIT=false, DELAY=4, TAGS=add, REAL_QID=0}, body=[-26, -120, -111, -25, -102, -124, -25, -84, -84, -28, -72, -128, -28, -72, -86, -26, -74, -120, -26, -127, -81, 55, 33], transactionId='null'}]] 我的第一个消息7! 收到消息->[MessageExt [queueId=0, storeSize=336, queueOffset=2, sysFlag=0, bornTimestamp=1680923690782, bornHost=/171.43.251.60:17995, storeTimestamp=1680923825416, storeHost=/114.xxx.xxx.xxx:10911, msgId=7273F69800002A9F00000000002667A9, commitLogOffset=2516905, bodyCRC=154797806, reconsumeTimes=3, preparedTransactionOffset=0, toString()=Message{topic='my-test-topic', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=%RETRY%HAOKE_IM, ORIGIN_MESSAGE_ID=7273F69800002A9F0000000000266045, RETRY_TOPIC=my-test-topic, MAX_OFFSET=3, CONSUME_START_TIME=1680923824478, UNIQ_KEY=C0A81FA7135C18B4AAC226765B1D0000, CLUSTER=DefaultCluster, WAIT=false, DELAY=5, TAGS=add, REAL_QID=0}, body=[-26, -120, -111, -25, -102, -124, -25, -84, -84, -28, -72, -128, -28, -72, -86, -26, -74, -120, -26, -127, -81, 55, 33], transactionId='null'}]]
第一条消息是生产者发出的，有msgId可知，后面发出的消息与第一条消息内容相同，但是不同的消息。
（2）、timeout
比如由于网络原因导致消息压根就没有从MQ到消费者上，那么在RocketMQ内部会不断的尝试发送这条消息，直至发送成功为止！
也就是说，服务端没有接收到消息的反馈，既不是成功也不是失败，这个时候定义为超时。