RocketMQ核心原理 java

RocketMQ核心原理简介 ? RocketMQ在阿里内部叫做Metaq（最早名为Metamorphosis，中文意思“变形记”，是作家卡夫卡的中篇小说代表作，可见是为了致敬Kafka）。RocketMQ是Metaq 3.0之后开源的版本。Metaq在阿里巴巴集团内部、蚂蚁金服、菜鸟等各业务中被广泛使用，接入了上万个应用系统中。并平稳支撑了历年的双十一大促（万亿级的消息），在性能、稳定性、可靠性等方面表现出色，在整个阿里技术体系和大中台战略中发挥着举足轻重的作用。Metaq最早源于Kafka，早期借鉴了Kafka很多优秀的设计。但是由于Kafka是Scale语言编写而阿里系主要使用Java，且无法满足阿里的电商、金融业务场景，所以誓嘉（花名）团队用Java重新造轮子，并做了大量的改造和优化。在此之前，淘宝有一款消息中间件名为Notify，目前已经逐步被Metaq所取代。第一代的Notify主要使用了推模型，解决了事务消息；第二代的MetaQ主要使用了拉模型，解决了顺序消息和海量堆积的问题。相比起Kafka使用的Scale语言编写，RabbitMQ 使用Erlang语言编写，基于Java的RocketMQ开源后更容易被广泛的研究，以及其他大厂定制开发。

文章图片

使用场景

应用解耦
流量削峰
数据分发

特性

发布与订阅
发布是指某个生产者向某个topic发送消息。消息的订阅是指某个消费者关注了某个topic中带有某些tag的消息。
消息顺序
一类消息消费时能按照发送的顺序来消费
消息过滤
可以根据Tag进行消息过滤，也支持自定义属性过滤。消息过滤目前是在Broker端实现的。
消息可靠性
至少一次
每个消息必须投递一次。Consumer先Pull消息到本地，消费完成后，才向服务器返回ack，如果没有消费一定不会ack消息。
回溯消费
Consumer已经消费成功的消息，由于业务上需求需要重新消费。
事务消息
应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。
延迟消息
是指消息发送到broker后，不会立即被消费，等待特定时间投递给真正的topic。
消息重试
Consumer消费消息失败后，要触发重试机制令消息再消费一次。
消息重投
生产者在发送消息时，同步消息失败会重投；异步消息有重试；oneway没有任何保证。

架构角色介绍

角色	说明
Producer	消息的发送者
Consumer	消息发送者
Broker	暂存和传输消息
NameServer	管理Broker
Topic	区分消息的种类，一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者可以订阅一个或者多个Topic消息
Message Queue	相当于是Topic的分区，用于并行发送和接收消息

架构图

文章图片

? NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master。Master与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。当前RocketMQ版本在部署架构上支持一Master多Slave，但只有BrokerId=1的从服务器才会参与消息的读负载。
? Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic 服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。
? Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，消费者在向Master拉取消息时，Master服务器会根据拉取偏移量与最大偏移量的距离（判断是否读老消息，产生读I/O），以及从服务器是否可读等因素建议下一次是从Master还是Slave拉取。
核心特性以及原理消息发送
? 不同的业务场景需要生产者采用不同的写入策略。比如同步发送、异步发送、Oneway发送、延迟发送、发送事务消息等。
消息发送后返回状态有如下四种：

FLUSH_DISK_TIMEOUT
FLUSH_SLAVE_TIMEOUT
SLAVE_NOT_AVAILABLE
SEND_OK

不同状态在不同的刷盘策略和同步策略的配置下含义是不同的
FLUSH_DISK_TIMEOUT ：表示没有在规定时间内完成刷盘（需要Broker的刷盘策略被设置成SYNC_FLUSH才会报这个错误）
FLUSH_SLAVE_TIMEOUT：表示在主备方式下，并且Broker被设置成SYNC_MASTER方式，没有在设定时间内完成主从同步。
SLAVE_NOT_AVAILABLE：这个状态产生的场景和FLUSH_SLAVE_TIMEOUT类似，表示在主备方式下，并且Broker被设置成SYNC_MASTER，但是没有找到被配置成Slave的Broker。
SEND_OK：表示发送成功，发送成功的具体含义，比如消息是否已经被存储到磁盘？消息是否被同步到了Slave上？消息在Slave上是否被写入磁盘？需要结合所配置的刷盘策略、主从策略来定。
? 这个状态还可以简单理解为，没有发生上面列出的三个问题状态就是SEND_OK。
Oneway发送 ? 在一些对速度要求高，但是可靠性要求不高的场景下，比如日志收集类应用，可以采用Oneway方式发送。Oneway方式只发送请求不等待应答，即将数据写入客户端的Socket缓冲区就返回，不等待对方返回结果。这种方式发送消息可以缩短到微妙级别。或者可以采用多个Producer并发发送的方式，这种方式下无需太多关注写入性能问题。RocketMQ引入了一个并发窗口，在窗口内消息可以并发地写入DirectMem中，然后异步地将连续一段无空洞的数据刷入文件系统当中。同时顺序写CommitLog可让RocketMQ无论在HDD还是SSD磁盘情况下都能保持较高的写入性能。在Linux操作系统层级进行调优，推荐使用EXT4文件系统，IO调度算法使用deadline算法。
消息消费
主要可以总结为以下几点

消息消费方式（Pull和Push）
消息消费的模式（广播模式和集群模式）
流量控制（可结合sentinel来实现，后面补充）
并发线程数设置
消息的过滤（Tag、Key） TagA||TagB||TagC * null

消息存储
存储方式选型 ? Apache下开源的另外一款MQ—ActiveMQ（默认采用的KahaDB做消息存储）可选用JDBC的方式来做消息持久化，通过简单的xml配置信息即可实现JDBC消息存储。由于，普通关系型数据库（如
Mysql）在单表数据量达到千万级别的情况下，其IO读写性能往往会出现瓶颈。在可靠性方面，该种方案非常依赖DB，如果一旦DB出现故障，则MQ的消息就无法落盘存储会导致线上故障。
? 目前业界较为常用的几款产品（RocketMQ/Kafka/RabbitMQ）均采用的是消息刷盘至所部署虚拟机/物理机的文件系统来做持久化（刷盘一般可以分为异步刷盘和同步刷盘两种模式）。消息刷盘为消
息存储提供了一种高效率、高可靠性和高性能的数据持久化方式。除非部署MQ机器本身或是本地磁盘挂了，否则一般是不会出现无法持久化的故障问题。
一般情况下，文件系统> 关系型数据库
存储结构 ? RocketMQ消息的存储是由ConsumeQueue和CommitLog配合完成的，消息真正的物理存储文件是CommitLog，ConsumeQueue是消息的逻辑队列，类似数据库的索引文件，存储的是
指向物理存储的地址。每个Topic下的每个Message Queue都有一个对应的ConsumeQueue文件。如下图所示

文章图片

CommitLog ? commitLog文件的存储地址：$HOME\store\commitlog\${fileName}，每个文件的大小默认1G =102410241024，commitLog的文件名fileName，名字长度为20位，左边补零，剩余为起始偏移量；比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当这个文件满了，第二个文件名字为00000000001073741824，起始偏移量为1073741824，以此类推，第三个文件名字为00000000002147483648，起始偏移量为2147483648消息存储的时候会顺序写入文件，当文件满了，写入下一个文件。MappedFileQueue可以看作是${ROCKET_HOME}/store/commitlog文件夹，而MappedFile则对应该文件夹下一个个的文件。
文件的存储结构

顺序编号	字段简称（单位字节）	大小	含义
1	msgSize	4	代表这个消息的大小
2	MAGICCODE	4	MAGICCODE = daa320a7
3	BODY CRC	4	消息体BODY CRC 当broker重启recover时会校验
4	queueId	4
5	flag	4
6	QUEUEOFFSET	8	这个值是个自增值不是真正的consume queue的偏移量，可以代表这个consumeQueue队列或者tranStateTable队列中消息的个数，若是非事务消息或者commit事务消息，可以通过这个值查找到consumeQueue中数据，QUEUEOFFSET * 20才是偏移地址；若是PREPARED或者Rollback事务，则可以通过该值从tranStateTable中查找数据
7	PHYSICALOFFSET	8	代表消息在commitLog中的物理起始地址偏移量
8	SYSFLAG	4	指明消息是事物事物状态等消息特征，二进制为四个字节从右往左数：当4个字节均为0（值为0）时表示非事务消息；当第1个字节为1（值为1）时表示表示消息是压缩的（Compressed）；当第2个字节为1（值为2）表示多消息（MultiTags）；当第3个字节为1（值为4）时表示prepared消息；当第4个字节为1（值为8）时表示commit消息；当第3/4个字节均为1时（值为12）时表示rollback消息；当第3/4个字节均为0时表示非事务消息
9	BORNTIMESTAMP	8	消息产生端(producer)的时间戳
10	BORNHOST	8	消息产生端(producer)地址(address:port)
11	STORETIMESTAMP	8	消息在broker存储时间
12	STOREHOSTADDRESS	8	消息存储到broker的地址(address:port)
13	RECONSUMETIMES	8	消息被某个订阅组重新消费了几次（订阅组之间独立计数），因为重试消息发送到了topic名字为%retry%groupName的队列queueId=0的队列中去了，成功消费一次记录为0；
14	PreparedTransaction Offset	8	表示是prepared状态的事物消息
15	messagebodyLength	4	消息体大小值
16	messagebody	bodyLength	消息体内容
17	topicLength	1	topic名称内容大小
18	topic	topicLength	topic的内容值
19	propertiesLength	2	属性值大小
20	properties	propertiesLength	propertiesLength大小的属性数据

ConsumeQueue ? 每个ConsumeQueue都有一个id，id 的值为0到TopicConfig配置的队列数量。比如某个Topic的消费队列数量为4，那么四个ConsumeQueue的id就分别为0、1、2、3。
ConsumeQueue是不负责存储消息的，只是负责记录它所属Topic的消息在CommitLog中的偏移量，这样当消费者从Broker拉取消息的时候，就可以快速根据偏移量定位到消息。
ConsumeQueue本身同样是利用MappedFileQueue进行记录偏移量信息的，可见MappedFileQueue的设计多么美妙，它没有与消息进行耦合，而是设计成一个通用的存储功能。
ConsumeQueue更新消息偏移量的整体过程大概如下图所示，其中涉及了几个概念。

ReputMessageService
CommitLogDispatcherBuildConsumeQueue

文章图片

大体上的存储分层设计可以如下：

文章图片

总结以上存储结构：
优点：
a、ConsumeQueue消息逻辑队列较为轻量级；
b、对磁盘的访问串行化，避免磁盘竟争，不会因为队列增加导致IOWAIT增高；
缺点：
a、对于CommitLog来说写入消息虽然是顺序写，但是读却变成了完全的随机读；
b、Consumer端订阅消费一条消息，需要先读ConsumeQueue，再读Commit Log，一定程度上增加了开销；
RocketMQ存储关键技术 Mmap （1）Mmap内存映射技术的特点

Mmap内存映射和普通标准IO操作的本质区别在于它并不需要将文件中的数据先拷贝至OS的内核IO缓冲区，而是可以直接将用户进程私有地址空间中的一块区域与文件对象建立映射关系，这样程序就好像可以直接从内存中完成对文件读/写操作一样。只有当缺页中断发生时，直接将文件从磁盘拷贝至用户态的进程空间内，只进行了一次数据拷贝。对于容量较大的文件来说（文件大小一般需要限制在1.5~2G以下），采用Mmap的方式其读/写的效率和性能都非常高。

文章图片

（2）JDK NIO的MappedByteBuffer

从JDK的源码来看，MappedByteBuffer继承自ByteBuffer，其内部维护了一个逻辑地址变量—address。在建立映射关系时，MappedByteBuffer利用了JDK NIO的FileChannel类提供的map()方法把文件对象映射到虚拟内存。仔细看源码中map()方法的实现，可以发现最终其通过调用native方法map0()完成文件对象的映射工作，同时使用Util.newMappedByteBuffer()方法初始化MappedByteBuffer实例，但最终返回的是DirectByteBuffer的实例。在Java程序中使用MappedByteBuffer的get()方法来获取内存数据是最终通过DirectByteBuffer.get()方法实现（底层通过unsafe.getByte()方法，以“地址 + 偏移量”的方式获取指定映射至内存中的数据）。

（3）使用Mmap的限制

Mmap映射的内存空间释放的问题
由于映射的内存空间本身就不属于JVM的堆内存区（Java Heap），因此其不受JVM GC的控制，卸载这部分内存空间需要通过系统调用 unmap()方法来实现。然而unmap()方法是FileChannelImpl类里实现的私有方法，无法直接显示调用。RocketMQ中的做法是，通过Java反射的方式调用“sun.misc”包下的Cleaner类的clean()方法来释放映射占用的内存空间；
MappedByteBuffer内存映射大小限制
因为其占用的是虚拟内存（非JVM的堆内存），大小不受JVM的-Xmx参数限制，但其大小也受到OS虚拟内存大小的限制。一般来说，一次只能映射1.5~2G 的文件至用户态的虚拟内存空间，所以RocketMQ默认设置单个CommitLog日志数据文件为1G
使用MappedByteBuffe的其他问题
会存在内存占用率较高和文件关闭不确定性的问题

PageCache ? 文件一般存放在硬盘（机械硬盘或固态硬盘）中，CPU 并不能直接访问硬盘中的数据，而是需要先将硬盘中的数据读入到内存中，然后才能被 CPU 访问。为了提升对文件的读写效率，Linux 内核会以页大小（4KB）为单位，将文件划分为多数据块。当用户对文件中的某个数据块进行读写操作时，内核首先会申请一个内存页（页缓存）与文件中的数据块进行绑定。所以PageCache是OS对文件的缓存，用于加速对文件的读写。一般来说，程序对文件进行顺序读写的速度几乎接近于内存的读写访问，这里的主要原因就是在于OS使用PageCache机制对读写访问操作进行了性能优化，将一部分的内存用作PageCache。下图借用了网上一张图片说明下流程。

文章图片

（1）对于数据文件的读取，如果一次读取文件时出现未命中PageCache的情况，OS从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取

顺序读入紧随其后的少数几个页面。这样只要下次访问的文件已经被加载至PageCache时，读取操作的速度基本等于访问内存。

（2）对于数据文件的写入，OS会先写入至Cache内，随后通过异步的方式由pdflush内核线程将Cache内的数据刷盘至物理磁盘上。
对于文件的顺序读写操作来说，读和写的区域都在OS的PageCache内，此时读写性能接近于内存。RocketMQ的大致做法是，将数据文件映射到OS的虚拟内存中（通过JDK NIO的MappedByteBuffer），写消息的时候首先写入PageCache，并通过异步刷盘的方式将消息批量的做持久化（同时也支持同步刷盘）；订阅消费消息时（对CommitLog操作是随机读取），由于PageCache的局部性热点原理且整体情况下还是从旧到新的有序读，因此大部分情况下消息还是可以直接从Page Cache中读取，不会产生太多的缺页（Page Fault）中断而从磁盘读取。

文章图片

PageCache机制也不是完全无缺点的，当遇到OS进行脏页回写，内存回收，内存swap等情况时，就会引起较大的消息读写延迟。对于这些情况，RocketMQ采用了多种优化技术，比如内存预分配，文件预热，mlock系统调用等，来保证在最大可能地发挥PageCache机制优点的同时，尽可能地减少其缺点带来的消息读写延迟。
存储优化（1）预先分配MappedFile
? 在消息写入过程中，调用CommitLog的putMessage()方法，CommitLog会先从MappedFileQueue队列中获取一个 MappedFile，如果没有就新建一个。这里，MappedFile的创建过程是将构建好的一个AllocateRequest请求，具体做法是，将下一个文件的路径、下下个文件的路径、文件大小为参数封装为AllocateRequest对象添加至队列中。后台运行的AllocateMappedFileService服务线程（在Broker启动时，该线程就会创建并运行），会不停地运行，只要请求队列里存在请求，就会去执行MappedFile映射文件的创建和预分配工作。
? 分配的时候有两种策略，一种是使用Mmap的方式来构建MappedFile实例，另外一种是从TransientStorePool堆外内存池中获取相应的DirectByteBuffer来构建MappedFile，具体采用哪种策略，也与刷盘的方式有关。并且，在创建分配完下个MappedFile后，还会将下下个MappedFile预先创建并保存至请求队列中等待下次获取时直接返回。RocketMQ中预分配MappedFile的设计非常巧妙，下次获取时候直接返回就可以不用等待MappedFile创建分配所产生的时间延迟。

文章图片

（2）mlock系统调用
? 其可以将进程使用的部分或者全部的地址空间锁定在物理内存中，防止其被交换到swap空间。对于RocketMQ这种的高吞吐量的分布式消息队列来说，追求的是消息读写低延迟，那么肯定希望尽可能地多使用物理内存，提高数据读写访问的操作效率。
（3）文件预热
? 预热的目的主要有两点；第一点，由于仅分配内存并进行mlock系统调用后并不会为程序完全锁定这些内存，因为其中的分页可能是写时复制的。因此，就有必要对每个内存页面中写入一个假的值。其中，RocketMQ是在创建并分配MappedFile的过程中，预先写入一些随机值至Mmap映射出的内存空间里。第二，调用Mmap进行内存映射后，OS只是建立虚拟内存地址至物理地址的映射表，而实际并没有加载任何文件至内存中。程序要访问数据时OS会检查该部分的分页是否已经在内存中，如果不在，则发出一次缺页中断。RocketMQ在做Mmap内存映射的同时进行madvise系统调用，目的是使OS做一次内存映射后对应的文件数据尽可能多的预加载至内存中，从而达到内存预热的效果。
零拷贝在这里，我想提一下这个零拷贝的概念。在 Java 程序员的世界，常用的零拷贝有 mmap 和 sendFile
MMAP

文章图片

sendFile

文章图片

虽然叫零拷贝，实际上sendfile有2次数据拷贝的。
第1次是从磁盘拷贝到内核缓冲区，第二次是从内核缓冲区拷贝到网卡（协议引擎）。
如果网卡支持 SG-DMA（The Scatter-GatherDirect Memory Access）技术，就无需从PageCache拷贝至 Socket 缓冲区
之所以叫零拷贝，是从内存角度来看的，数据在内存中没有发生过拷贝。
只是在内存和I/O设备之间传输。很多时候我们认为sendfile才是零拷贝，mmap严格来说不算。
Linux中的API为sendfile、mmap，Java中的API为FileChanel.transferTo()、FileChannel.map()等。
Netty、Kafka(sendfile)、Rocketmq（mmap）、Nginx等高性能中间件中，都有大量利用操作系统零拷贝特性。

消息过滤
? RocketMQ分布式消息队列的消息过滤方式有别于其它MQ中间件，是在Consumer端订阅消息时再做消息过滤的。RocketMQ这么做是在于其Producer端写入消息和Consumer端订阅消息采用分离存储的机制来实现的，Consumer端订阅消息是需要通过ConsumeQueue这个消息消费的逻辑队列拿到一个索引，然后再从CommitLog里面读取真正的消息实体内容，所以说到底也是还绕不开其存储结构。其ConsumeQueue的存储结构如下，可以看到其中有8个字节存储的Message Tag的哈希值，基于Tag的消息过滤正式基于这个字段值的。

文章图片

同步/异步复制
如果一个Broker组有Master和Slave，消息需要从Master复制到Slave 上，有同步和异步两种复制方式。
同步复制 ? 同步复制方式是等Master和Slave均写成功后才反馈给客户端写成功状态。在同步复制方式下，如果Master出故障，Slave上有全部的备份数据，容易恢复。
但是同步复制会增大数据写入延迟，降低系统吞吐量。
异步复制 ? 异步复制方式是只要Master写成功即可反馈给客户端写成功状态。在异步复制方式下，系统拥有较低的延迟和较高的吞吐量，但是如果Master出了故障，有些数据因为没有被写入Slave，有可能会丢失。
高可用机制
消息发送高可用 ? 在创建Topic的时候，把Topic的多个Message Queue创建在多个Broker组上（相同Broker名称，不同brokerId的机器组成一个Broker组），这样既可以在性能方面具有扩展性，也可以降低主节点故障
对整体上带来的影响，而且当一个Broker组的Master不可用后，其他组的Master仍然可用，Producer仍然可以发送消息的。
RocketMQ目前还不支持把Slave自动转成Master，如果机器资源不足，需要把Slave转成Master。

手动停止Slave角色的Broker。
更改配置文件。
用新的配置文件启动Broker。

消息消费高可用 ? 在Consumer的配置文件中，并不需要设置是从Master读还是从Slave 读，当Master不可用或者繁忙的时候，Consumer会被自动切换到从Slave 读。
有了自动切换Consumer这种机制，当一个Master角色的机器出现故障后，Consumer仍然可以从Slave读取消息，不影响Consumer程序。这就达到了消费端的高可用性。
刷盘机制
? RocketMQ 的所有消息都是持久化的，先写入系统 PageCache，然后刷盘，可以保证内存与磁盘都有一份数据，访问时，直接从内存读取。消息在通过Producer写入RocketMQ的时候，有两种写磁
盘方式，分布式同步刷盘和异步刷盘。
同步刷盘

文章图片

同步刷盘与异步刷盘的唯一区别是异步刷盘写完 PageCache直接返回，而同步刷盘需要等待刷盘完成才返回，同步刷盘流程如下：
(1) 写入 PageCache后，线程等待，通知刷盘线程刷盘。
(2) 刷盘线程刷盘后，唤醒前端等待线程，可能是一批线程。
(3) 前端等待线程向用户返回成功
异步刷盘

文章图片

在有 RAID 卡，SAS 15000 转磁盘测试顺序写文件，速度可以达到 300M 每秒左右，而线上的网卡一般都为千兆网卡，写磁盘速度明显快于数据网络入口速度，那么是否可以做到写完内存就向用户返
回，由后台线程刷盘呢？
（1）由于磁盘速度大于网卡速度，那么刷盘的进度肯定可以跟上消息的写入速度。
（2）万一由于此时系统压力过大，可能堆积消息，除了写入 IO，还有读取 IO，万一出现磁盘读取落后情况，会不会导致系统内存溢出，答案是否定的，原因如下：
写入消息到 PageCache时，如果内存不足，则尝试丢弃干净的 PAGE，腾出内存供新消息使用，策略是LRU 方式。如果干净页不足，此时写入 PageCache会被阻塞，系统尝试刷盘部分数据，大约每次尝试 32个 PAGE , 来找出更多干净 PAGE。综上，内存溢出的情况不会出现。
负载均衡
? RocketMQ中的负载均衡都在Client端完成，主要可以分为Producer端发消息时的负载均衡和Consumer端订阅消息时负载均衡。
Producer的负载均衡

文章图片

如图所示，5 个队列可以部署在一台机器上，也可以分别部署在 5 台不同的机器上，发送消息通过轮询队列的方式发送，每个队列接收平均的消息量。
通过增加机器，可以水平扩展队列容量。另外也可以自定义方式选择发往哪个队列。
Consumer的负载均衡

文章图片

如图所示，如果有 5 个队列，2 个 consumer，那么第一个 Consumer 消费 3 个队列，第二consumer 消费 2 个队列。这样即可达到平均消费的目的，可以水平扩展 Consumer 来提高消费能
力。但是 Consumer 数量要小于等于队列数量，如果 Consumer 超过队列数量，那么多余的Consumer 将不能消费消息。
消息重试
顺序消息的重试 ? 对于顺序消息，当消费者消费消息失败后，消息队列 RocketMQ 会自动不断进行消息重试（每次间隔时间为 1 秒），这时，应用会出现消息消费被阻塞的情况。因此，在使用顺序消息时，务必保证应
用能够及时监控并处理消费失败的情况，避免阻塞现象的发生。
无序消息的重试 ? 对于无序消息（普通、定时、延时、事务消息），当消费者消费消息失败时，您可以通过设置返回状态达到消息重试的结果。
无序消息的重试只针对集群消费方式生效；广播方式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息。
重试次数消息队列 RocketMQ 默认允许每条消息最多重试 16 次，每次重试的间隔时间如下：

文章图片

死信队列
? RocketMQ 中，这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。可以在控制台Topic列表中看到“DLQ”相关的
Topic，默认命名是：
%RETRY%消费组名称（重试Topic）
%DLQ%消费组名称（死信Topic）
死信队列也可以被订阅和消费，并且也会过期
特性死信消息具有以下特性

不会再被消费者正常消费。
有效期与正常消息相同，均为 3 天，3 天后会被自动删除。因此，请在死信消息产生后的 3天内及时处理。

死信队列具有以下特性：

一个死信队列对应一个 Group ID，而不是对应单个消费者实例。
如果一个 Group ID 未产生死信消息，消息队列 RocketMQ 不会为其创建相应的死信队列。
一个死信队列包含了对应 Group ID 产生的所有死信消息，不论该消息属于哪个 Topic。

延迟消息
? 定时消息（延迟队列）是指消息发送到broker后，不会立即被消费，等待特定时间投递给真正的topic。 broker有配置项messageDelayLevel，默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m
8m 9m 10m 20m 30m 1h 2h”，18个level。可以配置自定义messageDelayLevel。注意，messageDelayLevel是broker的属性，不属于某个topic。发消息时，设置delayLevel等级即可：
msg.setDelayLevel(level)。level有以下三种情况：

level == 0，消息为非延迟消息
1<=level<=maxLevel，消息延迟特定时间，例如level==1，延迟1s
level > maxLevel，则level== maxLevel，例如level==20，延迟2h

定时消息会暂存在名为SCHEDULE_TOPIC_XXXX的topic中，并根据delayTimeLevel存入特定的queue，queueId = delayTimeLevel – 1，即一个queue只存相同延迟的消息，保证具有相同发送延迟
的消息能够顺序消费。broker会调度地消费SCHEDULE_TOPIC_XXXX，将消息写入真实的topic。需要注意的是，定时消息会在第一次写入和调度写入真实topic时都会计数，因此发送数量、tps都
会变高。
顺序消息
? 顺序消息是指消息的消费顺序和产生顺序相同，在有些业务逻辑下，必须保证顺序。比如订单的生成、付款、发货，这3个消息必须按顺序处理才行。
顺序消息分为全局顺序消息和部分顺序消息：

全局顺序消息指某个Topic下的所有消息都要保证顺序；
部分顺序消息只要保证每一组消息被顺序消费即可，比如订单消息，只要保证同一个订单ID的消息能按顺序消费即可。

? 在多数的业务场景中实际上只需要局部有序就可以了。RocketMQ在默认情况下不保证顺序，比如创建一个Topic，默认八个写队列，八个读队列。这时
【RocketMQ核心原理】候一条消息可能被写入任意一个队列里；在数据的读取过程中，可能有多个Consumer，每个Consumer也可能启动多个线程并行处理，所以消息被哪个Consumer消费，被消费的顺序和写入的顺
序是否一致是不确定的。要保证全局顺序消息，需要先把Topic的读写队列数设置为一，然后Producer和Consumer的并发设置也要是一。简单来说，为了保证整个Topic的全局消息有序，只能消除所有的并发
处理，各部分都设置成单线程处理。

文章图片

部分消息有序要保证部分消息有序，需要发送端和消费端配合处理。在发送端，要做到把同一业务ID的消息发送到同一个Message Queue；在消费过程中，要做到从同一个Message Queue读取的消息不被并发处
理，这样才能达到部分有序。消费端通过使用MessageListenerOrderly类来解决单Message Queue的消息被并发处理的问题。
Consumer使用MessageListenerOrderly的时候，下面四个Consumer的设置依旧可以使用：

setConsumeThreadMin
setConsumeThreadMax
setPullBatchSize
setConsumeMessageBatchMaxSize

前两个参数设置Consumer的线程数；PullBatchSize指的是一次从Broker的一个Message Queue获取消息的最大数量，默认值是32；ConsumeMessageBatchMaxSize指的是这个Consumer的Executor（也就
是调用MessageListener处理的地方）一次传入的消息数（Listmsgs这个链表的最大长度），默认值是1。上述四个参数可以使用，说明MessageListenerOrderly并不是简单地禁止并发处理。
在MessageListenerOrderly的实现中，为每个Consumer Queue加个锁，消费每个消息前，需要先获得这个消息对应的Consumer Queue所对应的锁，这样保证了同一时间，同一个Consumer Queue的消息不
被并发消费，但不同Consumer Queue的消息可以并发处理。
事务消息
RocketMQ提供了事务消息的功能，采用2PC(两段式协议)+补偿机制（事务回查）的分布式事务功能，通过消息队列 RocketMQ 版事务消息能达到分布式事务的最终一致。

半事务消息：暂不能投递的消息，发送方已经成功地将消息发送到了消息队列 RocketMQ 版服务端，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，处于该种状态下的消息即半事务消息。
消息回查：由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，消息队列 RocketMQ 版服务端通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit 或是 Rollback），该询问过程即消息回查。
文章图片

事务消息发送步骤如下：

发送方将半事务消息发送至消息队列 RocketMQ 版服务端。
消息队列 RocketMQ 版服务端将消息持久化成功之后，向发送方返回 Ack 确认消息已经发送成功，此时消息为半事务消息。
发送方开始执行本地事务逻辑。
发送方根据本地事务执行结果向服务端提交二次确认（Commit 或是 Rollback），服务端收到 Commit 状态则将半事务消息标记为可投递，订阅方最终将收到该消息；服务端收到 Rollback 状态则删除半事务消息，订阅方将不会接受该消息。

事务消息回查步骤如下：