rocketMQ系列1

1.消息队列：先进先出的数据结构
2.应用解耦：
3.流量削峰：

应用系统如果遇到系统流量的瞬间猛增，有可能会将系统压垮，有了消息队列可以将大量请求缓存起来，分散到很长一段时间处理，这样可以大大提高系统的稳定性和用户体验。
用户----qps5k个请求----->MQ-----A系统从MQ中每秒拉取2K个请求--->A系统-->MySQL
业务系统正常时段的QPS如果是1000，流量最高峰是10000，为了应对流量高峰配置高性能的服务器显
然不划算，这时可以使用消息队列峰值流量削峰

数据分发：消息订阅

1.通过消息队列可以让数据再多个系统之间进行沟通，数据的产生方不需要关心谁来使用数据，只需要将
数据发送到消息队列，数据使用方直接再消息队列中直接获取数据即可。

MQ的优点和缺点
缺点：

1.系统可用性降低
系统引入的外部依赖越多，系统稳定性越差，一旦MQ宕机，就会对业务造成影响。
如何保证MQ的高可用？高可用
2.系统复杂度提高
MQ的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过MQ进行异步调用。
如何保证消息没有被重复消费？处理消息丢失情况？那么保证消息传递的顺序性？
3.一致性问题
A系统处理完业务，通过MQ给B、C、D三个系统发消息数据，如果B系统，C系统处理成功，D系统处理失败。
如果保证消息数据处理的一致性？

下载网址：

https://www.apache.org/dyn/closer.cgi?path=rocketmq-all-4.5.1-bin-release.zip
解压|

启动RocketMQ

1.启动Nameserver：进入到bin目录
nohub sh bin/mqnamesrv &
2.查看启动日志：
tail -f  ~/logs/rocketmqlogs/namesrv.log

启动Broker:

1.启动Broker
nohub sh bin/mqbroker -n localhost:9876 &
2.查看启动日志：
tail -f  ~/logs/rocketmqlogs/broker.log

问题描述

RocketMQ默认的虚拟机内存较大，启动Broker如果因为内存不足失败，需要编辑如下两个配置文件，修改JVM内存大小：
#编辑runbroker.sh和runserver.sh修改默认JVM大小bin目录下
vi runbroker.sh
vi runserver.sh
两个都改成如下设置
内存1G的情况下
参数设置：JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:Metaspace=128m -XX:MaxMetaspaceSize=320m"
1.关闭NameServer
sh bin/mqshutdown namesrc
2.关闭Broker
sh /bin/mqshutdoen broker

发送消息：

#1.设置环境变量：
export NAMESRV_ADDR=localhost:9876
#2.使用安装包的Demo发送消息
sh (bin/)tools.sh org.apache.rocketmq.example.quickstart.Producer

接受消息：

#1.设置环境变量：
export NAMESRV_ADDR=localhost:9876
#2.使用安装包的Demo发送消息bin目录下
sh (bin/)tools.sh org.apache.rocketmq.example.quickstart.Consumer

关闭RocketMQ:

#1.关闭NameServer
sh bin/mqshutdown namesrv
#2.关闭Broker
sh bin/mqshutdown broker

RocketMQ集群搭建：
各角色介绍：

1.Producer:消息的发送者
2.Consumer:消息的接收者
3.Broker:暂存和消息传输
4.NameServer:管理Broker;
5.Topic:区分消息的种类；一个发送者可以发送消息给一个或多个Topic;一个消息的接收者可以订阅
一个或者多个Topic消息
6.Message Queue:相当于是Topic的分区;用于并行发送和接收消息;
过程：Producer连接NameServer 发送消息到Broker NameServer 用于管理Broker，Consumer连接NameServer集群找到需要的Broker;

集群搭建方式：
集群特点：

1.NameServer是一个无状态节点，可集群部署，节点之间无任何信息同步
2.Broker部署相对复杂，Broker分为Master与Slave,一个Master可以对应多个Slave,但是一个Slave
只能对应一个Master,Master与Slave的对应关系通过指定相同的BrokerName,不同的BrokerId来定
义，BrokerId为0表示Master,非0表示Slave.Master也可以部署多个。每个Broker与NameServer集
群中的所有节点建立长连接，定时注册Topic信息所有NameServer
3.Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提高Topic服务的Master建立长连接，且定时向Master发送心跳，Producer完全无状态，可集群部署。
4.Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取
Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master,Slave发送心
跳。Consumer既可以从Master订阅信息，也可以从Slave订阅信息，订阅规则由Broker配置决定；

集群模式：

1.单Master模式
比较危险，适用本地测试；
2.多Master模式
一个集群无Slave,全是Master,例如2个Master或者3个Master,这种模式的优缺点如下:
优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10,即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量信息，同步刷盘一条不丢），性能最高；
缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。
3.多Master多Slave模式（异步） 实际项目采用
每个Master配置一个Slave,有多对Master-Slave,有多对Master-Slave，HA采用异步复制模式，主备有短暂消息延迟，这种模式的优缺点如下：
优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，同时
Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多
Master模式几乎一样
缺点：Master宕机后，磁盘损坏情况下，会丢失少量消息
4.多Master多Slave模式（同步）
每个Master配置一个Slave,有多对Master-Slave,有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功。这种模式的优缺点如下：
优点：数据与服务都无单点障碍，M宕机情况下，消息无延迟。服务可用性与数据可用性都非常高
缺点：性能比异步复制略低（大约10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机
后，备机不能自动切换主机；

双主双从集群搭建：

工作流程：

1.启动NameServer,NameServer起来后监听端口，等待Broker,Producer,Consumer连上来，相当于一个路由控制中心；
2.Broker启动，跟所有的NameServer保持长连接，定时发送心跳包；心跳包中包含当前Broker信息
（IP+接口）以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
3.收发消息前，先创建Topic,创建Topic时需要指定该Topic要存储在哪些Broker上，也可以发送消息时自动创建Topic
4.Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获
取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker
建立长连接从而向Broker发消息。
5.Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些
Broker上，然后直接跟Broker建立连接通道，开始消费消息。

集群搭建：
服务器：

    ip                     角色                          架构模式
1.192.168.25.135     nameServer,brokerserver          Master1,Slave2
2.192.168.25.138     nameserver,brokerserver          Master2,Slave1

host添加信息：

vim /etc/hosts
#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

配置完成后，重启网卡

systemctl restart network

防火墙配置：

宿主机需要远程访问虚拟机的rocketmq服务和web服务，需要开放相关的端口号，简单粗爆的方式是直接关闭防火墙
#关闭防火墙
systemctl stop firewall.service
#查看防火墙状态
firewall-cmd --state
#禁止firewall开启启动
systemctl disable firewalld.service
或者为了安全，只开放特定端口号 线上采用
nameserver 默认9876
master 默认10911
slave  默认1101

环境变量配置：

vim /etc/profile
#set rocketmq
ROCKETMQ_HOME=/usr/local/rocketmq/rocketmq-all-4.4.0-bin-release #指向安装路劲
PATH=$PATH:$ROCKETMQ_HOME/bin
export ROCKETMQ_HOME PATH
输入wq!保存并退出，并使得配置立刻生效:
source /etc/profile

创建消息存储路劲

mkdir /usr/local/rocketmq/store
mkdir /usr/local/rocketmq/store/Commitlog
mkdir /usr/local/rocketmq/store/consumequeue
mkdir /usr/local/rocketmq/store/index

broker配置文件
服务器：192.168.25.135

1.vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a.properties
2.修改配置如下
#所属集群名字：
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自动创建订阅组，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许Broker自动创建订阅组,建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨4点
deleteWhen=04
#文件保留时间,默认48小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ComsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFilesSizeConsumeQueue=300000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路劲
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路劲
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路劲
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路劲
storePathConsumeQueue=/usr/local/rocketmq/store/index
#checkpoint 文件存储路劲
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路劲
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#Broker 的角色
#-ASYNC_MASTER 异步复制Master
#-SYNC_MASTER 同步双写Master
#-SLAVE
brokerRole=SYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH

服务器:192.168.25.135
slave2

1.vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b-s.properties
2.修改配置如下
#所属集群名字：
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自动创建订阅组，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许Broker自动创建订阅组,建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=11011
#删除文件时间点，默认凌晨4点
deleteWhen=04
#文件保留时间,默认48小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ComsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFilesSizeConsumeQueue=300000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路劲
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路劲
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路劲
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储存储路劲
storeCheckpoint=/usr/local/rocketmq/store/index
#abort 文件存储路劲
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#Broker 的角色
#-ASYNC_MASTER 异步复制Master
#-SYNC_MASTER 同步双写Master
#-SLAVE
brokerRole=SLAVE
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#发消息线程数量
#checkTransactionMessageEnable=false
#sendMessageThreadPoolNums=128
#拉消息线程数量
#pullMessageThreadPoolNums=128

服务器:192.168.25.138 master2
配置：broker-b.properties

2.修改配置如下
#所属集群名字：
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自动创建订阅组，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许Broker自动创建订阅组,建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨4点
deleteWhen=04
#文件保留时间,默认48小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ComsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFilesSizeConsumeQueue=300000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路劲
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路劲
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路劲
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路劲
storePathConsumeQueue=/usr/local/rocketmq/store/index
#checkpoint 文件存储路劲
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路劲
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#Broker 的角色
#-ASYNC_MASTER 异步复制Master
#-SYNC_MASTER 同步双写Master
#-SLAVE
brokerRole=SYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH

服务器:192.168.25.138 slave1
配置：broker-a-s.properties

2.修改配置如下
#所属集群名字：
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自动创建订阅组，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许Broker自动创建订阅组,建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=11011
#删除文件时间点，默认凌晨4点
deleteWhen=04
#文件保留时间,默认48小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ComsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFilesSizeConsumeQueue=300000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路劲
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路劲
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路劲
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储存储路劲
storeCheckpoint=/usr/local/rocketmq/store/index
#abort 文件存储路劲
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#Broker 的角色
#-ASYNC_MASTER 异步复制Master
#-SYNC_MASTER 同步双写Master
#-SLAVE
brokerRole=SLAVE
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH

修改启动脚本：

1.runbroker.sh
vim /usr/local/rocketmq/bin/runbroker.sh
需要根据内存大小进行适当的对JVM参数进行调整
#开发环境配置 JVM Configuration
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"
2.runserver.sh
vim /usr/local/rocketmq/bin/runserver.sh
#开发环境配置 JVM Configuration
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m
-XX:MaxMetaspace=320m"

服务启动
1.启动NameServe集群
分别在192.168.25.135和192.168.25.138启动NameServer

1.cd /usr/local/rocketmq/bin
2.nohup sh mqnamesrv &

2.启动Broker集群 -c 指定配置文件
在192.168.25.135启动master1和slave2
master1:

1.cd /usr/local/rocketmq/bin
2.nohub sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-a.properties &

slave2:

1.cd /usr/local/rocketmq/bin
2.nohub sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b-s.properties &

在192.168.25.135启动master2和slave1
master2:

1.cd /usr/local/rocketmq/bin
2.nohub sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b.properties &

slave1:

1.cd /usr/local/rocketmq/bin
2.nohub sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-a-s.properties &

查看进程状态：jps
查看日志：

查看nameServer日志：
tail -$00f ~/logs/rocketmqlogs/namesrv.log
查看broker日志
tail -$00f ~/logs/rocketmqlogs/broker.log

使用mqadmin管理工具：

1.使用方式:命令方式
进入RocketMQ安装位置，在bin目录下执行./mqadmin {command} {args}
2.监控平台方式：
1.开源项目：https://github.com/apache/rocketmq-externals
下载源代码 找到rocketmq-console打成jar包
修改其中的application.propertities
rocketmq.config.namesrvAddr=192.168.25.135:9876;192.168.25.138:9876
mvn clean package -Dmaven.test.skip=true
java -jar rocket-console-ng-1.1.1.jar
访问：http:localhost:8040

消息发送案例：
导入MQ客户端依赖：

<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.4.0</version>
<dependency>

发送同步消息：
这种可靠性同步的发送方式使用比较广泛，比如：重要的消息通知，短信通知

public class SysncProducer{public static void main(String[] args) throws Exception{//1.实例化消息生产者Producer DefaultProducer producer = new DefaultMQProducer("group1");//2。设置NameServer的地址producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.启动Producer.start();producer.start();for(int i = 0;i < 100;i++){//4.创建消息，并指定topic,tag和消息体Message msg = new Message("TopicTest","TagA",("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));//5.发送消息sendResult sendResult = producer.send(msg);SendStatus status = sendResult.getSendStatus();String msgId = sendResult.getMsgId();int QueueId = sendResult.getMessageQueue.getQueueId();System.out.printf("s%%n",sendResult);TimeUnit.SECONDS.sleep(1);}//如果不再发送消息，关闭Producer实例producer.shutdown();}
}

发送异步消息
异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间的等待Broker的响应：

public class AsyncProducer{public static void main(String[] args) throws Exception{//1.实例化消息生产者Producer DefaultProducer producer = new DefaultMQProducer("group1");//2。设置NameServer的地址producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.启动Producer.start();producer.start();for(int i = 0;i < 100;i++){//4.创建消息，并指定topic,tag和消息体Message msg = new Message("TopicTest","Tag2",("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));//5.发送消息,通过回调获取响应信息producer.send(msg,new SendCallback(){//发送成功回调函数public void onSuccess(SendResult sendResult){System.out.printf("发送成功");}//发送失败回调函数public void onException(Throwable e){System.out.printf("发送失败");}});TimeUnit.SECONDS.sleep(1);}//如果不再发送消息，关闭Producer实例producer.shutdown();}
}

发送单向消息
这种方式主要用于在不特别关心发送结果的场景，例如日志发送

public class OnewayProducer{public static void main(String[] args) throws Exception{//1.实例化消息生产者Producer DefaultProducer producer = new DefaultMQProducer("group1");//2。设置NameServer的地址producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.启动Producer.start();producer.start();for(int i = 0;i < 100;i++){//4.创建消息，并指定topic,tag和消息体//4.创建消息，并指定topic,tag和消息体Message msg = new Message("TopicTest","Tag3",("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));//5.发送消息producer.sendOneway(msg);TimeUnit.SECONDS.sleep(1);}//如果不再发送消息，关闭Producer实例producer.shutdown();}
}

消费消息：
负载均衡模式消费：消息不能重复消费默认消费方式

public class Consumer{public static void main(String[] args) throws Exception{//1.实例化消息消费者Consumer DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2。设置NameServer的地址consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.订阅上题Topic和Tagconsumer.subcribe("TopicTest","Tag3");//4.设置回调函数，处理消息consumer.registerMeaasgeListener(new MessageListenerConcurrently(){//接受消息内容public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context){for (MessageExt msg : msgs){sout(msg.getBody());}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS}});//5.启动消费者consumer.start();}
}

广播模式

public class Consumer{public static void main(String[] args) throws Exception{//1.实例化消息消费者Consumer DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2。设置NameServer的地址consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.订阅上题Topic和Tag *表示所有的TAGconsumer.subcribe("TopicTest","Tag3");//设置消费模式为广播模式consumer.setMessageModel(MessageModel.BROADCASTING);//4.设置回调函数，处理消息consumer.registerMeaasgeListener(new MessageListenerConcurrently(){//接受消息内容public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context){for (MessageExt msg : msgs){sout(msg.getBody());}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS}});//5.启动消费者consumer.start();}
}

顺序消息分析：

消息有序指的是可以按照消息的发送顺序消费(FIFO)。RocketMQ可以严格保证消息有序，可以分为分区有序或者全局有序。
顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费时不能保证顺序，但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序;如果多个queue参与，则为分区有序，即相对每个queue.消息都是有序的；

顺序消息：创建、付款、推送、完成

@Data
public class OrderStep{private long orderId;private String desc;public static List<OrderStep> buildOrders(){//1039L:创建  付款  推送  完成//1065L:创建  付款  完成//7235L:创建  付款  完成List<OrderStep> orderList = new ArrayList<~>();OrderStep orderDemo = new OrderStep();orderDemo.setOrderId(1039L);orderDemo.setOrderId("创建");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(1065L);orderDemo.setOrderId("创建");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(1039L);orderDemo.setOrderId("付款");orderList.add(orderDemo); orderDemo = new OrderStep();orderDemo.setOrderId(7235L);orderDemo.setOrderId("创建");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(1065L);orderDemo.setOrderId("付款");orderList.add(orderDemo); orderDemo = new OrderStep();orderDemo.setOrderId(7235L);orderDemo.setOrderId("付款");orderList.add(orderDemo); orderDemo = new OrderStep();orderDemo.setOrderId(1065L);orderDemo.setOrderId("完成");orderList.add(orderDemo); orderDemo = new OrderStep();orderDemo.setOrderId(1039L);orderDemo.setOrderId("推送");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(7235L);orderDemo.setOrderId("完成");orderList.add(orderDemo); orderDemo = new OrderStep();orderDemo.setOrderId(1039L);orderDemo.setOrderId("完成");orderList.add(orderDemo);return orderList;}
}public class Producer{public static void main(String[] args) throws Exception{//1.实例化消息生产者Producer DefaultProducer producer = new DefaultMQProducer("group1");//2。设置NameServer的地址producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.启动Producer.start();producer.start();//构建消息集合List<OrderStep> orderStep = OrderStep.buildOrders();//发送消息for(int i = 0i<orderSteps.size();i++){String body = orderSteps.get(i)+"";;Message msg = new Message("OrderTopic","Order","i"+i,body.getBytes());orderId=orderSteps.get(i).getOrderId();//参数一 消息对象，参数二 消息队列选择器 参数三：选择队列的业务标识SsenResult sendResult = producer.send(msg,new MessageQueueSelector(){//mqs:队列集合//msg:消息对象//arg:就是orderId:业务标识参数 保证同一个orderId在同一个队列中@Overridepublic MessageQueue select(List<MessageQueue> mqs,Message msg,Object arg);long OrderId = (long)arg;long index = OrderId%mqs.size();return mqs.get((int)index);}},orderId);sout(sendReuslt);}producer.shutdown();}
}

消费者

public class Consumer{public static void main(String[] args) throws Exception{//1.实例化消息消费者Consumer DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2。设置NameServer的地址consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.订阅上题Topic和Tag *表示所有的TAGconsumer.subcribe("OrderTopic","*");//设置消费模式为广播模式// consumer.setMessageModel(MessageModel.BROADCASTING);//4.设置回调函数，处理消息//MessageListenerOrderly保证顺序消费consumer.registerMeaasgeListener(new MessageListenerOrderly(){//接受消息内容public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,ConsumeOrderlyContext context){for (MessageExt msg : msgs){sout(msg.getBody());}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS}});//5.启动消费者consumer.start();}
}

延时消息
比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

public class AsyncProducer{public static void main(String[] args) throws Exception{//1.实例化消息生产者Producer DefaultProducer producer = new DefaultMQProducer("group1");//2。设置NameServer的地址producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.启动Producer.start();producer.start();for(int i = 0;i < 100;i++){//4.创建消息，并指定topic,tag和消息体Message msg = new Message("DelayTopic","Tag1",("Hello World" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));//设置延迟时间//目前只支持“1s,5s,10s,30s,1m,2m,3m,4m,5m,6m,7m,8m,9m,10m,20m,30,1h,2h”msg.setDelayTimeLevel(2);//5.发送消息,通过回调获取响应信息producer.send(msg,new SendCallback(){//发送成功回调函数public void onSuccess(SendResult sendResult){System.out.printf("发送成功");}//发送失败回调函数public void onException(Throwable e){System.out.printf("发送失败");}});TimeUnit.SECONDS.sleep(1);}//如果不再发送消息，关闭Producer实例producer.shutdown();}
}

消费者：

public class Consumer{public static void main(String[] args) throws Exception{//1.实例化消息消费者Consumer DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2。设置NameServer的地址consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.订阅上题Topic和Tagconsumer.subcribe("DelayTopic","*");//4.设置回调函数，处理消息consumer.registerMeaasgeListener(new MessageListenerConcurrently(){//接受消息内容public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context){for (MessageExt msg : msgs){//获取延迟时间System.CurruntTimeMills()-msg.getStoreTimestamp();sout(msg.getBody());}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS}});//5.启动消费者consumer.start();sout("消费者启动")}
}

批量消息
批量发送消息能显著提高传递消息的性能.限制是这些批量消息应该有相同的topic,相同的waitStoreMsgOK,而且不能是延时消息，此外，这一批消息的总大小不应超过4MB.

public class Producer{public static void main(String[] args) throws Exception{//1.实例化消息生产者Producer DefaultProducer producer = new DefaultMQProducer("group1");//2.设置NameServer的地址producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.启动Producer.start();producer.start();List<Message> msgs = new ArrayList<Message>();//4.创建消息，并指定topic,tag和消息体Message msg1 = new Message("BatchTopic","Tag1",("Hello World" + 1).getBytes());Message msg2 = new Message("BatchTopic","Tag1",("Hello World" + 2).getBytes());
3          Message msg3 = new Message("BatchTopic","Tag1",("Hello World" + 1).getBytes());msgs.add(msg1);msgs.add(msg2);msgs.add(msg3);//设置延迟时间//目前只支持“1s,5s,10s,30s,1m,2m,3m,4m,5m,6m,7m,8m,9m,10m,20m,30,1h,2h”// msg.setDelayTimeLevel(2);//5.发送消息,通过回调获取响应信息SendResult result = producer.send(msgs);TimeUnit.SECONDS.sleep(1);//如果不再发送消息，关闭Producer实例producer.shutdown();}
}

消费者：

public class Consumer{public static void main(String[] args) throws Exception{//1.实例化消息消费者Consumer DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2。设置NameServer的地址consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.订阅上题Topic和Tagconsumer.subcribe("BatchTopic","*");//4.设置回调函数，处理消息consumer.registerMeaasgeListener(new MessageListenerConcurrently(){//接受消息内容public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context){for (MessageExt msg : msgs){//获取延迟时间System.CurruntTimeMills()-msg.getStoreTimestamp();sout(msg.getBody());}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS}});//5.启动消费者consumer.start();sout("消费者启动")}
}

如果消息的总长度可能大于4MB时，这时候最好把消息进行分割

public class ListSplitter implements Iterator<List<Message>>{private final int SIZE_LIMIT = 1024*1024*4;private final List<Message> message;private int currindex;public ListSplitter(List<Message> messages){this.messages = messages;}@Overridepublic boolean hasNext(){return currIndex < message.size();}@Overridepublic 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 : entry.getValue.length()){tmpSize += entry.getKey().length() + entry.getValue().length(); }tmpSize = tmpSize + 20;//增加日志的开销20字节if(tmpSize > SIZE_LIMIT){//单个消息超过了最大的限制//忽略，否则会短距离阻塞分割的线程if(nexzIndex - 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();}
}

过滤消息
在大多数情况下，TAG是一个简单而有用的设计，其可以来选择您想要的消息.例如：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC","TAGA || TAGB || TAGC");

消费者将接收包含TAGA或TAGB或TAGC的消息，但是限制是一个消息只能由一个标签，这对于复杂的场
景可能不起作用。在这种情况下，可以使用SQL表达式帅选消息。SQL特性可以通过发送消息的属性来
进行计算。

TAG过滤：

public class AsyncProducer{public static void main(String[] args) throws Exception{//1.实例化消息生产者Producer DefaultProducer producer = new DefaultMQProducer("group1");//2。设置NameServer的地址producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//3.启动Producer.start();producer.start();for(int i = 0;i < 10;i++){//4.创建消息，并指定topic,tag和消息体Message msg = new Message("FilterTagTopic","Tag1",("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));//SQL语法过滤msg.putUserProperty("i",String.valueOf(i));//consumer里面按如下即可，只消费i>5的消息//Consumer.subscribe("FilterTagTopic",MessageSelector.bySql("i>5"));//5.发送消息,通过回调获取响应信息producer.send(msg,new SendCallback(){//发送成功回调函数public void onSuccess(SendResult sendResult){System.out.printf("发送成功");}//发送失败回调函数public void onException(Throwable e){System.out.printf("发送失败");}});TimeUnit.SECONDS.sleep(1);}//如果不再发送消息，关闭Producer实例producer.shutdown();}
}

事务消息：

事务消息发送及提交
1.发送消息（half消息）
2.服务端响应消息写入结果；
3.根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）
4.根据本地事务状态执行Commit或者Rollback(Commit操作生成消息索引，消息对消费者可见);
事务补偿
1.对没有Commikt/Rollback的事务消息（Pending状态的消息），从服务端发起一次"回查"
2.Producer收到回查消息，检查回查消息对应的本地事务的状态
3.根据本地事务状态，重新Commit或者Rollback
其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。
事务消息状态
事务消息共有三种状态：提交，回滚，中间

发送事务消息：消费者跟之前一致

public class AsyncProducer{public static void main(String[] args) throws Exception{//1.实例化消息生产者Producer TransactionMQProducer producer = new TransactionMQProducer("group1");//2。设置NameServer的地址producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138.9876");//设置监听器producer.setTransactionListener(new TransactionListener(){//在该方法中执行本地事务@Overridepublic LocalTransactionState executeLocalTransaction(Message msg ,Object srg){if(StringUtils.equals("TAGA",MSG.GETtAGS())){return LocalTransactionState.COMMIT_MESSAGE;}  else if(StringUtils.equals("TAGB",MSG.GETtAGS())){return LocalTransactionState.ROLLBACK_MESSAGE;}else if (StringUtils.equals("TAGC",MSG.GETtAGS())){return LocalTransactionState.UNKONW;//进行回查}return LocalTransactionState.UNKONW; }//MQ事务状态回查@Overridepublic LocalTransactionState checklocalTransaction(MessageExt msg){sout("消息的tag:"+msg.getTags());return LocalTransactionState.COMMIT_MESSAGE;}});//3.启动Producer.start();producer.start();String[] tag = {"TAGA","TAGB","TAGC"};for(int i = 0;i < 100;i++){//4.创建消息，并指定topic,tag和消息体Message msg = new Message("TranctionTopic",tag[i],("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));//5.发送消息,通过回调获取响应信息null表示所有的消息都会受到事务控制producer.sendMessageInTransaction(msg,null);TimeUnit.SECONDS.sleep(1);}//如果不再发送消息，关闭Producer实例// producer.shutdown();}
}

rocketMQ系列1相关推荐

Springboot整合Rocketmq系列教程
Springboot整合Rocketmq系列教程本教程是基于Springboot2.6.3整合Rocketmq5.0,其中涉及了Rocketmq的安装,消息的发送及消费的代码实现. 本文不会对roc ...
从源码分析RocketMQ系列-Remoting通信架构源码详解
导语这篇博客要从官方给出的一张图开始说起,之前的分析我们都是简单的分析了一下消息传递的流程,以及消息传递流程过程中出现的一些类的封装,并且提出,所有的封装操作都是为了更加高效的服务于NameSe ...
从源码分析RocketMQ系列-MQClientInstance类详解
导语在之前的分析中,看到有一个类MQClientInstance,这个无论是在Producer端还是在Consumer端都是很重要的一个类,很多的功能都是从这个类发起的,这边分享中就来详细的看看 ...
从源码分析RocketMQ系列-Producer的SendResult来自哪里？
导语对于消息中间件大家都应该不陌生,现在比较主流的消息中间件有Kafka.RabbitMQ.RocketMQ.ActiveMQ等等.前段时间花了很长时间分析了关于RocketMQ源码,之前也分享 ...
从源码分析RocketMQ系列-消息拉取PullMessageProcessor详解
导语在之前的分析中分析了关于SendMessageProcessor,并且提供了对应的源码分析分析对于消息持久化的问题,下面来看另外一个PullMessageProcessor,在RocketM ...
从源码分析RocketMQ系列-RocketMQ消息持久化源码详解
导语在上篇分析中,提到了一个概念处理器,并且在进入到最终NettyIO的时候看到了一个Pair的对象,这个对象存储了两个对象,一个是执行器,一个是处理器,在进入Runable对象的时候看到封装到 ...
从源码分析RocketMQ系列-Consumer消息接收逻辑
导语在前面的分析中分析了关于Producer发送消息的逻辑,并且追踪到了在DefaultMQPushConsumerImpl 类中的有对应的消息监听方法,这个消息监听的方法是从Consumer调 ...
从源码分析RocketMQ系列-start()方法详解
导语在之前的分析中主要介绍的是关于Producer 发送消息的逻辑,但是在实例代码中有一个操作是producer.start()方法,在Consumer中看到的方法是consumer.start ...
从源码分析RocketMQ系列-Producer的SendResult的封装
导语通过之前博客的Producer的SendResult来自哪里分析到发送的核心机制,了解了在发送之前被使用的几个Hook,以及发送消息的RequestHeader的封装,但是这些封装都被一个t ...
RocketMQ系列---消息消费者及消费方式
1.消息消费 public class Consumer {public static void main(String[] args) throws InterruptedException, MQ ...

rocketMQ系列1

rocketMQ系列1相关推荐

最新文章

热门文章