代码拉取完成，页面将自动刷新
以下前言的几个问题选自博主割肉机的博文：
本文学习参考自B站尚硅谷的【尚硅谷】2021新版RocketMQ教程丨深度掌握MQ消息中间件
下文的所以代码工程都放在了我的gitee仓库中rocketmq-study
传统模式的缺点：
系统间耦合性太强，如上图所示，系统A在代码中直接调用系统B和系统C的代码，如果将来D系统接入，系统A还需要修改代码，过于麻烦！
中间件模式的优点：
将消息写入消息队列，需要消息的系统自己从消息队列中订阅，从而系统A不需要做任何修改
传统模式的缺点：
一些非必要的业务逻辑以同步的方式运行，太耗费时间。
中间件模式的的优点：
将消息写入消息队列，非必要的业务逻辑以异步的方式运行，加快响应速度
传统模式的缺点：
并发量大的时候，所有的请求直接怼到数据库，造成数据库连接异常
中间件模式的的优点：
系统A慢慢的按照数据库能处理的并发量，从消息队列中慢慢拉取消息。在生产中，这个短暂的高峰期积压是允许的
我们引入一个技术，要对这个技术的弊端有充分的认识，才能做好预防 从以下两个个角度来答
但是，我们该用还是要用的。
| 特性 | ActiveMQ | RabbitMQ | RocketMQ | kafka |
|---|---|---|---|---|
| 开发语言 | java | erlang | java | scala |
| 单机吞吐量 | 万级 | 万级 | 10万级 | 10万级 |
| 时效性 | ms级 | us级 | ms级 | ms级以内 |
| 可用性 | 高(主从架构) | 高(主从架构) | 非常高(分布式架构) | 非常高(分布式架构) |
| 功能特性 | 成熟的产品，在很多公司得到应用；有较多的文档；各种协议支持较好 | 基于erlang开发，所以并发能力很强，性能极其好，延时很低;管理界面较丰富 | MQ功能比较完备，扩展性佳 | 只支持主要的MQ功能，像一些消息查询，消息回溯等功能没有提供，毕竟是为大数据准备的，在大数据领域应用广 |
综合上面的材料得出以下两点:
在第二点说过了，引入消息队列后，系统的可用性下降。
在生产中，没人使用单机模式的消息队列。
因此，作为一个合格的程序员，应该对消息队列的高可用有很深刻的了解。
如果面试的时候，面试官问，你们的消息中间件如何保证高可用的？
你的回答只是表明自己只会订阅和发布消息，面试官就会怀疑你是不是只是自己搭着玩，压根没在生产用过。
这个问题其实换一种问法就是，如何保证消息队列的幂等性?
这个问题可以认为是消息队列领域的基本问题。换句话来说，是在考察你的设计能力，这个问题的回答可以根据具体的业务场景来答，没有固定的答案。
先来说一下为什么会造成重复消费?
其实无论是那种消息队列，造成重复消费原因其实都是类似的。
那造成重复消费的原因?
如何解决?
这个问题针对业务场景来答分以下几点
我们在使用消息队列的过程中，应该做到消息不能多消费，也不能少消费。
如果无法做到可靠性传输，可能给公司带来千万级别的财产损失。同样的，如果可靠性传输在使用过程中，没有考虑到，这不是给公司挖坑么，你可以拍拍屁股走了，公司损失的钱，谁承担？
还是那句话，认真对待每一个项目，不要给公司挖坑。
其实这个可靠性传输，每种MQ都要从三个角度来分析：
其实并非所有的公司都有这种业务需求，但是还是对这个问题要有所复习。
针对这个问题，通过某种算法，将需要保持先后顺序的消息放到同一个消息队列中(kafka中就是partition,rabbitMq中就是queue)。
然后只用一个消费者去消费该队列。
有的人会问：**那如果为了吞吐量，有多个消费者去消费怎么办？ **
这个问题，没有固定回答的套路。
比如我们有一个微博的操作，发微博、写评论、删除微博，这三个异步操作。
如果是这样一个业务场景，那只要重试就行。
比如你一个消费者先执行了写评论的操作，但是这时候，微博都还没发，写评论一定是失败的，等一段时间。等另一个消费者，先执行写评论的操作后，再执行，就可以成功。
总之，针对这个问题，我的观点是**保证入队有序就行，出队以后的顺序交给消费者自己去保证，没有固定套路。 **
MQ，message queue消息队列，一种提供消息队列服务的中间件，是一套提供了消息生产、储存、消费全过程的API软件系统
消息即数据
MQ可以将系统的超量请求暂存其中，以便系统慢慢处理，避免请求丢失或系统被压垮
上游系统对下游系统的调用若为同步调用，则会大大降低系统的吞吐量与并发度，且系统耦合度太高。而异步调用则会解决这些问题。所以两层之间若要实现由同步到异步的转化，一般性做法就是，在这两层间添加一一个MQ层。
后台保证了数据库一定能正常修改运行，所以我无序同步等待执行结束，而是直接在用户请求后返回订单号，接着异步去处理数据库
分布式系统会产生海量级数据流，如:业务日志、监控数据、用户行为等
针对这些数据流进行实时或批量采集汇总，然后对这些数据流进行大数据分析，这是当前互联网平台的必备技术。通过MQ完成此类数据收集是最好的选择。
MOM是什么？
MOM（Message Oriented Middleware）是面向消息的中间件，使用消息传送提供者来协调消息传送操作。MOM 需要提供 API 和管理工具。客户端使用api调用，把消息发送到由提供者管理的目的地。在发送消息之后，客户端会继续执行其他工作，并且在接收方收到这个消息确认之前，提供者一直保留该消息
- PO：面向过程，Process Oriented
- OO：面向对象，Object Oriented
- AO：面向切面，Aspect Oriented
JMS
java message service，java消息服务
是java平台有关MOM的技术规范，它便于消息系统中的java应用程序进行消息交换，并且通过提供标准的产生、发送、接收消息的接口，简化企业应用的开发。
ActiveMQ是该协议的典型实现
STOMP
Stream Text Orientated Message Protocol，面向文本流的消息协议
是一种MOM设计的简单文本协议，STOMP提供一个可互操作的连接格式，允许客户端与任意STOMP消息代理（Broker）进行交互。
ActiveMQ是该协议的典型实现，RabbitMQ通过插件可以支持该协议
AMQP
Advanced Message Queuing Protocol，高级消息队列协议
一个提供统一消息服务的应用层标准，是应用层协议的一个开放标准，是一种MOM设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同开发语言等条件的限制。
RabbitMQ是该协议的典型实现。
MQTT
Message Queuing Telemetry Transport，消息队列遥测传输
是IBM开发的一个即时通讯协议，是一种二进制协议，主要用于服务器和低功耗IoT设备（物联网）间的通信。该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当做传感器和致动器的通信协议。
RabbitMQ通过插件可以支持该协议。
以上协议，RockerMQ都不支持
消息是指，消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。
topic表示一类消息的集合，每个主题包含若干个消息，每条消息只属于一个主题，是对rocketmq进行消息订阅的基本单位
一个生产者可以发送多种topic的消息，而一个消费者只对某种特定的topic感兴趣，即只可以订阅和消费一种topic的消息
消息标签，用来进一步区分某个 Topic 下的消息分类，消息队列 RocketMQ 允许消费者按照 Tag 对消息进行过滤，确保消费者最终只消费到他关注的消息类型
Topic 与 Tag 都是业务上用来==归类==的标识，区分在于 Topic 是一级分类，而 Tag 可以说是二级分类，关系如图所示
储存消息的实体，一个topic中可以含有多个queue，每个queue中储存的是该topic的信息
一个topic的queue也被称为一个topic中消息的分区
一个topic的queue的消息只能被一个消费者组中的消费者进行读取消费，且一个queue的消息不允许同一个消费者组里多个消费者同时消费（后面说的广播模式推翻了这个说法）
queue：消费者 => 1：1
消费者：queue => 1：n
为了提升MQ的效率，所以将topic分为多个分区去进行读取（个人感觉有点像负载均衡）
注：1号狗消费了1号骨头，就不能和2号狗去争夺2号骨头了
如果此时只有1和2号两只狗的时候，就可以让1号消费1、2号骨头，而2号狗消费3号骨头了
就是说一个消费者组里不能互相争夺分区的消息
在学习参考其它相关资料时，还会看到一个概念：分片(Sharding)
官方是没有这个概念的，但是在网上有这种说法
分片不同于分区。在RocketMQ中，分片指的是存放相应Topic的Broker的数量。
每个分片中会创建出相应数量的分区，即Queue,每个Queue的大小都是相同的。
消息生产者，负责生产消息
比如我将系统的日志写入MQ，就是一个消息的生成
将用户的请求写入MQ，也是一个消息的生成
Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟
RocketMQ中的消息生产者都是以==生产者组==(Producer Group)的形式出现的。
生产者组是同一类生产者的集合，这类Producer发送==相同==Topic类型的消息。一个生产者组可以同时发送多个主题的消息（我的理解是可以生产多个主题的消息，但是发送时要保持消息的主题是一致的）
消息消费者，负责消费消息
从MQ取出日志信息进行解析，就是消息的消费
用户下单后，从MQ中读取用户的请求并进行处理的过程，也是消息的消费过程
一个消息消费者会从Broker服务器中获取到消息，并对消息进行相关业务处理
RocketMQ中的消息消费者都是以==消费者组==(Consumer Group)的形式出现的
消费者组是同一类消费者的集合，这类Consumer消费的是同一个Topic类型的消息，一类消费者只能消费一种topic的消息
消费者组使得在消息消费方面，实现负载均衡和容错的目标变得非常容易
消费者组中Consumer的数量应该小于等于订阅Topic的Queue数量。如果超出Queue数量，则多出的Consumer将不能消费消息。
一个topic类型的消息可以被多个消费者组消费
注：
NameServer是一个Broker与Topic路由的注册中心，支持Broker的动态注册与发现。
主要包括两个功能:
Broker管理:
路由信息管理:
NameServer通常也是以集群的方式部署，不过， NameServer是无状态的，即NameServer集群中的各个节点间是无差异的，各节点间相互不进行信息通讯
那各节点中的数据是如何进行数据同步的呢?
在Broker节点启动时，轮询NameServer列表，与每个NameServer节点建立长连接，发起注册请求。
在NameServer内部维护着1个Broker列表， 用来动态存储Broker的信息。
注意：
与其他zk、euraka、nacos等注册中心不同，别的都是注册一个然后内部通信，nameserver是所有都要注册
优点：无状态的集群搭建很简单
缺点：对于broker必须明确指出nameserver地址，未指出的不会进行注册，使得扩容、维护不方便
Broker节点为了证明自己是活着的，为了维护与NameServer间的长连接，会将最新的信息以心跳包的方式上报给NameServer,每30秒发送一次心跳。
心跳包中包含Brokerld、Broker地址（IP+port）、 Broker名称、Broker所属集群名称等等。
NameServer在接收到心跳包后，会更新心跳时间戳，记录这个Broker的最新存活时间。
由于Broker关机、宕机或网络抖动等原因，NameServer没有收到Broker的心跳，NameServer可能会将其从Broker列表中剔除
NameServer中有一个定时任务，每隔10秒就会扫描一次Broker表， 查看每一个Broker的最新心跳时间戳距离当前时间是否超过120秒，如果超过，则会判定Broker失效，然后将其从Broker列表中剔除。
RocketMQ的路由发现采用的是Pull模型。当Topic路由信息出现变化时，NameServer不会主动推送给客户端，而是客户端定时拉取主题最新的路由
默认客户端每30秒会拉取一次最新的路由。
Pull模型：客户端隔一段时间去拉取服务端的数据
- 可能存在“前脚拉完，后脚就改”的情况，实时性差
Push模型：服务端被客户端订阅后，一旦发生改变，立马推送到客户端
- 保证数据一致，实时性较好
- 但是需要一直维护两边的长连接，占用资源
Long Pulling模型：长轮询模型，客户端隔一段时间去拉取服务端的数据，且不立刻断开连接，而是服务端保持一段时间和客户端的连接，也就是Pull模型和Push模型的整合
- nacos的配置中心用的就是这个，监控服务端的配置信息，一旦变更，客户端立马变更
- 实时性较好
- 对资源占用较少
所以push模式适合Client不多，数据变化频繁，对实时性要求高的业务需求
这里的客户端指：producer和consumer
客户端在配置时必须要写上NameServer集群的地址，那么客户端到底连接的是哪个NameServer节点呢?
客户端首先会首先一个随机数， 然后再与NameServer节点数量取模，此时得到的就是所要连接的节点索引，然后就会进行连接。
如果连接失败，则会采用round robin轮询策略，逐个尝试着去连接其它节点。
即：首先采取随机策略进行选择，连接失败后再使用轮询策略
Broker充当着消息中转角色，负责存储消息、转发消息
Broker在RocketMQ系统中负责接收并存储从生产者发送来的消息，同时为后费者的拉取请求作准备
Broker同时也存储着消息相关的元数据，包括消费者组消费进度偏移量offset、主题topic、队列queue等
Remoting Module：
整个Broker的实体，负责处理来自clients端的请求。这个Broker实体则由以下模块构成。
**Client Manager：**客户端管理器
负责接收、解析客户端(Producer/Consumer)请求， 管理客户端
例如，维护Consumer的Topic订阅信息
Store Service：存储服务
HA Service：高可用服务
Index Service：索引服务
为了增强Broker性能与吞吐量，Broker一般都是以集群形式出现的。各集群节点中可能存放着相同Topic的不同Queue。
不过，这里有个问题，如果某Broker节点宕机，如何保证数据不丢失呢？
其解决方案是，将每个Broker集群节点进行横向扩展，将Broker节点建为一个高可用的HA集群，解决单点问题
Broker节点集群是一个主从集群，即集群中具有Master与Slave两种角色。Master负责处理读写操作请求，Slave也可以负责读写操作请求（当master宕机后需要从slave中读写）
正常情况下都是操作master，slave只是备份服务，如果master宕机，则由slave顶上
一个Master可以包含多个Slave，但一个Slave只能属于一个Master。
Master与Slave的对应关系是通过指定相同的BrokerName、不同的BrokerId 来确定的。
BrokerId为0表示Master，非0表示Slave。
每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。
启动Name Server，开始监听端口，等待broker、producer和consumer的连接
启动Broker时， Broker会与所有的NameServer建立并保持长连接，然后每30秒向NameServer定时发送心跳包
发送消息前，可以先创建Topic。创建Topic时需要指定该Topic要存储在哪些Broker上， 当然，在创建Topic时也会将Topic与Broker的关系写入到NameServer中。不过，这步是可选的，也可以在发送消息时自动创建Topic
手动创建Topic：
集群模式：该模式下创建的Topic在该集群中，所有Broker中的Queue数量是相同的
broker模式：该模式下创建的Topic在该集群中，每个Broker中的Queue数量可以是不同的
自动创建Topic：
默认采用的broker模式，默认给每个broker创建4个queue，可以在配置文件中更改
Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接（先随机再轮询），并从NameServer中获取路由信息，即当前发送的Topic的Queue 与Broker的地址(IP+Port) 的映射关系。然后根据算法策略从队选择-个Queue,与队列所在的Broker建 立长连接从而向Broker发消息。当然，在获取到路由信息后，Producer会首先将路由信息缓存到本地，再每30秒从NameServer更新一次路由信息
Consumer跟Producer类似， 跟其中一台NameServer建立长连接，获取其所订阅Topic的路由信息， 然后根据算法策略从路由信息中获取到其所要消费的Queue,然后直接跟Broker建立长连接，开始消费其中的消息。Consumer在获取到路由信息后，同样也会每30秒从NameServer更新一次路由信息。不过不同于Producer的是, Consumer还会向Broker发送心跳，以确保Broker的存活状态
从物理上来讲，读/写队列是同一个队列。
所以，不存在读/写队列数据同步问题。读/写队列是逻辑上进行区分的概念。
一般情况下，读/写队列数量是相同的
例如，创建Topic时设置的写队列数量为8，读队列数量为4，此时系统会创建8个Queue,分别是0 1 2 3 4 5 67。
Producer会将消息写入到这8个队列，但Consumer只会消费0 1 2 3这4个队列中的消息，4 5 6 7中的消息是不会被消费到的。
再如，创建Topic时设置的写队列数量为4，读队列数量为8，此时系统会创建8个Queue,分别是0 1 2 3 4 5 6 7。
Producer会将消息写入到0 1 2 3这4个队列，但Consumer只会消费0 1 2 3 4 5 6 7这8个队列中的消息，但是4 567中是没有消息的。此时假设Consumer Group中包含两个Consumer, Consumer1消费0 1 2 3，而Consumer2消费4567。但实际情况是，Consumer2是没有消息可消费的。
不管怎么样都会造成资源浪费等不好的情况，所以一般读写队列的数量要设计一致
准备一台linux，我用的vm虚拟机centos，输入ifconfig获得ens33的ip为inet 192.168.146.128
在C:\Windows\System32\drivers\etc\修改host文件，将ip指向我自定义命名的centos域名
查看我的linux有没有jdk环境echo $JAVA_HOME，答案是没有
去官网下载linux的安装包，传到虚拟机
配置环境变量，输入3个值保存后，刷新profile
vi /etc/profile
# java
export JAVA_HOME=/usr/bin/java/jdk1.8.0_291
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib
source /etc/profile
测试java环境
javac
java -version
echo $JAVA_HOME
CentOS升级到7之后，发现无法使用iptables控制Linux的端口，因为Centos 7使用firewalld代替了原来的iptables。下面记录如何使用firewalld开放Linux端口：
开启端口
firewall-cmd --zone=public --add-port=9800/tcp --permanent
查询端口是否开启
firewall-cmd --query-port=9800/tcp
重启防火墙
firewall-cmd --reload
查询有哪些端口是开启的
firewall-cmd --list-port
这里方法先放着，等会如果出现开启控制台无法访问时，应该是端口号没打开，例如我rocketmq-console选择9800为端口，所以需要开放9800端口
我这里下载的4.9.1版本的，根据自己需求下载即可
我们下载Binary文件，zip文件，然后传到linux上
unzip rocketmq-all-4.9.1-bin-release.zip解压文件
因为默认的内存太大了，2G、4G啥的，服务器的配置可能起不来，所以修改一下配置
runserver.sh
修改为256m
runbroker.sh
修改为256m
# 启动nameserver服务
> nohup sh bin/mqnamesrv &
# 查看日志是否启动成功
> tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...
# 查看当前java进程
> jps
240921 NamesrvStartup
246169 Jps
> nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf &
> tail -f ~/logs/rocketmqlogs/broker.log
The broker[%s, 172.30.30.233:10911] boot success...
> jps
253009 BrokerStartup
254698 Jps
240921 NamesrvStartup
端口必须开放9876和10911端口
在发送/接收消息之前，我们需要告诉客户端名称服务器的位置。RocketMQ提供了多种方法来实现这一点。为了简单起见，我们使用环境变量namersv_ADDR
以下为官方提供的一个发消息和消费消息的用例
> export NAMESRV_ADDR=localhost:9876
> sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
SendResult [sendStatus=SEND_OK, msgId=7F0000011BE34AA298B76656B91103E7, offsetMsgId=C0A8928000002A9F000000000002ECD2, messageQueue=MessageQueue [topic=TopicTest, brokerName=centos, queueId=2], queueOffset=249]
close the connection to remote address[127.0.0.1:9876] result: true
> sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
ConsumeMessageThread_2 Receive New Messages: [MessageExt [brokerName=centos, queueId=0, storeSize=192, queueOffset=145, sysFlag=0, bornTimestamp=1634734558909, bornHost=/192.168.146.128:33198, storeTimestamp=1634734558910, storeHost=/192.168.146.128:10911, msgId=C0A8928000002A9F000000000001B352, commitLogOffset=111442, bodyCRC=953417484, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=250, CONSUME_START_TIME=1634734753329, UNIQ_KEY=7F0000011BE34AA298B76656B6BD0245, CLUSTER=DefaultCluster, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 53, 56, 49], transactionId='null'}]]
> sh bin/mqshutdown broker
The mqbroker(36695) is running...
Send shutdown request to mqbroker(36695) OK
> sh bin/mqshutdown namesrv
The mqnamesrv(36664) is running...
Send shutdown request to mqnamesrv(36664) OK
下载解压，是一个maven项目
进入src、resource文件夹，修改properties文件
默认端口是8080，这我们当然要改动了
而且他也是相当于一个client，所以需要去获取nameserver，所以需要配置nameserver的地址
server.port=9800
rocketmq.config.namesrvAddr=localhost:9876
导入JAXB依赖：Java Architecture for XML Binding
允许Java开发人员将Java类映射为XML表示方式。
JAXB提供两种主要特性：
- 将一个Java对象序列化为XML，以及反向操作，将XML解析成Java对象。
- 换句话说，JAXB允许以XML格式存储和读取数据，而不需要程序的类结构实现特定的读取XML和保存XML的代码。
<dependency>
<groupId>javax.xml.bind</groupId>
<artifactId>jaxb-api</artifactId>
<version>2.3.0</version>
</dependency>
<dependency>
<groupId>com.sun.xml.bind</groupId>
<artifactId>jaxb-impl</artifactId>
<version>2.3.0</version>
</dependency>
<dependency>
<groupId>com.sun.xml.bind</groupId>
<artifactId>jaxb-core</artifactId>
<version>2.3.0</version>
</dependency>
<dependency>
<groupId>javax.activation</groupId>
<artifactId>activation</artifactId>
<version>1.1.1</version>
</dependency>
跳过测试环节，目前不满足测试条件，会报错
mvn clean package -Dmaven.test.skip=true
得到结果则打包成功
[INFO] Building jar: E:\speciality_apps\RocketMQ\rocketmq-console-1.0.0\rocketmq-externals-rocketmq-console-1.0.0\rocketmq-console\target\rocketmq-console-ng-1.0.0-sources.jar
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 54.114 s
[INFO] Finished at: 2021-10-20T21:29:12+08:00
[INFO] ------------------------------------------------------------------------
rocketmq-console-ng-1.0.0-sources.jar就是我们打包后的可执行jar，输入指令运行
java -jar rocketmq-console-ng-1.0.0.jar
出现运行成功和端口即可
[2021-10-20 21:41:23.209] INFO Tomcat started on port(s): 9800 (http)
接下来访问我们的控制台，我这里访问的虚拟机
在OPS运维栏好像是可以添加删除和更新rocketmq的地址的
我们重新发送消息测试样例
export NAMESRV_ADDR=localhost:9876
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
然后在导航栏选中Message，选中TopicTest点击search
查询出来刚刚发送过去但是还未消费的消息
producer和consumer集群只需要写入相同的group_name即可成为集群，启动一个就会增加一个节点，这里暂且不论
nameserver呢相互之间没有通讯，所以也是启动即节点，这里也不论
所以本章讲解的是broker集群
复制策略是Broker的Master与Slave间的数据同步方式
分为同步复制与异步复制：
由内存到磁盘的过程，就叫刷盘（持久化？）
刷盘策略指的是broker中消息的落盘方式，即消息发送到broker内存后消息持久化到磁盘的方式
分为同步刷盘与异步刷盘:
- 异步策略当然会降低延迟，提高吞吐量
- 消息写入到Broker的内存，一般是写入了PageCache
- 对于异步策略来说，写入到PageCache后不会立马进行刷盘操作，而是等待数据达到一定量的时候再进行落盘（应该也是提高效率，减少频繁的io）
只有一个broker (其本质上就不能称为集群)。这种方式也只能是在测试时使用，生产环境下不能使用
因为存在单点问题
什么是单点问题？
指系统中一点失效，就会让整个系统无法运作的部件，换句话说，单点故障即会整体故障
broker集群仅由多个master构成，不存在Slave。 同一Topic的各个Queue会平均分布在各个master节点上
优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢(异步刷盘丢失少量消息，同步刷盘一条不丢)，性能最高
缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅(不可消费)，消息实时性会受到影响
什么是raid10？
Raid 10是一个Raid 1与Raid0的组合体，它是利用奇偶校验实现条带集镜像，所以它继承了Raid0的快速和Raid1的安全。我们知道，RAID 1在这里就是一个冗余的备份阵列，而RAID 0则负责数据的读写阵列。其实，概述图只是一种RAID 10方式，更多的情况是从主通路分出两路，做Striping操作，即把数据分割，而这分出来的每一路则再分两路，做Mirroring操作，即互做镜像
broker集群由多个master构成，每个master又配置了多个slave
注：在配置了RAID磁盘阵列的情况下，一个master一般配置一个slave即可
master 与slave的关系是主备关系，即master负责处理消息的读写请求，而slave仅负责消息的备份与master宕机后的角色切换
异步复制即前面所讲的复制策略中的异步复制策略，即消息写入master成功后，master立即向producer返回成功ACK，无需等待slave同步数据成功
该模式的最大特点之一是：当master宕机后，slave能够自动切换为master
不过由于slave从master的同步具有短暂的延迟(毫秒级)，所以当master宕机后，这种异步复制方式可能会存在少量消息的丢失问题
该模式是多Master多slave模式的同步复制实现
所谓同步双写，指的是消息写入master成功后，master会等待slave同步数据成功后才向producer返回成功ACK，即master 与slave都要写入成功后才会返回成功ACK，也即双写
该模式与异步复制模式相比，优点是消息的安全性更高，不存在消息丢失的情况。但单个消息的RT略高，从而导致性能要略低(大约低10%)
该模式存在一个大的问题：对于目前的版本，Master宕机后， Slave不能自动切换到Master
一般会为Master配置RAID10磁盘阵列，然后再为其配置一个Slave
即利用了RAID10磁盘阵列的高效、安全性，又解决了可能会影响订阅的问题
但是RAID10磁盘阵列价格较高且可能会影响订阅的问题，所以个人认为还是多Master+slave+异步策略为最佳
将linux系统克隆一份，选用两台机器进行集群的搭建
进入conf文件夹，查看文件夹，有系统默认给出的3个方案
我们就进去2m2s异步策略文件夹
[root@centos conf]# cd 2m-2s-async/
[root@centos 2m-2s-async]# ll
总用量 16
-rwxr-xr-x 1 root root 929 8月 16 06:20 broker-a.properties
-rwxr-xr-x 1 root root 922 8月 16 06:20 broker-a-s.properties
-rwxr-xr-x 1 root root 929 8月 16 06:20 broker-b.properties
-rwxr-xr-x 1 root root 922 8月 16 06:20 broker-b-s.properties
很明显的可以看出，-s是slave配置文件，正常的就是master配置文件
我们这是选用
即
vim修改配置文件broker-a.properties
#指定整个broker集群的名称，或者说是RocketMQ集群的名称
brokercluste rName=Defaultcluster
#指定master-slave集群的名称。一个RocketMQ集群可以包含多个master-slave集群，所以m和s保持一致
brokerName=broker-a
# master的brokerId为0，0代表master，1代表slave
brokerId=0
#指定删除消息存储过期文件的时间为凌晨4点
deletewhen=04
#指定未发生更新的消息存储文件的保留时长为48小时，48小时后过期，将会被删除
fileReservedTime=48
#指定当前broker为异步复制master
brokerRole=ASYNC_MASTER
#指定刷盘策略为异步刷盘
f1ushDi skType=ASYNC_FLUSH
我们需要配置，指定Name Server的地址
#指定Name Server的地址
namesrvAddr=192.168.59.164:9876;192.168.59.165:9876
修改配置文件broker-b-s.properties
brokercluste rName=Defaultcluster
#指定这是另外个master-slave集群
brokerName=broker-b
# slave的brokerId为非0
brokerId=1
deletewhen=04
fileReservedTime=48
#指定当前broker为slave后
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
namesrvAddr=192.168.59.164:9876;192.168.59.165:9876
#指定Broker对外提供服务的端口，即Broker与producer与consumer通信的端口。默认10911。由于当前主机同时充当着master1与s]ave2，而前面的master1使用的是默认端口。这里需要将这两个端口加以区分，以区分出master1与s1ave2
listenPort=11911
#指定消息存储相关的路径。 默认路径为~/store目录。由于当前主机同时充当着master1与slave2, master1使用的是默认路径，这里就需要再指定一个不同路径
storePathRootDir=~/store-s
storePathcommi tLog=~/store-s/commit1og
storepathconsumeQueue=~/store-s/consumequeue
storepathIndex=~/store-s/ i ndex
storeCheckpoint=~/store-s/checkpoint
abortFile=~/store-s/abort
另一台机器同理设置另外两个配置
上文有写
nohup sh bin/mqnamesrv &
tail -f ~/logs/rocketmqlogs/namesrv.log
与上文不同，启动时需要指定文件启动
分别启动两个机器的broker master
nohup sh bin/mqbroker -C conf/2m-2s-async /broker-a.properties &
tail -f ~/logs/rocketmqlogs/broker.log
nohup sh bin/mqbroker -C conf/2m-2s-async /broker-b.properties &
tail -f ~/logs/rocketmqlogs/broker.log
nohup sh bin/mqbroker -C conf/2m-2s-async /broker-b-s.properties &
tail -f ~/logs/rocketmqlogs/broker.log
nohup sh bin/mqbroker -C conf/2m-2s-async /broker-a-s.properties &
tail -f ~/logs/rocketmqlogs/broker.log
rocketmq.config.namesrvAddr=ipA:9876;ipB:9877
应该也可以直接在控制台添加一个nameserver
在mq的bin目录有一个mqadmin命令，该命令是一个运维命令，用于对mq的主题、集群、broker等信息进行管理
这个命令能够完成的功能在控制台都可以完成
这里就不再阐述了，贴一个博客地址
这个是一个重点、难点、面试高频考点
Producer可以将消息写入到某Broker中的某Queue中，其经历了如下过程：
Producer发送消息之前，会先向NameServer发出获取消息Topic的路由信息的请求
NameServer返回该Topic的路由表及Broker列表
路由表：实际是一个Map, key为Topic名称，value是一个QueueData实例列表
QueueData并不是一个Queue, 对应一个QueueData, 而是一个Broker中该Topic的所有Queue对应一个QueueData即，只要涉及到该Topic的Broker，一个Broker对应一 个QueueData，QueueData中包含BrokerName
简单来说，路由表的key为Topic名称，value则为所有涉及该Topic的BrokerName列表
Broker列表：实际上也是一个Map，key为BrokerName，value是BrokerData
一个BrokerData对应了一套BrokerName相同的Master-Slave的小集群
BrokerData中包含了一个BrokerName和一个map
map的key为BrokerId（0-master，1-slave），value为broker的地址
Producer根据代码中指定的Queue选择策略，从Queue列表中选出一个队列，用于后续存储消息
Produer对消息做一些特殊处理，例如，消息本身超过4M,则会对其进行压缩
Producer向选择出的Queue所在的Broker发出RPC请求，将消息发送到选择出的Queue
有序消息可以指定queue，无序消息则由系统自己选择
对于无序消息，其Queue选择算法也称为消息投递算法，常见的有两种：
轮询算法：
默认选择算法。该算法保证了每个Queue中可以均匀的获取到消息
该算法存在一个问题:由于某些原因，在某些Broker. 上的Queue可能投递延迟较严重
从而导致Producer的缓存队列中出现较大的消息积压，影响消息的投递性能
最小投递延迟算法：
该算法会统计每次消息投递的时间延迟，然后根据统计出的结果将消息投递到时间延迟最小的Queue
如果延迟相同，则采用轮询算法投递。
如果存在一个queue一直延迟最低，则所有消息都会堆积到该queue中，分配不均，导致压力暴增
RocketMQ中的消息存储在本地文件系统中，这些相关文件默认在当前用户主目录下的store目录中。
RocketMQ中，无论是消息本身还是消息索引，都是存储在磁盘上的。其不会影响消息的消费吗?
当然不会。其实RocketMQ的性能在目前的MQ产品中性能是非常高的。因为系统通过一系列相关机制大大提升了性能。
首先，RocketMQ对文件的读写操作是通过mmap零拷贝进行的，将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率。
其次，consumequeue中的数据是顺序存放的，还引入了PageCache的预读取机制，使得对consumequeue文件的读取几乎接近于内存读取，即使在有消息堆积情况下也不会影响性能。
RocketMQ中可能会影响性能的是对commitlog文件的读取。因为对commitlog文件来说， 读取消息时会产生大量的随机访问，而随机访问会严重影响性能。
不过，如果选择合适的系统IO调度算法，比如设置调度算法为Deadline (采用SSD固态硬盘的话)，随机读的性能也会有所提升。
除了通过通常的指定Topic进行消息消费外，RocketMQ还提供了根据key进行消息查询的功能。
该查询是通过store目录中的index子目录中的indexFile进行索引实现的快速查询。
当然，这个indexFile中的索引数据是在包含了key的消息被发送到Broker时写入的。如果消息中没有包含key,则不会写入。
每个Broker中会包含一组indexFile, 每个indexFile都是以一 个时间戳命名的(这个indexFile被创建时的时间戳)。每个indexFile文件由三部分构成：
每个indexFile文件中包含500w个slot槽。而每个slot槽又可能会挂载很多的index索引单元。
indexHeader固定40个字节，其中存放着如下数据：
indexFile中最复杂的是Slots与Indexes间的关系。在实际存储时，Indexes 是在Slots后面的，但为了便于理解，将它们的关系展示为如下形式:
消费者从Broker中获取消息的方式有两种：
消费者组对于消息消费的模式又分为两种：
pull拉取模式：
Consumer主动从Broker中拉取消息，主动权由Consumer控制。一旦获取了批量消息，就会启动消费过程
不过，该方式的实时性较弱，即Broker中有了新的消息时消费者并不能及时发现并消费
push推送模式：
该模式下Broker收到数据后会主动推送给Consumer。该消费模式一般实时性较高。
该消费类型是典型的发布-订阅模式，即Consumer向其关联的Queue注册了监听器，一旦发现有新的消息到来就会触发回调的执行，回调方法是Consumer去Queue中拉取消息。
而这些都是基于Consumer与Broker间的长连接的。长连接的维护是需要消耗系统资源的。
对比：
广播消费
广播消费模式下，相同Consumer Group的每个Consumer实例都接收同一个Topic的全量消息
即：每条消息都会被发送到Consumer Group中的==每个==Consumer
集群消费
集群消费模式下，相同Consumer Group的每个Consumer实例平均分摊同一个Topic的消息。
即：每条消息只会被发送到Consumer Group中的==某个==Consumer
广播和集群模式的对比
广播模式：
集群模式：
什么是Rebalance：
Rebalance即再均衡，指的是，将一个Topic 下的多个Queue在同一个Consumer Group中的多个Consumer间进行重新分配的过程
Rebalance机制的本意是为了提升消息的并行消费能力
例如，一个Topic下5个队列，在只有1个消费者的情况下，这个消费者将负责消费这5个队列的消息。如果此时我们增加一一个消费者，那么就可以给其中一个消费者分配2个队列，给另一个分配3个队列，从而提升消息的并行消费能力
rebalance的限制：
由于一个队列最多分配给一个消费者， 因此当某个消费者组下的消费者实例数量大于队列的数量时，多余的消费者实例将分配不到任何队列
rebalance的缺点：
消费暂停：
消费重复：
Consumer 在消费新分配给自己的队列时，必须接着之前Consumer提交的消费进度的offset继续消费。然而默认情况下，offset是异步提交的，这个异步性导致提交到Broker的offset与Consumer实际消费的消息并不一致。这个不一致的差值就是可能会重复消费消息。
同步提交: consumer提交了其消费完毕的一批消息的offset给broker后， 需要等待broker的成功ACK。当收到ACK后，consumer才会继续获取并消费下一批消息。在等待ACK期间，consumer是阻塞的。 异步提交: consumer提交了其消费完毕的一批消息的offset给broker后， 不需要等待broker的成功ACK。consumer 可以直接获取并消费下一批消息
消费突刺：
Rebalance产生的原因 导致Rebalance产生的原因，无非就两个：
Rebalance过程
在Broker中维护着多个Map集合，这些集合中动态存放着当前Topic中Queue的信息、Consumer Group中Consumer实例的信息。
一旦发现消费者所订阅的Queue数量发生变化，或消费者组中消费者的数量发生变化，立即向Consumer Group中的每个实例发出Rebalance通知
TopicConfigManager: key是topic 名称，value是TopicConfig. TopicConfig 中维护着该Topic中所有Queue的数据。 ConsumerManager: key是Consumer Group Id, value 是ConsumerGroupInfo。ConsumerGroupInfo中维护着该Group中所有Consumer实例数据。
ConsumeroffsetManager: key为 Topic与订阅该Topic c的Group的组合，value是一 个内层Map。内层Map的key为QueueId,内层Map的value 为该Queue的消费进度offset。
Consumer实例在接收到通知后会采用Queue分配算法自己获取到相应的Queue,即由Consumer实例自主进行Rebalance
一个Topic中的Queue只能由Consumer Group中的一个Consumer进行消费，而一个Consumer可以同时消费多个Queue中的消息。那么Queue与Consumer间的配对关系是如何确定的
即Queue要分配给哪个Consumer进行消费，也是有算法策略的。
常见的有四种策略，这些策略是通过在创建Consumer时的构造器传进去的。
平均分配策略
该算法是要根据avg = QueueCount / ConsumerCount的计算结果进行分配的。如果能够整除，则按顺序将avg个Queue逐个分配Consumer.如果不能整除，则将多余出的Queue按照Consumer顺序逐个分配。
环形平均策略
环形平均算法是指，根据消费者的顺序，依次在由queue队列组成的环形图中逐个分配。
即：直接轮询一个一个来
一致性hash策略
该算法会将consumer的hash值作为Node节点存放到hash环上，然后将queue的hash值也放到hash环上， 通过顺时针方向，距离queue最近的那个consumer就是该queue要分配的consumer。
存在问题：分配不均
同机房策略
该算法会根据queue的部署机房位置和consumer的位置，过滤出当前consumer相同机房的queue
然后按照平均分配策略或环形平均策略对同机房queue进行分配
如果没有同机房queue，则按照平均分配策略或环形平均策略对所有queue进行分配
RocketMQ有一个原则：每条消息必须要被成功消费一次 那么什么是成功消费呢？
Consumer在消费完消息后会向其消费进度记录器提交其消费消息的offset，offset被成功记录到记录器中，那么这条消费就被成功消费了。
什么是消费进度记录器？
对于广播模式来说，Consumer本身就是消费进度记录器
对于集群模式来说，Broker是消费进度记录器
订阅关系的一致性指的是
同一个消费者组(Group ID相同)下所有Consumer实例所订阅的Topic与Tag及对消息的处理逻辑必须完全一致。否则，消息消费的逻辑就会混乱，甚至导致消息丢失
多个消费者组订阅了多个Topic，并且每个消费者组里的多个消费者实例的订阅关系保持了一致
一个消费者组订阅了多个Topic，但是该消费者组里的多个Consumer实例的订阅关系并没有保持一致。
这里的offset指的是Consumer中的消费进度offset
消费进度offset是用来记录每个Queue的不同消费组中每个consumer的消费进度的
根据消费进度记录器的不同，可以分为两种模式：本地模式和远程模式
广播模式下，同消费组的消费者相互独立，消费进度要单独存储
集群模式下，同一条消息只会被同一个消费组消费一次，消费进度会参与到负载均衡中，故消费进度是需要共享的
当消费模式为广播消费时，offset使用本地模式存储。因为每条消息会被所有的消费者消费，每个消费者管理自己的消费进度，各个消费者之间不存在消费进度的交集。
相关数据以json形式，储存在Consumer的本地磁盘下
当消费模式为集群消费时，offset使用远程模式管理。因为所有Consumer实例对消息采用的是均衡消费,所有Consumer共享Queue的消费进度。
相关数据以json形式，储存在Broker的本地磁盘下
当发生Rebalance时， 新的Consumer会从Broker中获取到相应的数据来继续消费
消费者是如何从最开始持续消费消息的?消费者要消费的第一条消息的起始位置是用户自 己通过consumer. setConsumeFromWhere()方法指定的。 在Consumer启动后，其要消费的第一条消息的起始位置常用的有三种，这三种位置可以通过枚举类型常量设置。这个枚举类型为ConsumeFromWhere。
当消费完一-批消息后，Consumer会提交其消费进度offset给Broker, Broker在收到消费进度后会将其更新到那个双层Map及consumerOffset.json文件中，然后向该Consumer进行ACK
而ACK内容中包含三项数据：当前消费队列的最小offset (minOffset) 、最大offset (minOffset) 、及下次消费的起始offset (nextBeginOffset)
CONSUME_FROM_LAST_OFFSET：从queue的当前最后一条消息开始消费 CONSUME_FROM_FIRST_ OFFSET：从queue的第一条消息开始消费 CONSUME_FROM_TIMESTAMP：从指定的具体时间戳位置的消息开始消费
这个具体时间戳是通过另外一个语句指定的。 consumer.setConsumeTimestamp(“20210701080000") yyyMMddHHmmss
当rocketMQ对消息的消费出现异常时,会将发生异常的消息的offset提交到Broker中的重试队列
系统在发生消息消费异常时会为当前的topic@group创建一个重试队列， 该队列以%RETRY%开头，到达重试时间后进行消费重试
集群消费模式下，Consumer消费完消息后会向Broker提交消费进度offset
提交了offset后，需要继续向broker拿取下一批消息的位置nextBeginOffset
其提交方式分为两种
同步提交：
消费者在消费完一批消息后会向broker提交这些消息的offset,然后等待broker的成功响应。若在等待超时之前收到了成功响应，则继续读取下一批消息进行消费。若没有收到响应，则会重新提交，直到获取到响应。而在这个等待过程中，消费者是阻塞的。其严重影响了消费者的吞吐量。
异步提交：
消费者在消费完一-批消息后向broker提交offset ，但无需等待Broker的成功响应，可以继续读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意，broker在收到提交的offset后，还是会向消费者进行响应的。
当出现消费者对某条消息重复消费的情况时，重复消费的结果与消费一次的结果是相同的，并且多次消费并未对业务系统产生任何负面影响，那么这个消费过程就是消费幂等的
幂等：若某操作执行多次与执行次对系统产生的影响是相同的，则称该操作是幂等的。
在互联网应用中，尤其在网络不稳定的情况下，消息很有可能会出现重复发送或重复消费。如果重复的消息可能会影响业务处理，那么就应该对消息做幂等处理
什么情况可能会发生消息的重复消费呢？
发送消息时重复：应答失败导致重复发送
当-条消息已被成功发送到Broker并完成持久化，此时出现了网络闪断，从而导致Broker对Producer应答失败
如果此时Producer意识到消息发送失败并尝试再次发送消息，此时Broker中就可能会出现两条内容相同并且Message ID也相同的消息，那么后续Consumer就一定会消费两次该消息。
消费时消息重复：应答失败导致重复投递消费
消息已投递到Consumer并完成业务处理,当Consumer给Broker反馈应答时网络闪断， Broker没有接收到消费成功响应。为了保证消息至少被消费一次的原则， Broker将在网络恢复后再次尝试投递之前已被处理过的消息。此时消费者就会收到与之前处理过的内容相同、Message ID也相同的消息。
rebalance时消息重复
当Consumer Group中的Consumer数量发生变化时，或其订阅的Topic的Queue数量发生变化时，会触发Rebalance,此时Consumer可能会收到曾经被消费过的消息。
两要素 幂等解决方案的设计中涉及到两项要素：幂等令牌与唯一性处理。
只要充分利用好这两要素，就可以设计出好的幂等解决方案。
解决方案 对于常见的系统，幂等性操作的通用性解决方案是:
因为缓存具有有效期，所以当然需要加一层db的判断
以支付场景举例：
完成支付任务（唯一性处理）
将当前支付流水号作为key,任意字符串作为value,通过set(key, value, expireTime)将数据写入到Redis缓存
将当前支付流水号作为主键，与其它相关数据共同写入到DBMS
消费幂等的解决方案很简单：为消息指定不会重复的唯一标识
因为Message ID有可能出现重复的情况，所以真正安全的幂等处理，不建议以Message ID作为处理依据。
最好的方式是以业务唯一标识作为幂等处理的关键依据，而业务的唯一标识可以通过消息Key设置。
处理消息的时候就根据业务id去进行缓存和db判断即可
消息处理流程中，如果Consumer的消费速度跟不上Producer的发送速度，MQ中未处理的消息会越来越多(进的多出的少)，这部分消息就被称为堆积消息
消息出现堆积，进而会造成消息的消费延迟
以下场景需要重点关注消息堆积和消费延迟问题：
一般只需要解决了消息堆积，就不会有消息延迟了
Consumer通过长轮询Pull模式批量拉取的方式从服务端获取消息，将拉取到的消息缓存到本地缓冲队列中。
对于拉取式消费，在内网环境下会有很高的吞吐量,所以这一阶段一般不会成为消息堆积的瓶颈。
消息消费
Consumer将本地缓存的消息提交到消费线程中，使用业务消费逻辑对消息进行处理，处理完毕后获取到一个结果。这是真正的消息消费过程。此时Consumer的消费能力就完全依赖于消息的消费耗时和消费并发度了
如果由于业务处理逻辑复杂等原因，导致处理单条消息的耗时较长，则整体的消息吞吐量肯定不会高，此时就会导致Consumer本地缓冲队列达到上限，停止从服务端拉取消息
消息堆积的主要瓶颈在于客户端的消费能力，而消费能力由消费耗时和消费并发度决定
优先解决消费耗时的问题（降低处理时间），再考虑消费并发度（增加线程等）
影响消息处理时长的主要因素是代码逻辑
而代码逻辑中可能会影响处理时长的主要有两种类型的代码：CPU内部计算型代码和外部I/O操作型代码
通常情况下代码中如果没有复杂的递归和循环的话，内部计算耗时相对外部I/0操作来说几乎可以忽略
所以外部IO型代码是影响消息处理时长的主要症结所在
一般情况下，消费者端的消费并发度由单节点线程数和节点数量共同决定，其值为单节点线程数节点数量
不过，通常需要优先调整单节点的线程数，若单机硬件资源达到了上限，则需要通过横向扩展来提高消费并发度
消息被消费完后会被清理嘛？不会的
消息是被顺序存储在commitlog文件的，且消息大小不定长，所以消息的清理是不可能以消息为单位进行清理的，而是以commitlog文件 为单位进行清理的。否则会极具下降清理效率，并实现逻辑复杂
commitlog文件存在一个过期时间， 默认为72小时，即三天。
除了用户手动清理外，在以下情况下也会被自动清理，无论文件中的消息是否被消费过:
文件过期，且到达清理时间点(默认为凌晨4点)后，自动清理过期文件
文件过期，且磁盘空间占用率已达过期清理警戒线(默认75%)后，无论是否达到清理时间点，都会自动清理过期文件
磁盘占用率达到清理警戒线(默认85%)后，开始按照设定好的规则清理文件，无论是否过期。默认会从最老的文件开始清理
磁盘占用率达到系统危险警戒线(默认90%)后，Broker将拒绝消息写入
切记切记开启防火墙，9876和10911端口！不然会连接不上
消息发送分类
Producer对于消息的发送也有许多的选择，不同的方法会产生不同的效果
同步发送消息是指，Producer发出一条消息后， 会在收到MQ返回的ACK之后才发下一条消息
该方式的消息可靠性最高，但消息发送效率太低
同步消息发送示例
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
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.exception.RemotingException;
/**
* @Author if
* @Description: 同步生产者
* @Date 2021-10-24 下午 05:52
*/
public class SyncProducer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
//创建一个producer，参数为producerGroup名称
DefaultMQProducer producer=new DefaultMQProducer("pg");
//设置name server地址
producer.setNamesrvAddr("centos:9876");
//启动生产者
producer.start();
//当发送失败时重试发送的次数，默认2次
producer.setRetryTimesWhenSendFailed(3);
//设置发送超时时间5000ms（5s），默认3000ms
producer.setSendMsgTimeout(5000);
//测试生产并发送100条消息
for(int i=0;i<100;i++){
byte[] body=("Hi,"+i).getBytes();
//创建Message，指定topic、tag和具体内容byte[]类型的body
Message msg=new Message("someTopic","someTag",body);
//发送消息并获得result
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
//关闭producer
producer.shutdown();
}
}
SendResult类是一个枚举类，有如下状态
public enum SendStatus {
SEND_OK, //发送成功
FLUSH_DISK_TIMEOUT, //刷盘超时（Broker刷盘策略设置同步刷盘可能产生这个状态）
FLUSH_SLAVE_TIMEOUT, //slave同步超时（Broker集群设置同步复制可能会导致）
SLAVE_NOT_AVAILABLE; //从站slave不可用（Broker集群设置同步复制可能会导致）
private SendStatus() {
}
}
出现下列文字则是消息发送成功sendStatus=SEND_OK
SendResult [sendStatus=SEND_OK, msgId=7F00000112D818B4AAC27A729CB90062, offsetMsgId=C0A8E98000002A9F00000000000667D4, messageQueue=MessageQueue [topic=someTopic, brokerName=centos, queueId=2], queueOffset=49]
异步发送消息是指，Producer发出消息后无需等待(异步发送响应)MQ返回ACK，直接发送下一条消息
该方式的消息可靠性可以得到保障，消息发送效率也可以
异步和同步只是发送时选用的方法不同public void send(Message msg, SendCallback sendCallback)
而且发送的时候需要TimeUnit.SECONDS.sleep(3);一下，不能让他立即关闭了
需要一个回调类SendCallback
public interface SendCallback {
void onSuccess(SendResult var1);
void onException(Throwable var1);
}
异步发送消息代码示例
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
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;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.concurrent.TimeUnit;
/**
* @Author if
* @Description: 异步生产者
* @Date 2021-10-24 下午 05:52
*/
public class AsyncProducer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
//创建一个producer，参数为producerGroup名称
DefaultMQProducer producer=new DefaultMQProducer("pg");
//设置name server地址
producer.setNamesrvAddr("centos:9876");
//启动生产者
producer.start();
//设置异步发送不重试
producer.setRetryTimesWhenSendAsyncFailed(0);
//设置发送超时时间5000ms（5s），默认3000ms
producer.setSendMsgTimeout(5000);
//设定新创建的topic的queue数量为2，默认4
producer.setDefaultTopicQueueNums(2);
//测试生产并发送100条消息
for(int i=0;i<100;i++){
byte[] body=("Hi,"+i).getBytes();
//创建Message，指定topic、tag和具体内容byte[]类型的body
Message msg=new Message("AsyncTopic","AsyncTopic",body);
//发送消息并获得result
producer.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println(sendResult);
}
@Override
public void onException(Throwable throwable) {
throwable.printStackTrace();
}
});
}
//这一步的休眠很重要！因为是异步发送，还没发送完就关闭producer了
TimeUnit.SECONDS.sleep(3);
//关闭producer
producer.shutdown();
}
}
单向发送消息是指，Producer仅负责发送消息， 不等待、不处理MQ的ACK。该发送方式时MQ也不返回ACK
该方式的消息发送效率最高，但消息可靠性较差
单向发送消息代码示例
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
/**
* @Author if
* @Description: 单向消息生产者
* @Date 2021-10-24 下午 07:21
*/
public class OneWayProducer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException {
DefaultMQProducer producer=new DefaultMQProducer("pg");
producer.setNamesrvAddr("centos:9876");
producer.start();
for(int i=0;i<100;i++){
byte[] body=("Hi,"+i).getBytes();
Message msg=new Message("OneWayTopic","OneWayTopic",body);
//单向发送
producer.sendOneway(msg);
}
producer.shutdown();
System.out.println("producer shutdown");
}
}
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
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.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import java.util.List;
/**
* @Author if
* @Description: 消费者
* @Date 2021-10-24 下午 07:25
*/
public class SomeConsumer {
public static void main(String[] args) throws MQClientException {
//定义一个Pull消费者
// DefaultMQPullConsumer pullConsumer=new DefaultMQPullConsumer("cg");
//定义一个Push消费者
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("cg");
consumer.setNamesrvAddr("centos:9876");
//指定从第一条消息开始消费
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//订阅消费的topic与tag（此处tag为通配符表示所有）
consumer.subscribe("someTopic","*");
//指定消费模式：广播模式，默认为集群模式
consumer.setMessageModel(MessageModel.BROADCASTING);
//注册消息监听器
consumer.registerMessageListener(new MessageListenerConcurrently() {
//消费消息代码，一旦broker中有了其订阅的消息就会触发该代码的执行
//返回值为当前消费的状态
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt messageExt : list) {
System.out.println(messageExt);
}
//CONSUME_SUCCESS消费成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//开启消费
consumer.start();
System.out.println("consumer start");
}
}
结果
MessageExt [brokerName=centos, queueId=1, storeSize=182, queueOffset=43, sysFlag=0, bornTimestamp=1635071931511, bornHost=/192.168.233.1:53685, storeTimestamp=1635071929343, storeHost=/192.168.233.128:10911, msgId=C0A8E98000002A9F000000000006560E, commitLogOffset=415246, bodyCRC=1980703418, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='someTopic', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=75, CONSUME_START_TIME=1635075197904, UNIQ_KEY=7F00000112D818B4AAC27A729C770049, CLUSTER=DefaultCluster, TAGS=someTag}, body=[72, 105, 44, 55, 51], transactionId='null'}]
我们清楚控制台内容，不关闭消费者，然后开启同步生产者生成消息发送，消费者就会处理刚刚生产的消息
我们可以rocket-console看到所以运行的消费者及其信息状态
顺序消息指的是，严格按照消息的发送顺序进行消费的消息
默认情况下生产者会把消息以Round Robin轮询方式发送到不同的Queue分区队列
而消费消息时会从多个Queue上拉取消息，这种情况下的发送和消费是不能保证顺序的
如果将消息仅发送到同一个Queue中，消费时也只从这个Queue上拉取消息，就严格保证了消息的顺序性
用例子说明：假设我们一个订单topic，有4个queue队列，该topic中不同的消息用于描述订单不同的状态
假设订单有4个状态：未支付、已支付、发货中、发货成功、发货失败
根据以上订单状态，生产者从时序上可以生成如下几个消息:
订单T0000001:未支付-->订单T0000001:已支付-->订单T0000001:发货中-->订单T0000001:发货失败
消息发送到MQ中之后，Queue的选择如果采用轮询策略，消息在MQ的存储可能如下:
这种情况下，我们希望Consumer消费消息的顺序和我们发送是一致的， 然而上述MQ的投递和消费方式，我们无法保证顺序是正确的
对于顺序异常的消息，Consumer即使设置有一定的状态容错，也不能完全处理好这么多种随机出现组合情况
基于上述的情况，可以设计如下方案:对于相同订单号的消息，通过-定的策略，将其放置在一个Queue中，然后消费者再采用一定的策略(例如，一个线程独立处理一个queue, 保证处理消息的顺序性)，能够保证消费的顺序性。
根据有序范围的不同，RocketMQ可以严格地保证两种消息的有序性：分区有序与全局有序
分区有序
如果有多个Queue参与，其仅可保证在该Queue分区队列上的消息顺序，则称为分区有序
也就是一个订单选用一个queue
如何实现Queue的选择?
在定Producer时我们可以指定消息队列选择器，而这个选择器是我们自己实现了MessageQueueSelector接口定义的。
在定义选择器的选择算法时，一般需要使用选择key。 这个选择key可以是消息key也可以是其它数据。但无论谁做选择key，都不能重复，都是唯一的，一般可以用业务id，即例子中的订单id
一般性的选择算法是， 让选择key（或其hash值）与该Topic所包含的Queue的数量取模，其结果即为选择出的Queue的Queueld
取模算法存在一个问题：
不同选择key与Queue数量取模结果可能会是相同的，即不同选择key的消息可能会出现在相同的Queue,即同一个Consumer可能会消费到不同选择key的消息。
这个问题如何解决?
一般性的作法是，从消息中获取到选择key，对其进行判断。
若是当前Consumer需要消费的消息，则直接消费，否则，什么也不做。这种做法要求选择key要能够随着消息一起被Consumer获取到。此时使用消息key作为选择key是比较好的做法
以上做法会不会出现如下新的问题呢?
不属于那个Consumer的消息被拉取走了，那么应该消费该消息的Consumer是否还能再消费该消息呢?同一个Queue中的消息不可能被同一个Group中的不同Cosuner同时消费所以，消费现一个Queue的不同选择key的消息的Consumer一定属于不同的Group。 而不同的Group中的Consumer间的消费是相互隔离的，互不影响的。
分区有序代码示例：两种选择算法
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.List;
/**
* @Author if
* @Description: 顺序消息生产者
* @Date 2021-10-24 下午 08:25
*/
public class OrderedProducer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer producer=new DefaultMQProducer("pg");
producer.setNamesrvAddr("centos:9876");
producer.start();
for(int i=0;i<100;i++){
//oderId表示订单id
int orderId=i;
byte[] body=("Hi,"+i).getBytes();
Message msg=new Message("OrderedTopic","OrderedTag",body);
//将oderId当做消息key并充当选择算法中的选择key
msg.setKeys(orderId+"");
//send()方法的第三个参数Object arg我们这使用业务id，即订单id
//这个arg会传递给选择器
//该send为同步发送，当然也可以加一个SendCallback将其变为异步发送
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
//订单选择器的选择算法
@Override
public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
//以下是使用消息key作为选择key的选择算法
String key=message.getKeys();
int id=Integer.parseInt(key);
//以下是使用arg作为选择key
// int id=(int) o;
int index=id % list.size();
return list.get(index);
}
},orderId);
System.out.println(sendResult);
}
producer.shutdown();
}
}
全局有序
当发送和消费参与的Queue只有一个时所保证的有序是整个Topic中消息的顺序，称为全局有序
3个方式指定topic的queue数量
producer发送时可以指定数量producer.setDefaultTopicQueueNums(2);
可视化控制台里手动创建topic时指定queue数量
mqadmin命令手动创建topic时指定queue数量
全局有序的实现就指定数量就行了
当消息写入到Broker后，在指定的时长后才可被消费处理的消息，称为延时消息
采用RocketMQ的延时消息可以实现定时任务的功能，而无需使用定时器
典型的应用场景是，电商交易中超时未支付关闭订单的场景，12306平台订票超时未支付取消订票的场景。
延时消息的延迟时长不支持随意时长的延迟，是通过特定的延迟等级来指定的
延时等级定义在RocketMQ服务端的MessageStoreConfig类中的如下变量中：
若指定延时等级为3，则表示延时为10s（从1开始计数）
如果需要自定义延时等级，可以去conf里修改broker的配置
messageDelayLevel=1s 5s10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 1d
修改消息
Producer将消息发送到Broker后，Broker会 首先将消息写入到commitlog文件，然后需要将其分发到相应的consumequeue。
不过，在分发之前，系统会先判断消息中是否带有延时等级msg.setDelayTimeLevel(3);
若没有，则直接正常分发
若有，则需要经历一个复杂的过程：
投递消息服务类
Broker内部有一个延迟消息服务类ScheuleMessageService,其会消费SCHEDULE TOPIC XXXX中的消息，即按照每条消息的投递时间，将延时消息投递到目标Topic中。
不过，在投递之前会从commitlog中将原来写入的消息再次读出，并将其原来的延时等级设置为0，即原消息变为了一条不延迟的普通消息。然后再次将消息投递到目标Topic中。
将消息重新写入commitlog 延迟消息服务类将延迟消息再次发送给了commitlog,并再次形成新的消息索引条目，分发到相应Queue
判断有无延迟等级->有则修改原topic为SCHEDULE _TOPIC并填入时间->时间到了将其变为普通消息并投到原topic
DelayConsumer表示消费者，创建并运行
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.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
/**
* @Author if
* @Description: What is it
* @Date 2021-10-25 下午 05:32
*/
public class DelayConsumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("cg");
consumer.setNamesrvAddr("centos:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("DelayTopic","*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt messageExt : list) {
//输出消费时的时间
System.out.println(new SimpleDateFormat("mm:ss").format(new Date()));
System.out.println(messageExt);
}
//CONSUME_SUCCESS消费成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("consumer start");
}
}
DelayProducer为延时消息生产者，创建并运行
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
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.exception.RemotingException;
import java.text.SimpleDateFormat;
import java.util.Date;
/**
* @Author if
* @Description: What is it
* @Date 2021-10-25 下午 05:29
*/
public class DelayProducer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("centos:9876");
producer.start();
for(int i=0;i<100;i++){
byte[] body=("Hi,"+i).getBytes();
Message msg=new Message("DelayTopic","DelayTag",body);
//指定消息延迟等级为3，即10s
msg.setDelayTimeLevel(3);
SendResult sendResult = producer.send(msg);
//输出消息发送的时间
System.out.println(new SimpleDateFormat("mm:ss").format(new Date()));
//输出消息返回结果
System.out.println(" , "+sendResult);
}
producer.shutdown();
}
}
查看输出结果，举例
先是生产者发送消息，可以看见时间为34分46秒
34:46 , SendResult [sendStatus=SEND_OK, msgId=7F000001258818B4AAC27F5E4D880063, offsetMsgId=C0A8E98000002A9F000000000008342E, messageQueue=MessageQueue [topic=DelayTopic, brokerName=centos, queueId=0], queueOffset=99]
然后是消费者消费消息，这时已经是延时结束了，时间为35分16秒
35:16 MessageExt [brokerName=centos, queueId=2, storeSize=235, queueOffset=4, sysFlag=0, bornTimestamp=1635154486499, bornHost=/192.168.233.1:5949, storeTimestamp=1635154495928, storeHost=/192.168.233.128:10911, msgId=C0A8E98000002A9F00000000000844B3, commitLogOffset=541875, bodyCRC=657998117, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='DelayTopic', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=DelayTopic, MAX_OFFSET=25, CONSUME_START_TIME=1635154516164, UNIQ_KEY=7F000001258818B4AAC27F5E4CE30011, CLUSTER=DefaultCluster, DELAY=3, WAIT=true, TAGS=DelayTag, REAL_QID=2}, body=[72, 105, 44, 49, 55], transactionId='null'}]
这里的一个需求场景是：工行用户A向建行用户B转账1万元
我们可以使用同步消息来处理该需求场景：
如果扣款失败，可是消息已经到了mq，只要消息写入成功就可以被消费，就会出现问题
解决思路是，让第1、2、3步具有原子性，要么全部成功，要么全部失败
即消息发送成功后，必须要保证扣款成功。如果扣款失败，则回滚发送成功的消息。而该思路即使用事务消息
这里要使用分布式事务解决方案
分布式事务
分布式事务通俗地说就是，一次操作由若干分支操作组成，这些分支操作分属不同应用，分布在不同服务器上
分布式事务需要保证这些分支操作要么全部成功，要么全部失败
分布式事务与普通事务一样，就是为了保证操作结果的一致性
事务消息
RocketMQ提供了类似X/Open XA的分布式事务功能，通过事务消息能达到分布式事务的最终一致
XA是一种分布式事务解决方案，一 种分布式事务处理模式
半事务消息
暂不能投递的消息，发送方已经成功地将消息发送到了Broker,但是Broker未收到最终确认指令，此时该消息被标记成‘暂不能投递”状态，即不能被消费者看到
处于该种状态下的消息即半事务消息
本地事务状态
Producer回调操作执行的结果为本地事务状态，其会发送给TC,而TC会再发送给TM
TM会根据TC发送来的本地事务状态来决定全局事务确认指令
消息回查
消息回查，即重新查询本地事务的执行状态。本例就是重新到DB中查看预扣款操作是否执行成功
常见的两个引发消息回查的原因：
- 回调操作返回UNKNWON
- TC没有接收到TM的最终全局事务确认指令
XA协议
XA (Unix Transaction)是一种分布式事务解决方案，一种分布式事务处理模式，是基于XA协议的
XA协议由Tuxedo (Transaction for Unix has been Extended for Distributed Operation,分布式操作扩展之后的Unix事务系统)首先提出的，并交给X/Open组织，作为资源管理器与事务管理器的接口标准
XA模式中有三个重要组件: TC、TM、RM
TC
事务协调者，Transaction Coordinator
维护全局和分支事务的状态，驱动全局事务提交或回滚
RocketMQ中Broker充当着TC事务协调者
TM
事务管理器，Transaction Manager
定义全局事务的范围：开始全局事务、提交或回滚全局事务。它实际是全局事务的发起者
RocketMQ中Producer充当着TM事务管理器
RM
资源管理器，Resource Manager
管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚
RocketMQ中Broker和Producer均充当着RM资源管理器
创建事务监听器IcbcTransactionListener类
import org.apache.commons.lang3.StringUtils;
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;
/**
* @Author if
* @Description: 爱存不存事务监听器
* @Date 2021-10-25 下午 09:56
*/
public class IcbcTransactionListener implements TransactionListener {
/**
* 回调操作
* 消息预提交成功就会触发该方法的执行，用于执行本地事务
* 这里模拟一下
* tagA就是扣款成功
* tagB就是扣款失败
* tagC就是不知道成没成功，要执行消息回查
* @param message
* @param o
* @return
*/
@Override
public LocalTransactionState executeLocalTransaction(Message message, Object o) {
System.out.println("预提交消息成功："+message);
if(StringUtils.equals("TAGA",message.getTags())){
return LocalTransactionState.COMMIT_MESSAGE;
}else if(StringUtils.equals("TAGB",message.getTags())){
return LocalTransactionState.ROLLBACK_MESSAGE;
}else if(StringUtils.equals("TAGC",message.getTags())){
return LocalTransactionState.UNKNOW;
}
return LocalTransactionState.UNKNOW;
}
/**
* 消息回查方法
* 这里模拟出来表示回查就返回成功状态
*
* 常见的两个引发消息回查的原因：
* 1. 回调操作返回UNKNWON
* 2. TC没有接收到TM的最终全局事务确认指令
* @param messageExt
* @return
*/
@Override
public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
System.out.println("执行消息回查："+messageExt.getTags());
return LocalTransactionState.COMMIT_MESSAGE;
}
}
创建生产者TransactionProducer类
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @Author if
* @Description: What is it
* @Date 2021-10-25 下午 09:44
*/
public class TransactionProducer {
public static void main(String[] args) throws MQClientException {
TransactionMQProducer producer=new TransactionMQProducer("tpg");
producer.setNamesrvAddr("centos:9876");
/**
* 定义一个线程池
* @param corePoolSize，核心线程数
* @param maximumPoolSize，最多线程数
* @param keepAliveTime，保持活跃的时间：当线程池中线程数量大于核心线程数量时，多余空闲线程的存活时长
* @param unit，时间单位
* @param workQueue，临时存放任务的工作队列
* @param threadFactory，线程工厂
*/
ThreadPoolExecutor executorService = new ThreadPoolExecutor(
2,
5,
100,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(200),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("client-transaction-msg-check-thread");
return thread;
}
});
//为生产者指定刚刚定义的线程池
producer.setExecutorService(executorService);
//为生产者指定任务监听器
producer.setTransactionListener(new IcbcTransactionListener());
producer.start();
String[] tags={"TAGA","TAGB","TAGC"};
for(int i=0;i<3;i++){
byte[] body=("Hi"+i).getBytes();
Message msg=new Message("TransactionTopic",tags[i],body);
//发送事务消息
//第二个参数用于指定在执行本地事务时需要使用的业务参数
TransactionSendResult result = producer.sendMessageInTransaction(msg, null);
System.out.println("发送结果为："+result.getSendStatus());
}
}
}
启动生产者得到结果
预提交消息成功：Message{topic='TransactionTopic', flag=0, properties={TRAN_MSG=true, UNIQ_KEY=7F0000011DC018B4AAC280588D660000, WAIT=true, PGROUP=tpg, TAGS=TAGA}, body=[72, 105, 48], transactionId='7F0000011DC018B4AAC280588D660000'} 发送结果为：SEND_OK
预提交消息成功：Message{topic='TransactionTopic', flag=0, properties={TRAN_MSG=true, UNIQ_KEY=7F0000011DC018B4AAC280588DF80001, WAIT=true, PGROUP=tpg, TAGS=TAGB}, body=[72, 105, 49], transactionId='7F0000011DC018B4AAC280588DF80001'} 发送结果为：SEND_OK
预提交消息成功：Message{topic='TransactionTopic', flag=0, properties={TRAN_MSG=true, UNIQ_KEY=7F0000011DC018B4AAC280588E210002, WAIT=true, PGROUP=tpg, TAGS=TAGC}, body=[72, 105, 50], transactionId='7F0000011DC018B4AAC280588E210002'} 发送结果为：SEND_OK 执行消息回查：TAGC
可以看到TAGC时，监听器返回了LocalTransactionState.UNKNOW表示不知道成没成功，所以执行了消息回查方法
此时我们使用消费者去消费TransactionTopic消息
consumer start
MessageExt [brokerName=centos, queueId=0, storeSize=284, queueOffset=0, sysFlag=8, bornTimestamp=1635170887201, bornHost=/192.168.233.1:18864, storeTimestamp=1635170920413, storeHost=/192.168.233.128:10911, msgId=C0A8E98000002A9F0000000000089736, commitLogOffset=562998, bodyCRC=2099282948, reconsumeTimes=0, preparedTransactionOffset=562706, toString()=Message{topic='TransactionTopic', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TransactionTopic, TRANSACTION_CHECK_TIMES=1, MAX_OFFSET=1, TRAN_MSG=true, CONSUME_START_TIME=1635171216788, UNIQ_KEY=7F0000011DC018B4AAC280588E210002, CLUSTER=DefaultCluster, PGROUP=tpg, WAIT=true, TAGS=TAGC, REAL_QID=0}, body=[72, 105, 50], transactionId='7F0000011DC018B4AAC280588E210002'}]
MessageExt [brokerName=centos, queueId=2, storeSize=258, queueOffset=0, sysFlag=8, bornTimestamp=1635170887015, bornHost=/192.168.233.1:18864, storeTimestamp=1635170886680, storeHost=/192.168.233.128:10911, msgId=C0A8E98000002A9F0000000000089402, commitLogOffset=562178, bodyCRC=321840424, reconsumeTimes=0, preparedTransactionOffset=561380, toString()=Message{topic='TransactionTopic', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TransactionTopic, MAX_OFFSET=1, TRAN_MSG=true, CONSUME_START_TIME=1635171216902, UNIQ_KEY=7F0000011DC018B4AAC280588D660000, CLUSTER=DefaultCluster, PGROUP=tpg, WAIT=true, TAGS=TAGA, REAL_QID=2}, body=[72, 105, 48], transactionId='7F0000011DC018B4AAC280588D660000'}]
可以看见只消费了正常的TAGA和执行了消息回查的TAGC，而TAGB因为在监听器里触发时返回了失败状态，所以在整体回滚之后，消息就没有被提交到broker，自然就没有被消费者消费
查看DefaultMQPushConsumer生产者类可以看见，send消息不仅是参数可以为单条Message，也可以说集合类型
public SendResult send(Collection<Message> msgs)
发送限制
生产者进行消息发送时可以一次发送多条消息，这可以大大提升Producer的发送效率
不过需要注意以下几点:
批量发送大小
默认情况下，一批发送的消息总大小不能超过4MB字节。如果想超出该值，有两种解决方案：
生产者通过send()方法发送的Message,并不是直接将Message序列化后发送到网络上的，而是通过这个Message生成了一个字符串发送出去的
这个字符串由四部分构成：Topic、 消息Body、消息日志(占20字节)，及用于描述消息的一堆属性key-value
这些属性中包含例如生产者地址、生产时间、要发送的Queueld等。最终写入到Broker中消息 单元中的数据都是来自于这些属性
Consumer的MessageListenerConcurrently监听接口的consumeMessage(方法的第一个参数为消息列表，但默认情况下每次只能消费一条消息。
//注册消息监听器
consumer.registerMessageListener(new MessageListenerConcurrently() {
//消费消息代码，一旦broker中有了其订阅的消息就会触发该代码的执行
//返回值为当前消费的状态
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt messageExt : list) {
System.out.println(messageExt);
}
//CONSUME_SUCCESS消费成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
若要使其一次可以消费多条消息， 则可以通过修改Consumer的consumeMessageBatchMaxSize属性来指定
consumer.setConsumeMessageBatchMaxSize(10);
默认值为1，,该值不能超过32，即消费者每次可以拉取的消息最多是32条
/**
* Batch consumption size 批量消费规模
*/
private int consumeMessageBatchMaxSize = 1;
若要修改一次拉取的最大值，则可通过修改Consumer的pullBatchSize属性来指定
consumer.setPullBatchSize(32);
默认值为32
/**
* Batch pull size 批量拉取大小
*/
private int pullBatchSize = 32;
存在的问题
Consumer的pullBatchSize属性与consumeMessageBatchMaxSize属性是否设置的越大越好？当然不是
消息者在进行消息订阅时，除了可以指定要订阅消息的Topic外，还可以对指定Topic中的消息根据指定条件进行过滤，即可以订阅比Topic更加细粒度的消息类型
对于指定Topic消息的过滤有两种过滤方式：Tag过滤与SQL过滤
通过consumer的subscribe()方法指定要订阅消息的Tag.如果订阅多个Tag的消息，Tag间使用或运算符双竖线||连接。
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("cg");
consumer.subscribe("Topic","TAGA || TAGB || TAGC");
SQL过滤是一种通过特定表达式对事先埋入到消息中的用户属性进行筛选过滤的方式。
通过SQL过滤，可以实现对消息的复杂过滤。
不过，只有使用PUSH模式的消费者才能使用SQL过滤。
SQL过滤表达式中支持多种常量类型与运算符。
支持的常量类型：
支持的运算符：
先来个Tag过滤
定义消费者FilterConsumer
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.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import java.util.List;
/**
* @Author if
* @Description: What is it
* @Date 2021-10-25 下午 10:43
*/
public class FilterConsumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("cg");
consumer.setNamesrvAddr("centos:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//消费FilterTopic中的TagA和TagB（TagC过滤）
consumer.subscribe("FilterTopic","TagA || TagB");
consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt messageExt : list) {
System.out.println("消费的tag为："+messageExt.getTags());
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("consumer start");
}
}
定义生产者FilterProducer
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
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.exception.RemotingException;
/**
* @Author if
* @Description: What is it
* @Date 2021-10-25 下午 10:43
*/
public class FilterProducer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer producer=new DefaultMQProducer("pg");
producer.setNamesrvAddr("centos:9876");
producer.start();
//创建3中tag，让消费者过滤掉TagC
String[] tags={"TagA","TagB","TagC"};
for(int i=0;i<10;i++){
byte[] body=("Hi,"+i).getBytes();
String tag=tags[i % tags.length];
Message msg=new Message("FilterTopic",tag,body);
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
producer.shutdown();
}
}
可以看见消费者消费的结果为
consumer start 消费的tag为：TagB 消费的tag为：TagB 消费的tag为：TagB 消费的tag为：TagA 消费的tag为：TagA 消费的tag为：TagA 消费的tag为：TagA
也就是说通过consumer.subscribe("FilterTopic","TagA || TagB");将TagC过滤掉了（只消费A和B）
再来个SQL过滤
创建消费者SQLFilterConsumer
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.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import java.util.List;
/**
* @Author if
* @Description: What is it
* @Date 2021-10-25 下午 10:43
*/
public class SQLFilterConsumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("cg");
consumer.setNamesrvAddr("centos:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//消费SQLFilterTopic，根据MessageSelector.bySql()方法过滤出age在0到6的消息
consumer.subscribe("SQLFilterTopic", MessageSelector.bySql("age between 0 and 6"));
consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt messageExt : list) {
System.out.println("消费的用户属性为："+messageExt.getProperty("age"));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("consumer start");
}
}
创建生产者SQLFilterProducer
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
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.exception.RemotingException;
/**
* @Author if
* @Description: What is it
* @Date 2021-10-25 下午 10:43
*/
public class SQLFilterProducer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer producer=new DefaultMQProducer("pg");
producer.setNamesrvAddr("centos:9876");
producer.start();
for(int i=0;i<10;i++){
byte[] body=("Hi,"+i).getBytes();
Message msg=new Message("SQLFilterTopic","SQLFilterTopic",body);
//在message中put用户属性
msg.putUserProperty("age",i + "");
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
producer.shutdown();
}
}
查看结果
consumer start 消费的用户属性为：5 消费的用户属性为：6 消费的用户属性为：0 消费的用户属性为：1 消费的用户属性为：4 消费的用户属性为：3 消费的用户属性为：2
证明sql过滤MessageSelector.bySql("age between 0 and 6")成功
如果运行报错
CODE: 1 DESC: The broker does not support consumer to filter message by SQL9
可能是broker不支持属性过滤，需要在conf中修改broker.conf文件，加上enablePropertyFilter=true ，且在运行broker时用-c命令指定conf文件，即可解决
Producer对发送失败的消息进行重新发送的机制，称为消息发送重试机制，也称为消息重投机制
对于消息重投，需要注意以下几点：
对于普通消息，消息发送默认采用轮询round robin策略来选择所发送到的队列。
如果发送失败，默认重试2次。但在重试时是不会选择上次发送失败的Broker，而是选择其它Broker。
//当发送失败时重试发送的次数，默认2次
producer.setRetryTimesWhenSendFailed(3);
同时，Broker还具有失败隔离功能，使Producer尽量选择未发生过发送失败的Broker作为目标Broker。
如果超过重试次数，则抛出异常，由Producer去保证消息不丢。 当然当生产者出现RemotingException、MQClientException和MQBrokerException时，Producer会自动重投消息。
异步发送失败重试时，异步重试不会选择其他broker, 仅在同一个broker上做重试，所以该策略无法保证消息不丢。
//设置异步发送不重试
producer.setRetryTimesWhenSendAsyncFailed(0);
消息刷盘超时（master或slave不可用）或slave不可用(返回状态非SEND OK)时，默认是不会将消息尝试发送到其他Broker的。
不过，对于重要消息可以通过在Broker的配置文件设置retryAnotherBrokerWhenNotStoreOK属性为true来开启。
对于顺序消息，当Consumer消费消息失败后，为了保证消息的顺序性，其会自动不断地进行消息重试，直到消费成功。重试期间应用会出现消息消费被阻塞的情况。
由于对顺序消息的重试是无休止的，不间断的，直至消费成功，所以对于顺序消息的消费，务必要保证应用能够及时监控并处理消费失败的情况，避免消费被永久性阻塞
//顺序消息消费失败的消费重试时间间隔，默认为1000亳秒，其取值范围为[10,30000]毫秒
consumer.setSuspendCurrentQueueTimeMillis(100);
对于无序消息，当Consumer消费消息失败时，可以通过设置返回状态达到消息重试的效果。不过需要注意，无序消息的重试只对集群消费方式生效，广播消费方式不提供失败重试特性
即对于广播消费，消费失败后，失败消息不再重试，继续消费后续消息。
对于无序消息集群消费下的重试消费，每条消息默认最多重试16次
//设定消息重试最多10次
consumer.setMaxReconsumeTimes(10);
对于这个自定义次数：
若修改值小于16，则按照指定间隔进行重试 若修改值大于16，则超过16次的重试时间间隔均为2小时
对于Consumer Group,若仅修改了-个Consumer的消费重试次数，则会应用到该Group中所有其它Consumer实例
若出现多个Consumer均做了修改的情况，则采用覆盖方式生效。即最后被修改的值会覆盖前面设置的值
但每次重试的间隔时间是不同的，会逐渐变长。每次重试的间隔时间如下表。
对于需要重试消费的消息，并不是Consumer在等待了指定时长后再次去拉取原来的消息进行消费
而是将这些需要重试消费的消息放入到了一个特殊Topic的队列中，而后进行再次消费的
这个特殊的队列就是重试队列
当出现需要进行重试消费的消息时Broker会为每个消费组都设置一个Topic名称为%RETRY%consumerGroup@consumerGroup的重试队列
集群消费方式下，消息消费失败后若希望消费重试，则需要在消息监听器接口的实现中明确进行如下三种方式之一的配置：
return ConsumeConcurrentlyStatus.RECONSUME_LATER;集群消费方式下，消息消费失败后若不希望消费重试，则在捕获到异常后同样也返回与消费成功后的相同的结果，即ConsumeConcurrentlyStatus.CONSUME_SUCCESS，则不进行消费重试。
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
当一条消息初次消费失败，消息队列会自动进行消息重试
达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时消息队列不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中
这个队列就是死信队列(Dead-Letter Queue, DLQ)
其中的消息则称为死信消息(Dead-Letter Message, DLM)
实际上，当一条消息进入死信队列，就意味着系统中某些地方出现了问题，从而导致消费者无法正常消费该消息，比如代码中原本就存在Bug
因此，对于死信消息，通常需要开发人员进行特殊处理。最关键的步骤是要排查可疑因素，解决代码中可能存在的Bug.然后再将原来的死信消息再次进行投递消费
此处可能存在不合适展示的内容，页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容，可点击提交进行申诉，我们将尽快为您处理。