首页  思维导图  详情

分布式中间件RocketMQ

2022-04-18 00:23:48   25  举报





AI智能生成

为你推荐

查看更多



rocketmq技术学习

作者其他创作

大纲/内容

release v4.9.0

版本选型

MQ，Message Queue，是一种提供消息队列服务的中间件，也称为消息中间件，是一套提供了消息生产、存储、消费全过程API的软件系统。消息即数据。一般消息的体量不会很大。

简介

限流削峰

异步解耦

数据收集

参考文档

参考这位博客的文章：https://www.cnblogs.com/armyfai/p/13595055.html

用途

ActiveMQ是使用Java语言开发一款MQ产品。早期很多公司与项目中都在使用。但现在的社区活跃度已经很低。现在的项目中已经很少使用了。

ActiveMQ

RabbitMQ是使用ErLang语言开发的一款MQ产品。其吞吐量较Kafka与RocketMQ要低，且由于其不是Java语言开发，所以公司内部对其实现定制化开发难度较大。

RabbitMQ

Kafka是使用Scala/Java语言开发的一款MQ产品。其最大的特点就是高吞吐率，常用于大数据领域的实时计算、日志采集等场景。其没有遵循任何常见的MQ协议，而是使用自研协议。对于Spring Cloud Netçix，其仅支持RabbitMQ与Kafka。

Kafka

RocketMQ是使用Java语言开发的一款MQ产品。经过数年阿里双11的考验，性能与稳定性非常高。其没有遵循任何常见的MQ协议，而是使用自研协议。对于Spring Cloud Alibaba，其支持RabbitMQ、 Kafka，但提倡使用RocketMQ。

RocketMQ

常见的MQ产品

一般情况下MQ的实现是要遵循一些常规性协议的。常见的协议如下：

JMS，Java Messaging Service（Java消息服务）。是Java平台上有关MOM（Message Oriented Middleware，面向消息的中间件 PO/OO/AO）的技术规范，它便于消息系统中的Java应用程序进行消息交换，并且通过提供标准的产生、发送、接收消息的接口，简化企业应用的开发。ActiveMQ是该协议的典型实现。

JMS

STOMP，Streaming Text Orientated Message Protocol（面向流文本的消息协议），是一种MOM设计的简单文本协议。STOMP提供一个可互操作的连接格式，允许客户端与任意STOMP消息代理（Broker）进行交互。ActiveMQ是该协议的典型实现，RabbitMQ通过插件可以支持该协议。

STOMP

AMQP，Advanced Message Queuing Protocol（高级消息队列协议），一个提供统一消息服务的应用层标准，是应用层协议的一个开放标准，是一种MOM设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同开发语言等条件的限制。 RabbitMQ是该协议的典型实现。

AMQP

MQTT，Message Queuing Telemetry Transport（消息队列遥测传输），是IBM开发的一个即时通讯协议，是一种二进制协议，主要用于服务器和低功耗IoT（物联网）设备间的通信。该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当做传感器和致动器的通信协议。 RabbitMQ通过插件可以支持该协议。

MQTT

说明

MQ常见协议

MQ的概述

官网地址

RocketMQ的概述

消息是指，消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。

消息（Message）

topic:message 1:n

message:topic 1:1

Topic表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行消息订阅的基本单位

producer:topic 1:n

consumer:topic 1:1

一个生产者可以同时发送多种Topic的消息；而一个消费者只对某种特定的Topic感兴趣，即只可以订阅和消费一种Topic的消息

通过代码指定

通过rockemq控制台指定

通过mqadmin 命令行指定

创建topic的方式有三种（这里先知道就行)

topic理解图

图展示

主题（Topic）

为消息设置的标签，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。

Topic是消息的一级分类，Tag是消息的二级分类

tag: 上海

tag: 北京

tag: 南京

Topic: 货物

生产者

topic=货物 tag = 上海

topic=货物 tag = 上海|浙江

topic=货物 tag = *

消费者

标签（Tag）

存储消息的物理实体。一个Topic中可以包含多个Queue，每个Queue中存放的就是该Topic的消息。一个Topic的Queue也被称为一个Topic中消息的分区（Partition）

一个Topic的Queue中的消息只能被一个消费者组中的一个消费者消费

一个Queue中的消息不允许同一个消费者组中的多个消费者同时消费

producer.setDefaultTopicQueueNums(2);

通过代码的方式指定

通过rocketmq控制台指定

在创建topic时指定queue的数量的方式有三种

队列（Queue）

RocketMQ中每个消息拥有唯一的MessageId，且可以携带具有业务标识的Key，以方便对消息的查询。不过需要注意的是，MessageId有两个：在生产者send()消息时会自动生成一个MessageId（msgId)，当消息到达Broker后，Broker也会自动生成一个MessageId(offsetMsgId)。msgId、offsetMsgId与key都称为消息标识

producerIp + 进程pid + MessageClientIDSetter类的ClassLoader的hashCode +当前时间 + AutomicInteger自增计数器

由producer端生成，其生成规则为

msgId

brokerIp + 物理分区的offset（Queue中的偏移量）

由broker端生成，其生成规则为

offsetMsgId

由用户指定的业务相关的唯一标识

key

消息标识（MessageId/Key）

基本概念

如果架构图无法打开，请评论联系我，或者去看官网，我也仅仅是重画了官网提供的架构图

系统架构图

例如，业务系统产生的日志写入到MQ的过程，就是消息生产的过程

再如，电商平台中用户提交的秒杀请求写入到MQ的过程，就是消息生产的过程

消息生产者，负责生产消息。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟。

RocketMQ中的消息生产者都是以生产者组（Producer Group）的形式出现的。生产者组是同一类生产者的集合，这类Producer发送相同Topic类型的消息。一个生产者组可以同时发送多个主题的消息。这句话矛盾呀？？？、

生产者（Producer）

例如，QoS系统从MQ中读取日志，并对日志进行解析处理的过程就是消息消费的过程

再如，电商平台的业务系统从MQ中读取到秒杀请求，并对请求进行处理的过程就是消息消费的过程。

消息消费者，负责消费消息。一个消息消费者会从Broker服务器中获取到消息，并对消息进行相关业务处理

RocketMQ中的消息消费者都是以消费者组（Consumer Group）的形式出现的。消费者组是同一类消费者的集合，这类Consumer消费的是同一个Topic类型的消息。消费者组使得在消息消费方面，实现负载均衡（将一个Topic中的不同的Queue平均分配给同一个Consumer Group的不同的Consumer，注意，并不是将消息负载均衡）和容错（一个Consmer挂了，该Consumer Group中的其它Consumer可以接着消费原Consumer消费的Queue）的目标变得非常容易

消费者（Consumer）

说明：RocketMQ的思想来自于Kafka，而Kafka是依赖了Zookeeper的。所以，在RocketMQ的早期版本，也是依赖于Zookeeper的。从MetaQ v3.0，即RocketMQ开始去掉了Zookeeper依赖，使用了自己的NameServer

NameServer是一个Broker与Topic路由的注册中心，支持Broker的动态注册与发现

Broker管理：接受Broker集群的注册信息并且保存下来作为路由信息的基本数据；提供心跳检测机制，检查Broker是否还存活

路由信息管理：每个NameServer中都保存着Broker集群的整个路由信息和用于客户端查询的队列信息。Producer和Conumser通过NameServer可以获取整个Broker集群的路由信息，从而进行消息的投递和消费

主要包括两个功能

NameServer集群搭建简单，方便扩容

优点

对于Broker而言，必须明确指定所有的NameServer地址，未指定的将不会去注册。也正因为如此，NameServer并不能随便扩容，因为此时Broker若不进行重新配置，新增的NameServer对于Broker来说是不可见的，更不会向这个NameServer注册

缺点

NameServer无状态的优缺点

NameServer通常也是以集群的方式部署，不过，NameServer是无状态的，即NameServer集群中的各个节点间是无差异的，各节点间相互不进行信息通讯。那各节点中的数据是如何进行数据同步的呢？在Broker节点启动时，轮询NameServer列表，与每个NameServer节点建立长连接，发起注册请求。在NameServer内部维护着⼀个 Broker列表，用来动态存储Broker的信息。

Broker节点为了证明自己是活着的，为了维护与NameServer间的长连接，会将最新的信息以心跳包的方式上报给NameServer，每30秒发送一次心跳。心跳包中包含 BrokerId、Broker地址(IP+Port)、 Broker名称、Broker所属集群名称等等。NameServer在接收到心跳包后，会更新心跳时间戳，记录这个Broker的最新存活时间。

路由注册

由于Broker关机、宕机或网络抖动等原因，NameServer没有收到Broker的心跳，NameServer可能会将其从Broker列表中剔除。

NameServer中有⼀个定时任务，每隔10秒就会扫描⼀次Broker表，查看每一个Broker的最新心跳时间戳距离当前时间是否超过120秒，如果超过，则会判定Broker失效，然后将其从Broker列表中剔除。

OP需要将Broker的读写权限禁掉。一旦client(Consumer或Producer)向broker发送请求，都会收到broker的NO_PERMISSION响应，然后client会进行对其它Broker的重试。当OP观察到这个Broker没有流量后，再关闭它，实现Broker从NameServer的移除。

可以尝试自己测试下

扩展：对于RocketMQ日常运维工作，例如Broker升级，需要停掉Broker的工作。OP需要怎么做？

路由剔除

RocketMQ的路由发现采用的是Pull模型。当Topic路由信息出现变化时，NameServer不会主动推送给客户端，而是客户端定时拉取主题最新的路由。默认客户端每30秒会拉取一次最新的路由。

实时性要求较高

Client数量不多，Server数据变化较频繁

其实时性较好，是一个“发布-订阅”模型，需要维护一个长连接。而长连接的维护是需要资源成本的。该模型适合于的场景：

Push模型：推送模型

存在的问题是，实时性较差

Pull模型：拉取模型

其是对Push与Pull模型的整合，充分利用了这两种模型的优势，屏蔽了它们的劣势。

Long Polling模型：长轮询模型

扩展

路由发现

这里的客户端指的是Producer与Consumer

客户端在配置时必须要写上NameServer集群的地址，那么客户端到底连接的是哪个NameServer节点呢？客户端首先会生产一个随机数，然后再与NameServer节点数量取模，此时得到的就是所要连接的节点索引，然后就会进行连接。如果连接失败，则会采用round-robin策略，逐个尝试着去连接其它节点

首先采用的是随机策略进行的选择，失败后采用的是轮询策略

客户端NameServer选择策略

NameServer

Broker充当着消息中转角色，负责存储消息、转发消息。Broker在RocketMQ系统中负责接收并存储从生产者发送来的消息，同时为消费者的拉取请求作准备。Broker同时也存储着消息相关的元数据，包括消费者组消费进度偏移offset、主题、队列等

功能介绍

组成架构图

整个Broker的实体，负责处理来自clients端的请求。而这个Broker实体则由以下模块构成

Remoting Module

客户端管理器。负责接收、解析客户端(Producer/Consumer)请求，管理客户端。例如，维护Consumer的Topic订阅信息

Client Manager

存储服务。提供方便简单的API接口，处理消息存储到物理硬盘和消息查询功能。

Store Service

高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能

HA Service

索引服务。根据特定的Message key，对投递到Broker的消息进行索引服务，同时也提供根据Message Key对消息进行快速查询的功能。

Index Service

模块构成

为了增强Broker性能与吞吐量，Broker一般都是以集群形式出现的。各集群节点中可能存放着相同Topic的不同Queue。不过，这里有个问题，如果某Broker节点宕机，如何保证数据不丢失呢？其解决方案是，将每个Broker集群节点进行横向扩展，即将Broker节点再建为一个HA集群，解决单点问题。

Broker节点集群是一个主从集群，即集群中具有Master与Slave两种角色。Master负责处理读写操作请求，Slave负责对Master中的数据进行备份。当Master挂掉了，Slave则会自动切换为Master去工作。所以这个Broker集群是主备集群。一个Master可以包含多个Slave，但一个Slave只能隶属于一个Master。 Master与Slave 的对应关系是通过指定相同的BrokerName、不同的BrokerId 来确定的。BrokerId为0表示Master，非0表示Slave。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。

集群部署

Broker

角色

启动NameServer，NameServer启动后开始监听端口，等待Broker、Producer、Consumer连接。

启动Broker时，Broker会与所有的NameServer建立并保持长连接，然后每30秒向NameServer定时发送心跳包。

发送消息前，可以先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，当然，在创建Topic时也会将Topic与Broker的关系写入到NameServer中。不过，这步是可选的，也可以在发送消息时自动创建Topic。

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在该集群中，所有Broker中的Queue数量是相同的

Broker模式：该模式下创建的Topic在该集群中，每个Broker中的Queue数量可以不同

手动创建Topic时，有两种模式

自动创建Topic时，默认采用的是Broker模式，会为每个Broker默认创建4个Queue

Topic的创建模式

从物理上来讲，读/写队列是同一个队列。所以，不存在读/写队列数据同步问题。读/写队列是逻辑上进行区分的概念。一般情况下，读/写队列数量是相同的

例如，创建Topic时设置的写队列数量为8，读队列数量为4，此时系统会创建8个Queue，分别是0 1 2 3 4 5 6 7。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 5 6 7中是没有消息的。此时假设Consumer Group中包含两个Consuer，Consumer1消费0 1 2 3，而Consumer2消费4 5 6 7。但实际情况是，Consumer2是没有消息可消费的

例如，原来创建的Topic中包含16个Queue，如何能够使其Queue缩容为8个，还不会丢失消息？可以动态修改写队列数量为8，读队列数量不变。此时新的消息只能写入到前8个队列，而消费都消费的却是16个队列中的数据。当发现后8个Queue中的消息消费完毕后，就可以再将读队列数量动态设置为8。整个缩容过程，没有丢失任何消息

其这样设计的目的是为了，方便Topic的Queue的缩容

也就是说，当读/写队列数量设置不同时，总是有问题的。那么，为什么要这样设计呢？

perm用于设置对当前创建Topic的操作权限：2表示只写，4表示只读，6表示读写

队列

读/写队列

工作流程

系统架构

1.概述

选择自己要下载的版本

https://rocketmq.apache.org/release_notes/release-notes-4.9.3/

rocketmq

官网下载

我比较喜欢下载完了在上传上去

wget -c https://dlcdn.apache.org/rocketmq/4.9.3/rocketmq-all-4.9.3-bin-release.zip

也可以使用命令直接下载

上传到服务器

将这一行的初始内存修改为这样

JAVA_OPT=\"${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m\"

修改 runserver.sh

JAVA_OPT=\"${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m\"

修改改runbroker.sh

修改初始内存

https://rocketmq.apache.org/docs/quick-start/

官方文档

参考官方文档

说明：因为我是在阿里云ECS服务器上部署的，所以我启动的方式是下面这样，如果你在本地部署，就可以参考上面的官方文档进行启动

将localhost 替换为公网ip

nohup ./bin/mqnamesrv -n 47.96.111.123:9876 &

先启动 nameserver

nohup sh bin/mqbroker -n 你的公网IP:9876 -c conf/broker.conf autoCreateTopicEnable=true &

在启动broker

注意：使用以上命令的时候注意路径，需要在rocketmq的安装根目录下执行

启动

看上面的官方文档吧，直接执行命令就可以

发送/接收消息测试

sh bin/mqshutdown broker

先关闭broker

sh bin/mqshutdown namesrv

再关闭nameserver

关闭服务

是nameServer的端口

9876

是broker的端口号，作为对Producer和Consumer提供服务的端口号

listenPort=10911

10911

是高可用服务使用的端口号

默认值为listenPort+1

haListenPort=10912

10912

关于端口

单机安装与启动

https://github.com/apache/rocketmq-externals/tree/release-rocketmq-console-1.0.0/rocketmq-console

rocketmq console

1.官网下载

1.解压缩

修改为一个不常用的端口比如 8888，默认是8080

1.修改 server.port

ip 是我的阿里云公网ip

例如：rocketmq.config.namesrvAddr=rocketmq.config.namesrvAddr=47.96.111.162:9876

注意：由于我使用的是阿里云的ECS服务器，所以需要去配置一下9876端口，不然无法访问

2.修改 rocketmq.config.namesrvAddr

2.修改 resources 目录下的 application.properties 文件

可选，如果遇到问题可以添加上

3.修改pom.xml

2.修改配置

得到一个jar包：rocketmq-console-ng-1.0.0.jar

mvn clean package -Dmaven.test.skip=true

下载完修改配置后在终端中执行命令

方式一

将项目导入到IDEA中然后通过maven 插件编译打包

方式二

3.打包

前台启动，关闭linux窗口后服务就停了，不建议使用

java -jar rocketmq-console-ng-1.0.0.jar

后台启动

java -jar rocketmq-console-ng-1.0.0.jar &

他会在执行命令的当前目录下创建一个 nohup.out 的输出日志

后台启动，推荐使用

nohup java -jar rocketmq-console-ng-1.0.0.jar &

4.启动

因为我是在我们linxu上执行的jar包，所以是服务器的ip（或者域名），注意：端口要记得开放，不然无法访问

http://ip:8888

5.访问

控制台的安装与启动

同步复制：消息写入master后，master会等待slave同步数据成功后才向producer返回成功ACK

异步复制：消息写入master后，master立即向producer返回成功ACK，无需等待slave同步数据成功

复制策略是Broker的Master与Slave间的数据同步方式。分为同步复制与异步复制

复制策略

同步刷盘：当消息持久化到broker的磁盘后才算是消息写入成功

异步刷盘策略会降低系统的写入延迟，RT变小，提高了系统的吞吐量

消息写入到Broker的内存，一般是写入到了PageCache

对于异步刷盘策略，消息会写入到PageCache后立即返回成功ACK。但并不会立即做落盘操作，而是当PageCache到达一定量时会自动进行落盘

异步刷盘：当消息写入到broker的内存后即表示消息写入成功，无需等待消息持久化到磁盘

刷盘策略指的是broker中消息的落盘方式，即消息发送到broker内存后消息持久化到磁盘的方式。分为同步刷盘与异步刷盘

刷盘策略

数据复制与刷盘策略

https://github.com/apache/rocketmq/blob/master/docs/cn/operation.md

文档

只有一个broker（其本质上就不能称为集群）。这种方式也只能是在测试时使用，生产环境下不能使用，因为存在单点问题

单Master

以上优点的前提是，这些Master都配置了RAID磁盘阵列。如果没有配置，一旦出现某Master宕机，则会发生大量消息丢失的情况

优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高

缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅（不可消费），消息实时性会受到影响

broker集群仅由多个master构成，不存在Slave。同一Topic的各个Queue会平均分布在各个master节点上

多Master

broker集群由多个master构成，每个master又配置了多个slave（在配置了RAID磁盘阵列的情况下，一个master一般配置一个slave即可）。master与slave的关系是主备关系，即master负责处理消息的读写请求，而slave仅负责消息的备份与master宕机后的角色切换

异步复制即前面所讲的复制策略中的异步复制策略，即消息写入master成功后，master立即向producer返回成功ACK，无需等待slave同步数据成功

该模式的最大特点之一是，当master宕机后slave能够自动切换为master。不过由于slave从master的同步具有短暂的延迟（毫秒级），所以当master宕机后，这种异步复制方式可能会存在少量消息的丢失问题

对于Master的RAID磁盘阵列，若使用的也是异步复制策略，同样也存在延迟问题，同样也可能会丢失消息。但RAID阵列的秘诀是微秒级的（因为是由硬盘支持的），所以其丢失的数据量会更少

多Master多Slave模式-异步复制

该模式是多Master多Slave模式的同步复制实现。所谓同步双写，指的是消息写入master成功后，master会等待slave同步数据成功后才向producer返回成功ACK，即master与slave都要写入成功后才会返回成功ACK，也即双写

优点：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高

缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机

多Master多Slave模式-同步双写

一般会为Master配置RAID10磁盘阵列，然后再为其配置一个Slave。即利用了RAID10磁盘阵列的高效、安全性，又解决了可能会影响订阅的问题

多Master+RAID阵列，其仅仅可以保证数据不丢失，即不影响消息写入，但其可能会影响到消息的订阅。但其执行效率要远高于多Master多Slave集群

多Master多Slave集群(同步双写），其不仅可以保证数据不丢失，也不会影响消息写入。其运行效率要低于多Master+RAID阵列

多Master+RAID阵列，与多Master多Slave集群的区别是什么？

最佳实践

集群模式

Broker集群模式

集群搭建理论

https://www.yuque.com/qiuguan9/yy08il/nlrpgt#DeoC1

集群搭建实践

举例：./mqadmin queryMsgById -n 47.96.111.45:9876 -i '7F0000013BC018B4AAC292A0F8AA0000'

mqadmin命令

2.安装与启动

Producer发送消息之前，会先向NameServer发出获取消息Topic的路由信息的请求

NameServer返回该Topic的路由表及Broker列表

Producer根据代码中指定的Queue选择策略，从Queue列表中选出一个队列，用于后续存储消息

Produer对消息做一些特殊处理，例如，消息本身超过4M，则会对其进行压缩

Producer向选择出的Queue所在的Broker发出RPC请求，将消息发送到选择出的Queue

消息的生产过程

默认选择算法。该算法保证了每个Queue中可以均匀的获取到消息

该算法存在一个问题：由于某些原因，在某些Broker上的Queue可能投递延迟较严重。从而导致 Producer的缓存队列中出现较大的消息积压，影响消息的投递性能

轮询算法

该算法会统计每次消息投递的时间延迟，然后根据统计出的结果将消息投递到时间延迟最小的Queue。如果延迟相同，则采用轮询算法投递。该算法可以有效提升消息的投递性能

该算法也存在一个问题：消息在Queue上的分配不均匀。投递延迟小的Queue其可能会存在大量的消息。而对该Queue的消费者压力会增大，降低消息的消费能力，可能会导致MQ中消息的堆积

最小投递延迟算法

对于无序消息，其Queue选择算法，也称为消息投递算法，常见的有两种

Queue选择算法

消息的生产

RocketMQ中的消息存储在本地文件系统中，这些相关文件默认在当前用户主目录下的store目录中。(比如 /root/store/

abort 文件：该文件在Broker启动后会自动创建，正常关闭Broker，该文件会自动消失。若在没有启动Broker的情况下，发现这个文件是存在的，则说明之前Broker的关闭是非正常关闭

checkpoint 文件：其中存储着commitlog、consumequeue、index文件的最后刷盘时间戳

在很多资料中commitlog目录中的文件简单就称为commitlog文件。但在源码中，该文件被命名为mappedFile

mapedFileSizeCommitLog=1073741824

commitlog目录中存放着很多的mappedFile文件，当前Broker中的所有消息都是落盘到这些mappedFile文件中的

mappedFile文件是顺序读写的文件，所有其访问效率很高

无论是SSD磁盘还是SATA磁盘，通常情况下，顺序存取效率都会高于随机存取

需要注意的是，一个Broker中仅包含一个commitlog目录，所有的mappedFile文件都是存放在该目录中的。即无论当前Broker中存放着多少Topic的消息，这些消息都是被顺序写入到了mappedFile文件中的。也就是说，这些消息在Broker中存放时并没有被按照Topic进行分类存放

不难发现，消息单元中包含了Queue相关属性，这里要特别关注一下commitlog与queue间的关系是什么？

示意图

commitLog（mappedFile）文件的组成单元示意图

CommitLog 采用混合型存储，也就是所有 Topic 都存在一起，顺序追加写入，文件名用起始偏移量命名

消息先写入 CommitLog 再通过后台线程分发到 ConsumerQueue 和 IndexFile 中

消费者先读取 ConsumerQueue 得到真正消息的物理地址，然后访问 CommitLog 得到真正的消息

利用了 mmap 机制减少一次拷贝，利用文件预分配和文件预热提高性能

提供同步和异步刷盘，根据场景选择合适的机制

小结一下

commitlog 目录：其中存放着commitlog文件，而消息是写在commitlog文件中的

消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值

一个consumequeue文件中所有消息的Topic一定是相同的。但每条消息的Tag可能是不同的

注意看存储路径：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}，它已经明确了topic和queueId的

consumequeue文件可以看成是基于topic的commitlog索引文件，故consumequeue文件夹的组织方式如下：topic/queue/file三层组织结构，具体存储路径为：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样consumequeue文件采取定长设计，每一个条目共20个字节，分别为8字节的commitlog物理偏移量、4字节的消息长度、8字节tag hashcode，单个文件由30W个条目组成，可以像数组一样随机访问每一个条目，每个ConsumeQueue文件大小约5.72M（30w *20 字节）

consumequue 示意图

commitLog示意图

consumequeue目录下的一个文件的索引条目结构示意图

consumequeue的位置

consumequeue 目录：其中存放着consumequeue文件，队列就存放在这个目录中

index 目录：其中存放着消息索引文件indexFile

config 目录：存放着Broker运行期间的一些配置数据

lock 文件：运行期间使用到的全局资源锁

消息存储相关目录和文件

数据到达broker之后，将数据文件映射到OS的虚拟内存中（通过JDK NIO的MappedByteBuffer）

写消息的时候首先写入PageCache，并通过异步刷盘的方式将消息批量的做持久化（同时也支持同步刷盘），写到commitlog文件中（顺序写入，效率非常高）

消息先写入 CommitLog 再通过后台线程，将索引数据分发到 ConsumerQueue 和 IndexFile 中

图解过程

消息写入（一条消息进入到Broker后经历了以下几个过程才最终被持久化）

offset理解

请看演示内容

消费offset即消费进度，consumer对某个Queue的消费offset，即消费到了该Queue的第几条消息; 消息offset = 消费offset + 1

请看后面的消费模式

在$HOME/store/config/consumerOffset.json 文件中记录了每个topic下的queue的消费进度

consumer获取到其要消费消息所在Queue的消费偏移量offset，计算出其要消费消息的消息offset

Consumer向Broker发送拉取请求，其中会包含其要拉取消息的Queue、消息offset及消息Tag

消息位置

queueOffset = 消息offset * 20字节

Broker计算在该consumequeue中的queueOffset

从该queueOffset处开始向后查找第一个指定Tag的索引条目

解析该索引条目的前8个字节，即可定位到该消息在commitlog中的commitlog offset

从对应commitlog offset中读取消息单元，并发送给Consumer

消息拉取（当Consumer来拉取消息时会经历以下几个步骤）

RocketMQ中，无论是消息本身还是消息索引，都是存储在磁盘上的。其不会影响消息的消费吗？当然不会。其实RocketMQ的性能在目前的MQ产品中性能是非常高的。因为系统通过一系列相关机制大大提升了性能

首先，RocketMQ对文件的读写操作是通过mmap零拷贝进行的，将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率

PageCache机制，页缓存机制，是OS对文件的缓存机制，用于加速对文件的读写操作。一般来说，程序对文件进行顺序读写的速度几乎接近于内存读写速度，主要原因是由于OS使用 PageCache机制对读写访问操作进行性能优化，将一部分的内存用作PageCache

写入到commitlog中是顺序写入，效率极高

写操作：OS会先将数据写入到PageCache中，随后会以异步方式由pdæush（page dirty æush) 内核线程将Cache中的数据刷盘到物理磁盘

从commitlog读取消息是随机读取

读操作：若用户要读取数据，其首先会从PageCache中读取，若没有命中，则OS在从物理磁盘上加载该数据到PageCache的同时，也会顺序对其相邻数据块中的数据进行预读取

其次，consumequeue中的数据是顺序存放的，还引入了PageCache的预读取机制，使得对consumequeue文件的读取几乎接近于内存读取，即使在有消息堆积情况下也不会影响性能

RocketMQ中可能会影响性能的是对commitlog文件的读取。因为对commitlog文件来说，读取消息时会产生大量的随机访问，而随机访问会严重影响性能。不过，如果选择合适的系统IO调度算法，比如设置调度算法为Deadline（采用SSD固态硬盘的话），随机读的性能也会有所提升

性能的优势

文件的读写

消息存储

消息的存储

indexFile

注意：这个key是生产消息时指定的

注意：如果消息中没有包含key，则不会写入到indexFile文件中

消费者除了可以根据Topic进行消息消费外，RocketMQ还提供了根据key进行消息查询的功能。该查询是通过store目录中的index子目录中的indexFile进行索引实现的快速查询。当然，这个indexFile中的索引数据是在包含了key的消息被发送到Broker时写入的

indexFile 是什么？

文件名fileName是以创建时的时间戳命名的

存储位置：$HOME/store/index/{fileName}

每个indexFile文件由三部分构成：indexHeader，slots槽位，indexes索引数据

组成

indexHeader固定40个字节

8byte

也就是文件的创建时间

该indexFile中第一条消息的存储时间

beginTimestamp

该indexFile中最后一条消息存储时间

endTimestamp

该indexFile中第一条消息在commitlog中的偏移量commitlog offset

beginPhyoffset

该indexFile中最后一条消息在commitlog中的偏移量commitlog offset

endPhyoffset

4byte

已经填充有index的slot数量（并不是每个slot槽下都挂载有index索引单元，这里统计的是所有挂载了index索引单元的slot槽的数量）

hashSlotCount

该indexFile中包含的索引单元个数（统计出当前indexFile中所有slot槽下挂载的所有index索引单元的数量之和）

indexCount

indexHeader

index索引单元默写20个字节，其中存放着以下四个属性

消息中指定的业务key的hash值

phyOffset

keyHash

当前key对应的消息在commitlog中的偏移量commitlog offset

当前key对应消息的存储时间与当前indexFile创建时间的时间差

timeDiff

当前slot下当前index索引单元的前一个index索引单元的indexNo

slotValue(preIndexNo)

理解indexFile

index

存放的就是该slot槽中最新的一条索引单元的position

理解slot

slot

indexFile文件结构

查询流程

消费者从Broker中获取消息的方式有两种：pull拉取方式和push推动方式

消费者组对于消息消费的模式又分为两种：集群消费Clustering和广播消费Broadcasting

Consumer主动从Broker中拉取消息，主动权由Consumer控制。一旦获取了批量消息，就会启动消费过程。不过，该方式的实时性较弱，即Broker中有了新的消息时消费者并不能及时发现并消费

由于拉取时间间隔是由用户指定的，所以在设置该间隔时需要注意平稳：间隔太短，空请求比例会增加；间隔太长，消息的实时性太差

拉取式消费（pull）

该模式下Broker收到数据后会主动推送给Consumer。该获取方式一般实时性较高

默认就是 ConsumeType.CONSUME_PASSIVELY(\"PUSH\")

该获取方式是典型的发布-订阅模式，即Consumer向其关联的Queue注册了监听器，一旦发现有新的消息到来就会触发回调的执行，回调方法是Consumer去Queue中拉取消息。而这些都是基于Consumer 与Broker间的长连接的。长连接的维护是需要消耗系统资源的

推送式消费(push)

逻辑要用户自己实现

pull：需要应用去实现对关联Queue的遍历，实时性差；但便于应用控制消息的拉取

push：封装了对关联Queue的遍历，实时性强，但会占用较多的系统资源

对比

push和pull

消费类型

适用场景适用于消费端集群化部署，每条消息只需要被处理一次的场景

广播消费模式下，相同Consumer Group的每个Consumer实例都接收同一个Topic的全量消息。即每条消息都会被发送到Consumer Group中的每个Consumer

广播消费Broadcasting

适用场景适用于消费端集群化部署，每条消息需要被集群下的每个消费者处理的场景

注意：集群消费模式下，不保证每一次失败重投的消息路由到同一台机器上。

消费端集群消费模式下，相同Consumer Group的每个Consumer实例平均分摊同一个Topic的消息。即每条消息只会被发送到Consumer Group中的某个Consumer

集群消费Clustering

消费进度保存在consumer端。因为广播模式下consumer group中每个consumer都会消费所有消息，但它们的消费进度是不同。所以consumer各自保存各自的消费进度

广播模式

消费进度保存在broker中。consumer group中的所有consumer共同消费同一个Topic中的消息，同一条消息只会被消费一次。消费进度会参与到了消费的负载均衡中，故消费进度是需要共享的

消费进度

消息进度保存

阿里云

消费模式

集群消费下broker的下的 $HOME/store/config/consumerOffset.json 文件中保存了每个topic下的各个queue的消费进度

Rebalance机制讨论的前提是：集群消费

Rebalance(再均衡)机制指的是：将一个Topic下的多个队列，在同一个消费者组(consumer group)下的多个消费者实例(consumer instance)之间进行重新分配

Rebalance机制的本意是为了提升消息的并行消费能力。例如，⼀个Topic下5个队列，在只有1个消费者的情况下，这个消费者将负责消费这5个队列的消息。如果此时我们增加⼀个消费者，那么就可以给其中⼀个消费者分配2个队列，给另⼀个分配3个队列，从而提升消息的并行消费能力

什么是Rebalance

由于⼀个队列最多分配给⼀个消费者，因此当某个消费者组下的消费者实例数量大于队列的数量时，多余的消费者实例将分配不到任何队列

Rebalance限制

在只有一个Consumer时，其负责消费所有队列；在新增了一个Consumer后会触发Rebalance。此时原Consumer就需要暂停部分队列的消费，等到这些队列分配给新的Consumer后，这些暂停消费的队列才能继续被消费

消费暂停

同步提交：consumer提交了其消费完毕的一批消息的offset给broker后，需要等待broker的成功 ACK。当收到ACK后，consumer才会继续获取并消费下一批消息。在等待ACK期间，consumer 是阻塞的

异步提交：consumer提交了其消费完毕的一批消息的offset给broker后，不需要等待broker的成功ACK。consumer可以直接获取并消费下一批消息

对于一次性读取消息的数量，需要根据具体业务场景选择一个相对均衡的是很有必要的。因为数量过大，系统性能提升了，但产生重复消费的消息数量可能会增加；数量过小，系统性能会下降，但被重复消费的消息数量可能会减少

Consumer 在消费新分配给自己的队列时，必须接着之前Consumer 提交的消费进度的offset继续消费。然而默认情况下，offset是异步提交的，这个异步性导致提交到Broker的offset与Consumer实际消费的消息并不一致。这个不一致的差值就是可能会重复消费的消息

重复消费

由于Rebalance可能导致重复消费，如果需要重复消费的消息过多，或者因为Rebalance暂停时间过长从而导致积压了部分消息。那么有可能会导致在Rebalance结束之后瞬间需要消费很多消息

消费突刺

Rebalance危害

broker扩容或者缩容

Broker扩容或缩容

Broker升级运维

Broker与NameServer间的网络异常

Queue扩容或缩容

消费者所订阅Topic的Queue数量发生变化

Consumer Group扩容或缩容

Consumer升级运维

就意味着这个consumer就不存在了

Consumer与NameServer间网络异常

消费者组中消费者的数量发生变化

Rebalance产生的原因

rocketmq中的rebalance是consumer实例自身完成的

在Broker中维护着多个Map集合，这些集合中动态存放着当前Topic中Queue的信息、Consumer Group 中Consumer实例的信息。一旦发现消费者所订阅的Queue数量发生变化，或消费者组中消费者的数量发生变化，立即向Consumer Group中的每个实例发出Rebalance通知。

key是topic名称，value是TopicConfig。TopicConfig中维护着该Topic中所有Queue的数据。

TopicConfigManager

key是Consumser Group Id，value是ConsumerGroupInfo。 ConsumerGroupInfo中维护着该Group中所有Consumer实例数据

ConsumerManager

ConsumerOffsetManager

几个维护Map的类

Consumer实例在接收到通知后会采用Queue分配算法自己获取到相应的Queue，即由Consumer实例自主进行Rebalance

Rebalance过程

Rebalance机制

常见的有四种策略。这些策略是通过在创建Consumer时的构造器传进去的

一个Topic中的Queue只能由Consumer Group中的一个Consumer进行消费，而一个Consumer可以同时消费多个Queue中的消息。那么Queue与Consumer间的配对关系是如何确定的，即Queue要分配给哪个Consumer进行消费，也是有算法策略的

该算法是要根据avg = QueueCount / ConsumerCount 的计算结果进行分配的。如果能够整除，则按顺序将avg个Queue逐个分配Consumer；如果不能整除，则将多余出的Queue按照Consumer顺序逐个分配

平均分配策略

该算法不用事先计算每个Consumer需要分配几个Queue，直接一个一个分即可

环形平均策略是指，根据消费者的顺序，依次在由queue队列组成的环形图中逐个分配

环形平均策略

该算法存在的问题：分配不均

该策略会将consumer的hash值作为Node节点存放到hash环上，然后将queue的hash值也放到hash环上，通过顺时针方向，距离queue最近的那个consumer就是该queue要分配的consumer

一致性hash策略

该算法会根据queue的部署机房位置和consumer的位置，过滤出当前consumer相同机房的queue。然后按照平均分配策略或环形平均策略对同机房queue进行分配。如果没有同机房queue，则按照平均分配策略或环形平均策略对所有queue进行分配

同机房策略

两种平均分配策略的分配效率较高，一致性hash策略的较低。因为一致性hash算法较复杂。另外，一致性hash策略分配的结果也很大可能上存在不平均的情况

一致性hash算法存在的问题

其可以有效减少由于消费者组扩容或缩容所带来的大量的Rebalance

一致性hash算法存在的意义

Consumer数量变化较频繁的场景

一致性hash算法的应用场景

Queue分配算法

RocketMQ有一个原则：每条消息必须要被成功消费一次。

对于广播消费模式来说，Consumer本身就是消费进度记录器

记录在 $HOME/store/config/consumerOffset.json文件中

对于集群消费模式来说，Broker是消费进度记录器

什么是消费进度记录器？

Consumer在消费完消息后会向其消费进度记录器提交其消费消息的offset， offset被成功记录到记录器中，那么这条消费就被成功消费了

那么什么是成功消费呢？

至少一次原则

消息的消费

订阅关系的一致性指的是，同一个消费者组（Group ID相同）下所有Consumer实例所订阅的Topic与 Tag及对消息的处理逻辑必须完全一致。否则，消息消费的逻辑就会混乱，甚至导致消息丢失

多个消费者组订阅了多个Topic，并且每个消费者组里的多个消费者实例的订阅关系保持了一致

正确的订阅关系

正确订阅关系

订阅了不同的topic

订阅了不同Topic

同一个消费者组中的两个Consumer订阅了相同Topic的不同Tag

不同的tag

订阅了不同Tag

订阅失败

订阅了不同数量的Topic

一个消费者组订阅了多个Topic，但是该消费者组里的多个Consumer实例的订阅关系并没有保持一致

错误订阅关系代码

错误订阅关系

RocketMQ是支持一个消费者订阅多个topic，需要保证组内的消费者订阅的topic都必须一致，否则就会出现订阅的topic被覆盖的情况

订阅关系一致性

订阅关系的一致性

这里的offset指的是Consumer的消费进度offset。

消费进度2

消费进度offset是用来记录每个Queue的不同消费组的消费进度的。根据消费进度记录器的不同，可以分为两种模式：本地模式和远程模式

当消费模式为广播消费时，offset使用本地模式存储。因为每条消息会被所有的消费者消费，每个消费者管理自己的消费进度，各个消费者之间不存在消费进度的交集

可以通过 System.propeties key=rocketmq.client.localOffsetStoreDir 设置本地存储路径

在本地测试时可以将 consumer.setMessageModel(MessageModel.BROADCASTING) 设置为广播模式，然后可以看 C:\\Users\\odc-fyh\\.rocketmq_offsets\\172.12.91.33@DEFAULT\ocketmq_fyh_consumer_gp_3\\offsets.json

Consumer在广播消费模式下offset相关数据以json的形式持久化到Consumer本地磁盘文件中，默认文件路径为当前用户主目录下的.rocketmq_offsets/${clientId}/${group}/Offsets.json 。其中${clientId}为当前消费者id，默认为ip@DEFAULT；${group}为消费者组名称。

LocalFileOffsetStore

offset本地管理模式

当消费模式为集群消费时，offset使用远程模式管理。因为所有Cosnumer实例对消息采用的是均衡消费，所有Consumer共享Queue的消费进度

Broker启动时会加载这个文件，并写入到一个双层Map（ConsumerOffsetManager）外层map的key 为topic@group，value为内层map。内层map的key为queueId，value为offset。当发生Rebalance时，新的Consumer会从该Map中获取到相应的数据来继续消费

Consumer在集群消费模式下offset相关数据以json的形式持久化到Broker磁盘文件中，文件路径为当前用户主目录下的store/config/consumerOffset.json

集群模式下offset采用远程管理模式，主要是为了保证Rebalance机制

RemoteBrokerOffsetStore

offset远程管理模式

消费者是如何从最开始持续消费消息的？消费者要消费的第一条消息的起始位置是用户自己通过consumer.setConsumeFromWhere()方法指定的

CONSUME_FROM_LAST_OFFSET

CONSUME_FROM_FIRST_OFFSET

consumer.setConsumeTimestamp(“20210701080000”) yyyyMMddHHmmss

一个新的订阅组第一次启动从指定时间点开始消费，后续再启动接着上次消费的进度开始消费

CONSUME_FROM_TIMESTAMP

就是你配置啥都不会生效

对于已经存在的消费者组+topic+queue的订阅关系，无论如何都是遵循历史进度进行消费

取决于你配置了哪个策略

对于新的消费者组+topic+queue关系，在正常情况下，遵循客户端配置的策略

对于特殊的场景被broker认为queue是新queue的情况下，一律从头开始消费（令可杀错不放过）

关于第一二参数的说明

在Consumer启动后，其要消费的第一条消息的起始位置常用的有三种，这三种位置可以通过枚举类型常量设置。这个枚举类型为ConsumeFromWhere

当消费完一批消息后，Consumer会提交其消费进度offset给Broker，Broker在收到消费进度后会将其更新到那个双层Map（ConsumerOffsetManager）及consumerOffset.json文件中，然后向该Consumer进行ACK，而ACK内容中包含三项数据：当前消费队列的最小offset（minOffset）、最大offset（maxOffset）、及下次消费的起始offset（nextBeginOffset）

offset用途

当rocketMQ对消息的消费出现异常时，会将发生异常的消息的offset提交到Broker中的重试队列。系统在发生消息消费异常时会为当前的topic@group创建一个重试队列，该队列以%RETRY%开头，到达重试时间后进行消费重试

重试队列

消费者在消费完一批消息后会向broker提交这些消息的offset，然后等待broker的成功响应。若在等待超时之前收到了成功响应，则继续读取下一批消息进行消费（从ACK中获取nextBeginOffset）。若没有收到响应，则会重新提交，直到获取到响应。而在这个等待过程中，消费者是阻塞的。其严重影响了消费者的吞吐量

同步提交

默认是异步提交

消费者在消费完一批消息后向broker提交offset，但无需等待Broker的成功响应，可以继续读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意，broker在收到提交的offset后，还是会向消费者进行响应的。可能还没有收到ACK，此时Consumer会从Broker中直接获取nextBeginOffset

异步提交

集群消费模式下，Consumer消费完消息后会向Broker提交消费进度offset，其提交方式分为两种

offset的同步提交与异步提交

offset管理

在互联网应用中，尤其在网络不稳定的情况下，消息很有可能会出现重复发送或重复消费。如果重复的消息可能会影响业务处理，那么就应该对消息做幂等处理

当出现消费者对某条消息重复消费的情况时，重复消费的结果与消费一次的结果是相同的，并且多次消费并未对业务系统产生任何负面影响，那么这个消费过程就是消费幂等的

幂等：若某操作执行多次与执行一次对系统产生的影响是相同的，则称该操作是幂等的

什么是消费幂等?

这里可能会对 Message Id 相同感到疑惑，因为在最前面有说过它的生产规则。注意，它不是消息生产者再次生产相同的消息再次发送，而是将之前发送失败的消息拿过来重新发送，所以和之前的消息是一模一样的

当然也有可能消费者主动重发，这样就可能会存在相同的消息有两个不同msgId的情况

当一条消息已被成功发送到Broker并完成持久化，此时出现了网络闪断，从而导致Broker对Producer应答失败。如果此时Producer意识到消息发送失败并尝试再次发送消息，此时Broker中就可能会出现两条内容相同并且Message ID也相同的消息，那么后续Consumer就一定会消费两次该消息

发送时消息重复

消息已投递到Consumer并完成业务处理，当Consumer给Broker反馈应答时网络闪断，Broker没有接收到消费成功响应。为了保证消息至少被消费一次的原则，Broker将在网络恢复后再次尝试投递之前已被处理过的消息。此时消费者就会收到与之前处理过的内容相同、Message ID也相同的消息

消费时消息重复

请看前面的rebalance机制

当Consumer Group中的Consumer数量发生变化时，或其订阅的Topic的Queue数量发生变化时，会触发Rebalance，此时Consumer可能会收到曾经被消费过的消息

Rebalance时消息重复

消息重复的场景

幂等令牌：是生产者和消费者两者中的既定协议，通常指具备唯⼀业务标识的字符串。例如，订单号、流水号。一般由Producer随着消息一同发送来的

唯一性处理：RocketMQ无法避免消息重复（Exactly-Once），所以如果业务对消费重复非常敏感，消费端务必在业务层面进行去重处理。可以借助关系数据库进行去重

幂等解决方案的设计中涉及到两项要素：幂等令牌，与唯一性处理。只要充分利用好这两要素，就可以设计出好的幂等解决方案

两要素

1. 首先通过缓存去重。在缓存中如果已经存在了某幂等令牌，则说明本次操作是重复性操作；若缓存没有命中，则进入下一步

2. 在唯一性处理之前，先在数据库中查询幂等令牌作为索引的数据是否存在。若存在，则说明本次操作为重复性操作；若不存在，则进入下一步

3. 在同一事务中完成三项操作：唯一性处理后，将幂等令牌写入到缓存，并将幂等令牌作为唯一索引的数据写入到DB中

对于常见的系统，幂等性操作的通用性解决方案是

1. 当支付请求到达后，首先在Redis缓存中却获取key为支付流水号的缓存value。若value不空，则说明本次支付是重复操作，业务系统直接返回调用侧重复支付标识；若value为空，则进入下一步操作

2. 到DBMS中根据支付流水号查询是否存在相应实例。若存在，则说明本次支付是重复操作，业务系统直接返回调用侧重复支付标识；若不存在，则说明本次操作是首次操作，进入下一步完成唯一性处理

完成支付任务

将当前支付流水号作为主键，与其它相关数据共同写入到DBMS

3. 在分布式事务中完成三项操作

解决方案举例：以支付场景为例

解决方案

如何解决消息重复

消费幂等的解决方案很简单：为消息指定不会重复的唯一标识。因为Message ID有可能出现重复的情况，所以真正安全的幂等处理，不建议以Message ID作为处理依据。最好的方式是以业务唯一标识作为幂等处理的关键依据，而业务的唯一标识可以通过消息Key设置

消费幂等的实现

RocketMQ能够保证消息不丢失，但不能保证消息不重复

可以，但没必要

rocketmq 能否解决消息重复呢？

xiaofei midng

消费幂等

业务系统上下游能力不匹配造成的持续堆积，且无法自行恢复

业务系统对消息的消费实时性要求较高，即使是短暂的堆积造成的消费延迟也无法接受

以下场景需要重点关注消息堆积和消费延迟问题

消息处理流程中，如果Consumer的消费速度跟不上Producer的发送速度，MQ中未处理的消息会越来越多（进的多出的少），这部分消息就被称为堆积消息。消息出现堆积进而会造成消息的消费延迟

概念理解

阿里云文档

参考阿里云文档

SDK获取消息的方式为批量拉取，常见内网环境下都会有很高的吞吐量，例如：1个单线程单分区的低规格机器（4C8GB）可以达到几万TPS，如果是多个分区可以达到几十万TPS。所以这一阶段一般不会成为消息堆积的瓶颈

阶段一：获取消息，SDK客户端通过长轮询批量拉取的方式从消息队列RocketMQ版服务端获取消息，将拉取到的消息缓存到本地缓冲队列中。

阶段二：提交消费线程，SDK客户端将本地缓存的消息提交到消费线程中，使用业务消费逻辑进行处理。此时客户端的消费能力就完全依赖于业务逻辑的复杂度（消费耗时）和消费逻辑并发度了。如果业务处理逻辑复杂，处理单条消息耗时都较长，则整体的消息吞吐量肯定不会高，此时就会导致客户端本地缓冲队列达到上限，停止从服务端拉取消息

消费原理图

SDK客户端使用Push模式消费消息时，分为以下两个阶段：

通过以上客户端消费原理可以看出，消息堆积的主要瓶颈在于本地客户端的消费能力，即消费耗时和消费并发度。想要避免和解决消息堆积问题，必须合理的控制消费耗时和消息并发度，其中消费耗时的优先级高于消费并发度，必须先保证消费耗时的合理性，再考虑消费并发度问题

客户端消费原理

读写外部数据库，例如MySQL数据库读写

读写外部缓存等系统，例如Redis读写

下游系统调用，例如Dubbo调用或者下游HTTP接口调用

影响消费耗时的消费逻辑主要分为CPU内存计算和外部I/O操作，通常情况下代码中如果没有复杂的递归和循环的话，内部计算耗时相对外部I/O操作来说几乎可以忽略。外部I/O操作通常包括如下业务逻辑

这类外部调用的逻辑和系统容量您需要提前梳理，掌握每个调用操作预期的耗时，这样才能判断消费逻辑中I/O操作的耗时是否合理。通常消费堆积都是由于这些下游系统出现了服务异常、容量限制导致的消费耗时增加

例如：某业务消费逻辑中需要写一条数据到数据库，单次消费耗时为1 ms，平时消息量小未出现异常。业务侧进行大促活动时，写数据库TPS爆发式增长，并很快达到数据库容量限制，导致消费单条消息的耗时增加到100 ms，业务侧可以明显感受到消费速度大幅下跌。此时仅通过调整消息队列RocketMQ版SDK的消费并发度并不能解决问题，需要对数据库容量进行升配才能从根本上提高客户端消费能力

消费耗时

单节点线程数，即单个Consumer所包含的线程数量

节点数量就是consumer group 中的 consumer 个数

其值为单节点线程数*节点数量

对于普通消息、延时消息及事务消息，并发度计算都是= (单节点线程数*节点数量)

该类型消息的Topic只有一个Queue分区。其可以保证该Topic的所有消息被顺序消费。为了保证这个全局顺序性，Consumer Group中在同一时刻只能有一个Consumer的一个线程进行消费。所以其并发度为1

全局顺序消息

该类型消息的Topic有多个Queue分区。其仅可以保证该Topic的每个Queue 分区中的消息被顺序消费，不能保证整个Topic中消息的顺序消费。为了保证这个分区顺序性，每个Queue分区中的消息在Consumer Group中的同一时刻只能有一个Consumer的一个线程进行消费。即，在同一时刻最多会出现多个Queue分蘖有多个Consumer的多个线程并行消费。所以其并发度为Topic的分区数量。

分区顺序消息

顺序消息的消费并发度等于 Min(单节点线程数*节点数量，分区数）数量

客户端消费并发度由单节点线程数和节点数量共同决定，一般情况下需要优先调整单节点的线程数，若单机硬件资源达到上限，则必须通过扩容节点来提高消费并发度。

阿里云并发度文档

消费并发度

单机vCPU核数为C

线程切换耗时忽略不计，I/O操作不消耗CPU

线程有足够消息等待处理，且内存充足

逻辑中CPU计算耗时为T1，外部I/O操作为T2

单节点的并发度需要谨慎设置，不能盲目直接调大线程数，设置过大的线程数反而会带来大量的线程切换的开销。理想环境下单节点的最优线程数计算模型如下

里计算的最大线程数仅仅是在理想环境下得到的理论数据，实际应用环境中建议逐步调大线程数并观察效果再进行调整。

则单个线程能达到的TPS为1/（T1+T2），如果CPU使用率达到理想状态100%，那么单机达到最大能力时需要设置 C*（T1+T2）/T1 个线程

单机线程数计算

消息消费逻辑的计算复杂度是否过高，代码是否存在无限循环和递归等缺陷。

消息消费逻辑中的I/O操作（如：外部调用、读写存储等）是否是必须的，能否用本地缓存等方案规避。

消费逻辑中的复杂耗时的操作是否可以做异步化处理，如果可以是否会造成逻辑错乱（消费完成但异步操作未完成）。

梳理消息的消费耗时

逐步调大线程的单个节点的线程数，并观测节点的系统指标，得到单个节点最优的消费线程数和消息吞吐量

节点数就是consumer group 中 consumer的个数，就是增加消费者机器

得到单个节点的最优线程数和消息吞吐量后，根据上下游链路的流量峰值计算出需要设置的节点数，节点数=流量峰值/单线程消息吞吐量。

设置消息的消费并发度

了避免在业务使用时出现非预期的消息堆积和延迟问题，您需要在前期设计阶段对整个业务逻辑进行完善的排查和梳理。整理出正常业务运行场景下的性能基线，才能在故障场景下迅速定位到阻塞点。其中最重要的就是梳理消息的消费耗时和消息消费的并发度

如何避免消息堆积和延迟

消息堆积与消费延迟

消息被消费过后会被清理掉吗？不会的

消息是被顺序存储在commitlog文件的，且消息大小不定长，所以消息的清理是不可能以消息为单位进行清理的，而是以commitlog文件为单位进行清理的。否则会急剧下降清理效率，并实现逻辑复杂

commitlog文件存在一个过期时间，默认为72小时，即三天

文件已经超过3天，但是假如此时是凌晨2点，还没有到达清理时间，是不会清理的，等到了4点在清理

文件过期，且到达清理时间点（默认为凌晨4点）后，自动清理过期文件

文件过期，且磁盘空间占用率已达过期清理警戒线（默认75%）后，无论是否达到清理时间点，都会自动清理过期文件

磁盘占用率达到清理警戒线（默认85%）后，开始按照设定好的规则清理文件，无论是否过期。默认会从最老的文件开始清理

磁盘占用率达到系统危险警戒线（默认90%）后，Broker将拒绝消息写入

除了用户手动清理外，在以下情况下也会被自动清理，无论文件中的消息是否被消费过

对于RocketMQ系统来说，删除一个1G大小的文件，是一个压力巨大的IO操作。在删除过程中，系统性能会骤然下降。所以，其默认清理时间点为凌晨4点，访问量最小的时间。也正因如果，我们要保障磁盘空间的空闲率，不要使系统出现在其它时间点删除commitlog文件的情况

官方建议RocketMQ服务的Linux文件系统采用ext4。因为对于文件删除操作，ext4要比ext3性能更好

需要注意以下几点

消息的清理

3.工作原理

说明：如果使用的不是阿里云的ECS服务器请忽略

说明：我使用的是单机

10909

开放Rocketmq使用到的端口

默认是8080，但是一般都是修改为一个不常用的

在conf/broker.conf 中加入 brokerIP1=你的公网IP

配置

但是实际上是无法连同的，需要做一下修改，请看下面

参考文档：https://blog.csdn.net/qq_21460229/article/details/104351178

这是官网提供的启动方法

先启动nameserver

再启动broker

使用的是单机部署

阿里云配置

同步发送消息是指，Producer发出⼀条消息后，会在收到MQ返回的ACK之后才发下⼀条消息。该方式的消息可靠性最高，但消息发送效率太低

同步发送消息

异步发送消息是指，Producer发出消息后无需等待MQ返回ACK，直接发送下⼀条消息。该方式的消息可靠性可以得到保障，消息发送效率也可以

异步发送消息

单向发送消息是指，Producer仅负责发送消息，不等待、不处理MQ的ACK。该发送方式时MQ也不返回ACK。该方式的消息发送效率最高，但消息可靠性较差

单向发送消息

在 com.quguan.rocketmq.general 包下

gitee代码地址

代码演示地址

普通消息

默认情况下生产者会把消息以Round Robin轮询方式发送到不同的Queue分区队列；而消费消息时会从多个Queue上拉取消息，这种情况下的发送和消费是不能保证顺序的。如果将消息仅发送到同一个Queue中，消费时也只从这个Queue上拉取消息，就严格保证了消息的顺序性

顺序消息指的是，严格按照消息的发送顺序进行消费的消息(FIFO)

什么是顺序消息?

顺序消息

为什么需要顺序消息?

有序性分类

代码地址仓库

监听其他消息的接口是 MessageListenerConcurrently

监听顺序消息的接口是 MessageListenerOrderly

当消息写入到Broker后，在指定的时长后才可被消费处理的消息，称为延时消息

什么是延时消息？

采用RocketMQ的延时消息可以实现定时任务的功能，而无需使用定时器。典型的应用场景是，电商交易中超时未支付关闭订单的场景，12306平台订票超时未支付取消订票的场景

在电商平台中，订单创建时会发送一条延迟消息。这条消息将会在30分钟后投递给后台业务系统（Consumer），后台业务系统收到该消息后会判断对应的订单是否已经完成支付。如果未完成，则取消订单，将商品再次放回到库存；如果完成支付，则忽略

在12306平台中，车票预订成功后就会发送一条延迟消息。这条消息将会在45分钟后投递给后台业务系统（Consumer），后台业务系统收到该消息后会判断对应的订单是否已经完成支付。如果未完成，则取消预订，将车票再次放回到票池；如果完成支付，则忽略

延时消息的作用

messageDelayLevel = \"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h\"

等级分别是1-18，若指定的延时等级为3，则表示延迟时长为10s，若延迟等级为18，则表示延时2h

延时消息的延迟时长不支持随意时长的延迟，是通过特定的延迟等级来指定的。延时等级定义在RocketMQ服务端的 MessageStoreConfig 类中的如下变量中：

延迟等级

延时等级

Producer将消息发送到Broker后，Broker会首先将消息写入到commitlog文件，然后需要将其分发到相应的consumequeue。不过，在分发之前，系统会先判断消息中是否带有延时等级。若没有，则直接正常分发；若有则需要经历一个复杂的过程

XXXX 多少有点随意了

第一次没有就去创建，以后都使用这一个

注意：所有的延时消息都使用这一个TOPIC

为了不让消费者去消费，等到期了在消费

为什么要修改topic的名字？

新建topic

1.修改消息的Topic为SCHEDULE_TOPIC_XXXX

延迟等级delayLevel与queueId的对应关系为queueId = delayLevel -1

假如我的延时等级是3，则在SCHEDULE_TOPIC_XXXX 主题下创建一个 queueId=3-1=2的目录

需要注意，在创建queueId目录时，并不是一次性地将所有延迟等级对应的目录全部创建完毕，而是用到哪个延迟等级创建哪个目录

3.根据延时等级，在consumequeue目录中SCHEDULE_TOPIC_XXXX主题下创建出相应的queueId目录与consumequeue文件

索引单元存储内容

消息索引单元存储内容

4.修改消息索引单元内容。索引单元中的Message Tag HashCode部分原本存放的是消息的Tag的 Hash值。现修改为消息的投递时间。投递时间是指该消息被重新修改为原Topic后再次被写入到commitlog中的时间。投递时间 = 消息存储时间 + 延时等级时间。消息存储时间指的是消息被发送到Broker时的时间戳

是按照消息投递时间排序的。一个Broker中同一等级的所有延时消息会被写入到consumequeue 目录中SCHEDULE_TOPIC_XXXX目录下相同Queue中。即一个Queue中消息投递时间的延迟等级时间是相同的。那么投递时间就取决于于消息存储时间了。即按照消息被发送到Broker的时间进行排序的。

SCHEDULE_TOPIC_XXXX目录中各个延时等级Queue中的消息是如何排序的？

5.将消息索引写入到SCHEDULE_TOPIC_XXXX主题下相应的consumequeue中的文件中

延迟消息原理

延迟队列

ScheuleMessageService在Broker启动时(BrokerStartup)，会创建并启动一个定时器TImer，用于执行相应的定时任务。系统会根据延时等级的个数，定义相应数量的TimerTask，每个TimerTask负责一个延迟等级消息的消费与投递。每个TimerTask都会检测相应Queue队列的第一条消息是否到期。若第一条消息未到期，则后面的所有消息更不会到期（消息是按照投递时间排序的）；若第一条消息到期了，则将该消息投递到目标Topic，即消费该消息

投递延时消息

延时消息

所谓事务消息应该理解成 MQ的事务，而不是开发人员所理解的事务，给你数据库进行回滚之类的。这里的事务是保证消息能正常发送和消费，是这个事务

事务消息

参考官网

1. 工行系统发送一个给B增款1万元的同步消息M给Broker

2. 消息被Broker成功接收后，向工行系统发送成功ACK

3. 工行系统收到成功ACK后从用户A中扣款1万元

4. 建行系统从Broker中获取到消息M

5. 建行系统消费消息M，即向用户B中增加1万元

转账

这里的一个需求场景是：工行用户A向建行用户B转账1万元

这其中是有问题的：若第3步中的扣款操作失败，但消息已经成功发送到了Broker。对于MQ来说，只要消息写入成功，那么这个消息就可以被消费。此时建行系统中用户B增加了1万元。出现了数据不一致问题

问题

解决思路是，让第1、2、3步具有原子性，要么全部成功，要么全部失败。即消息发送成功后，必须要保证扣款成功。如果扣款失败，则回滚发送成功的消息。而该思路即使用事务消息。这里要使用分布式事务解决方案

如何解决？

这个事务管理器TM 就是我们的消息生产者，后面在说

TC 就是Broker

1. 事务管理器TM向事务协调器TC发起指令，开启全局事务

2. 工行系统发一个给B增款1万元的事务消息M给TC

3. TC会向Broker发送半事务消息prepareHalf，将消息M预提交到Broker。此时的建行系统是看不到Broker中的消息M的

4. Broker会将预提交执行结果上报给TC。

5. 如果预提交失败，则TC会向TM上报预提交失败的响应，全局事务结束；如果预提交成功，TC会调用工行系统的回调操作，去完成工行用户A的预扣款1万元的操作

6. 工行系统会向TC发送预扣款执行结果，即本地事务的执行状态

本地事务执行成功

LocalTransactionState#COMMIT_MESSAGE

本地事务执行失败

LocalTransactionState#ROLLBACK_MESSAGE

不确定，表示需要进行回查以确定本地事务的执行结果

LocalTransactionState#UNKNOW

执行结果就是告诉TM本地事务执行结果，是成功了还是失败了，有三种状态

7. TC收到预扣款执行结果后，会将结果上报给TM。

若预扣款成功（本地事务状态为COMMIT_MESSAGE），则TM向TC发送Global Commit指令

若预扣款失败（本地事务状态为ROLLBACK_MESSAGE），则TM向TC发送Global Rollback指令

回查操作会将回查结果，即COMMIT_MESSAGE或ROLLBACK_MESSAGE 上报给TC。TC将结果上报给TM，TM会再向TC发送最终确认指令Global Commit或Global Rollback

若现未知状态（本地事务状态为UNKNOW），则会触发工行系统的本地事务状态回查操作

8. TM会根据上报结果向TC发出不同的确认指令

TC接收的若是Global Commit指令，则向Broker与工行系统发送Branch Commit指令。此时Broker中的消息M才可被建行系统看到；此时的工行用户A中的扣款操作才真正被确认

TC接收到的若是Global Rollback指令，则向Broker与工行系统发送Branch Rollback指令。此时Broker中的消息M将被撤销；工行用户A中的扣款操作将被回滚

9. TC在接收到指令后会向Broker与工行系统发出确认指令

以上方案就是为了确保消息投递与扣款操作能够在一个事务中，要成功都成功，有一个失败，则全部回滚。

分布式事务解决转账

分布式事务解决

问题引入

对于分布式事务，通俗地说就是，一次操作由若干分支操作组成，这些分支操作分属不同应用，分布在不同服务器上。分布式事务需要保证这些分支操作要么全部成功，要么全部失败。分布式事务与普通事务一样，就是为了保证操作结果的一致性

分布式事务

RocketMQ提供了类似X/Open XA的分布式事务功能，通过事务消息能达到分布式事务的最终一致。XA是一种分布式事务解决方案，一种分布式事务处理模式

暂不能消费的消息，发送方已经成功地将消息发送到了Broker，但是Broker未收到最终确认指令，此时该消息被标记成“暂不能消费”状态，即不能被消费者看到。处于该种状态下的消息即半事务消息

半事务消息

Producer回调操作执行的结果为本地事务状态，其会发送给TC，而TC会再发送给TM。TM会根据TC发送来的本地事务状态来决定全局事务确认指令

本地事务状态

注意，消息回查不是重新执行回调操作。回调操作是进行预扣款操作，而消息回查则是查看预扣款操作执行的结果

回调操作返回UNKNWON

TC没有接收到TM的最终全局事务确认指令

引发消息回查的原因最常见的有两个

消息回查，即重新查询本地事务的执行状态。本例就是重新到DB中查看预扣款操作是否执行成功。

关于消息回查，有三个常见的属性设置。它们都在broker加载的配置文件中设置

transactionTimeout=20，指定TM在20秒内应将最终确认状态发送给TC，否则引发消息回查。默认为60秒

transactionCheckMax=5，指定最多回查5次，超过后将丢弃消息并记录错误日志。默认15次。

transactionCheckInterval=10，指定设置的多次消息回查的时间间隔为10秒。默认为60秒

RocketMQ中的消息回查设置

消息回查

XA（Unix Transaction）是一种分布式事务解决方案，一种分布式事务处理模式，是基于XA协议的。XA协议由Tuxedo（Transaction for Unix has been Extended for Distributed Operation，分布式操作扩展之后的Unix事务系统）首先提出的，并交给X/Open组织，作为资源管理器与事务管理器的接口标准

XA协议

Transaction Coordinator，事务协调者。维护全局和分支事务的状态，驱动全局事务提交或回滚

RocketMQ中Broker充当着TC

Transaction Manager，事务管理器。定义全局事务的范围：开始全局事务、提交或回滚全局事务。它实际是全局事务的发起者

RocketMQ中事务消息的Producer充当着TM

Resource Manager，资源管理器。管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚

RocketMQ中事务消息的Producer及Broker均是RM

XA模式中有三个重要组件：TC、TM、RM

1. TM向TC发起指令，开启一个全局事务。

2. 根据业务要求，各个RM会逐个向TC注册分支事务，然后TC会逐个向RM发出预执行指令。

3. 各个RM在接收到指令后会在进行本地事务预执行。

4. RM将预执行结果上报给TC。当然，这个结果可能是成功，也可能是失败。

若所有结果都是成功响应，则向TC发送Global Commit指令。

只要有结果是失败响应，则向TC发送Global Rollback指令

5. TC在接收到各个RM的Report后会将汇总结果上报给TM，根据汇总结果TM会向TC发出确认指令

6. TC在接收到指令后再次向RM发送确认指令。

XA模式是一个典型的2PC，其执行原理如下

XA模式架构

XA模式

事务消息不支持延时消息和批量消息

用户可以通过重写 AbstractTransactionalMessageCheckListener 类来修改这个行为

为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的 transactionCheckMax参数来修改此限制。如果已经检查某条消息超过 N 次的话（ N = transactionCheckMax ）则 Broker 将丢弃此消息，并在默认情况下同时打印错误日志。

事务消息将在 Broker 配置文件中的参数 transactionTimeout 这样的特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionTimeout 参数

对于事务消息要做好幂等性检查，因为事务消息可能不止一次被消费（因为存在回滚后再提交的情况）

提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制

事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者

事务消息使用上的限制

批量发送的消息必须具有相同的Topic

批量发送的消息必须具有相同的刷盘策略

批量发送的消息不能是延时消息与事务消息

生产者消息发送时可以一次发送多条消息，这可以大大提升Producer的发送效率。不过需要注意以下几点

发送限制

方案一：将批量消息进行拆分，拆分为若干不大于4M的消息集合分多次批量发送

Producer端需要在发送之前设置Producer的maxMessageSize属性

Broker端需要修改其加载的配置文件中的maxMessageSize属性

方案二：在Producer端与Broker端同时修改属性

默认情况下，一批发送的消息总大小不能超过4MB字节。如果想超出该值，有两种解决方案

批量发送大小

组成结构

生产者发送消息组成结构

批量发送消息

Consumer消费时注册的监听接口MessageListenerConcurrently，监听接口的consumeMessage()方法的第一个参数为消息列表，但默认情况下每次只能消费一条消息。若要使其一次可以消费多条消息，则可以通过修改Consumer的consumeMessageBatchMaxSize属性来指定。不过，该值不能超过32。因为默认情况下消费者每次可以拉取的消息最多是32条。若要修改一次拉取的最大值，则可通过修改Consumer的 pullBatchSize属性来指定

默认值是1，表示从broker拉取消息后，一次性给多少条消息给Consumer去消费

Consumer#setConsumeMessageBatchMaxSize(10);

默认值是32，表示从Broker拉取32条消息后给消费者消费

Consumer#setPullBatchSize(40)

修改批量属性

当然不是！

pullBatchSize值设置的越大，Consumer每拉取一次需要的时间就会越长，且在网络上传输出现问题的可能性就越高。若在拉取过程中若出现了问题，那么本批次所有消息都需要全部重新拉取

consumeMessageBatchMaxSize值设置的越大，Consumer的消息并发消费能力越低，且这批被消费的消息具有相同的消费结果。因为consumeMessageBatchMaxSize指定的一批消息只会使用一个线程进行处理，且在处理过程中只要有一个消息处理异常，则这批消息需要全部重新再次消费处理

Consumer的pullBatchSize属性与consumeMessageBatchMaxSize属性是否设置的越大越好？

存在的问题

批量消费消息

批量消息代码

批量消息

对于指定Topic消息的过滤有两种过滤方式：Tag过滤与SQL过滤

消息者在进行消息订阅时，除了可以指定要订阅消息的Topic外，还可以对指定Topic中的消息根据指定条件进行过滤，即可以订阅比Topic更加细粒度的消息类型

什么是消息过滤？

通过consumer的subscribe()方法指定要订阅消息的Tag。如果订阅多个Tag的消息，Tag间使用或运算符(双竖线||)连接

Tag过滤

注意：只有使用PUSH模式的消费者才能使用SQL过滤

SQL过滤是一种通过特定表达式对事先埋入到消息中的用户属性进行筛选过滤的方式。通过SQL过滤，可以实现对消息的复杂过滤。

数值：比如：123，3.1415

字符：必须用单引号包裹起来，比如：'abc'

布尔：TRUE 或 FALSE

NULL：特殊的常量，表示空

支持的常量类型

数值比较：>，>=，<，<=，BETWEEN，=

字符比较：=，<>，IN

逻辑运算：AND，OR，NOT

NULL判断：IS NULL 或者 IS NOT NULL

支持的运算符有

SQL过滤表达式中支持多种常量类型与运算符

enablePropertyFilter = true

默认情况下Broker没有开启消息的SQL过滤功能，需要在Broker加载的配置文件中添加如下属性，以开启该功能

SQL过滤

消息过滤代码

消息过滤

Producer对发送失败的消息进行重新发送的机制，称为消息发送重试机制，也称为消息重投机制

什么是消息重试?

生产者在发送消息时，若采用同步或异步发送方式，发送失败会重试，但oneway消息发送方式发送失败是没有重试机制的

对于同步发送顺序消息，我觉得也是可以重试的

对于异步发送顺序消息，这个不可以重试，因为可能会破坏顺序关系

原因是什么？

只有普通消息具有发送重试机制，顺序消息是没有的

消息重投机制可以保证消息尽可能发送成功、不丢失，但可能会造成消息重复。消息重复在RocketMQ中是无法避免的问题

消息重复在一般情况下不会发生，当出现消息量大、网络抖动，消息重复就会成为大概率事件producer主动重发、consumer负载变化（发生Rebalance，不会导致消息重复，但可能出现重复消费）也会导致重复消息

避免消息重复消费的解决方案是，为消息添加唯一标识（例如消息key），使消费者对消息进行消费判断来避免重复消费

消息重复无法避免，但要避免消息的重复消费。

消息重试的注意事项

retryTimesWhenSendFailed 默认是2

同步重试次数 = retryTimesWhenSendFailed + 1

方案一：Producer中维护Map集合，其key是发生失败的时间戳，value为Broker实例。Producer中还维护着一个Set集合，其中存放着所有未发生发送异常的Broker实例。选择目标Broker是从该Set集合中选择的。再定义一个定时任务，定期从Map集合中将长期未发生发送异常的Broker清理出去，并添加到Set集合

方案二：为Producer中的Broker实例添加一个标识，例如是一个AtomicBoolean属性。只要该Broker上发生过发送异常，就将其置为true。选择目标Broker就是选择该属性值为false的 Broker。再定义一个定时任务，定期将Broker的该属性置为false

方案三：为Producer中的Broker实例添加一个标识，例如是一个AtomicLong属性。只要该Broker上发生过发送异常，就使其值增一。选择目标Broker就是选择该属性值最小的Broker。若该值相同，采用轮询方式选择

思考：让我们自己实现失败隔离功能，如何来做？

Producer#setSendLatencyFaultEnable(true);

失败隔离机制默认是关闭的，如果要使用这个机制，我们需要在发送消息时手动打开

对于普通消息，消息发送默认采用round-robin策略来选择所发送到的队列。如果发送失败，默认重试3次。但在重试时是不会选择上次发送失败的Broker，而是选择其它Broker，这是因为它具有失败隔离功能，使Producer尽量选择未发生过发送失败的Broker作为目标Broker。其可以保证其它消息尽量不发送到问题Broker，为了提升消息发送效率，降低消息发送耗时

当然，若只有一个Broker其也只能发送到该Broker，但其会尽量发送到该Broker上的其它Queue

异常重试，当发生 RemotingException，MQClientException， MQBrokerException 异常时都会发生重试

Producer#setRetryAnotherBrokerWhenNotStoreOK(true)

这种情况下默认时不会重试的，如果想要重试，则修改发送消息时作如下修改

同步发送失败策略

异步发送失败重试时，异步重试不会选择其他broker，仅在同一个broker上做重试，所以该策略无法保证消息不丢

Producer#setRetryTimesWhenSendAsyncFailed(3);

异步发送失败策略

消息刷盘超时（Master或Slave）或slave不可用（slave在做数据同步时向master返回状态不是SEND_OK）时，默认是不会将消息尝试发送到其他Broker的。不过，对于重要消息可以通过在Broker的配置文件设置retryAnotherBrokerWhenNotStoreOK属性为true来开启

消息刷盘失败策略

消息发送重试三种策略

消息发送重试机制

对于顺序消息，当Consumer消费消息失败后，为了保证消息的顺序性，其会自动不断地进行消息重试，直到消费成功。消费重试默认间隔时间为1000毫秒。重试期间应用会出现消息消费被阻塞的情况

顺序消息重试间隔

Consumer#setSuspendCurrentQueueTimeMillis(2000)

它的重试间隔是通过时间来控制的

对顺序消息的重试是无休止的，不间断的，直至消费成功，所以，对于顺序消息的消费，务必要保证应用能够及时监控并处理消费失败的情况，避免消费被永久性阻塞

注意，顺序消息没有发送失败重试机制，但具有消费失败重试机制

顺序消息的消费重试

不过需要注意，无序消息的重试只对集群消费方式生效，广播消费方式不提供失败重试特性

对于广播消费，消费失败后，失败消息不再重试，继续消费后续消息

对于无序消息（普通消息、延时消息、事务消息），当Consumer消费消息失败时，可以通过设置返回状态达到消息重试的效果

一条消息无论重试多少次，这些重试消息的 Message ID 不会改变

无序消息的消费重试

对于无序消息集群消费下的重试消费，每条消息默认最多重试16次，但每次重试的间隔时间是不同的，会逐渐变长

若一条消息在一直消费失败的前提下，将会在正常消费后的第 4小时46分后进行第16次重试。若仍然失败，则将消息投递到死信队列

重试间隔

重试间隔如图

默认值是16

若修改值小于16，则按照指定间隔进行重试

若修改值大于16，则超过16次的重试时间间隔均为2小时

maxReconsumeTimes

修改重试次数

对于Consumer Group，若仅修改了一个Consumer的消费重试次数，则会应用到该Group中所有其它Consumer实例。若出现多个Consumer均做了修改的情况，则采用覆盖方式生效。即最后被修改的值会覆盖前面设置的值。

消费重试次数与间隔（无序消息)

对于需要重试消费的消息，并不是Consumer在等待了指定时长后再次去拉取原来的消息进行消费，而是将这些需要重试消费的消息放入到了一个特殊Topic的队列中，而后进行再次消费的。这个特殊的队列就是重试队列

这个重试队列是针对消息者组的，而不是针对每个Topic设置的（一个Topic的消息可以让多个消费者组进行消费，所以会为这些消费者组各创建一个重试队列）

只有当出现需要进行重试消费的消息时，才会为该消费者组创建重试队列

当出现需要进行重试消费的消息时，Broker会为每个消费组都设置一个Topic名称为%RETRY%consumerGroup@consumerGroup 的重试队列

延迟登记展示

延迟等级在服务端的代码 MessageStoreConfig 类中，共计18个等级

时间间隔展示

时间间隔如图

注意，消费重试的时间间隔与延时消费的延时等级十分相似，除了没有延时等级的前两个时间外，其它的时间都是相同的

Broker对于重试消息的处理是通过延时消息实现的。先将消息保存到SCHEDULE_TOPIC_XXXX延迟队列中，延迟时间到后，会将消息投递到 %RETRY%consumerGroup@consumerGroup 重试队列中

重试队列(无序消息)

官方推荐

Consumer 返回 ConsumeConcurrentlyStatus.RECONSUME_LATER 状态码

Consumer 返回 Null

Consumer 抛出异常

集群消费方式下，消息消费失败后若希望消费重试，则需要在消息监听器接口的实现中明确进行如下三种方式之一的配置

消费重试配置方式

集群消费方式下，消息消费失败后若不希望消费重试，则在捕获到异常后同样也返回与消费成功后的相同的结果，即ConsumeConcurrentlyStatus.CONSUME_SUCCESS，则不进行消费重试

消费不重试配置方式

消息消费重试机制

当一条消息初次消费失败，消息队列会自动进行消费重试；达到最大重试次数后（默认是16次，间隔时间是4 小时 46 分钟），若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。这个队列就是死信队列（Dead-Letter Queue，DLQ），而其中的消息则称为死信消息（Dead-Letter Message，DLM）

死信队列是用于处理无法被正常消费的消息的

什么是死信队列

死信队列中的消息不会再被消费者正常消费，即DLQ对于消费者是不可见的

不管是什么消息，消息内容都在commitlog中

死信存储有效期与正常消息相同，均为 3 天（commitlog文件的过期时间），3 天后会被自动删除

死信队列就是一个特殊的Topic，名称为%DLQ%consumerGroup@consumerGroup ，即每个消费者组都有一个死信队列

如果⼀个消费者组未产生死信消息，则不会为其创建相应的死信队列

死信队列的特征

实际上，当⼀条消息进入死信队列，就意味着系统中某些地方出现了问题，从而导致消费者无法正常消费该消息，比如代码中原本就存在Bug。因此，对于死信消息，通常需要开发人员进行特殊处理。最关键的步骤是要排查可疑因素，解决代码中可能存在的Bug，然后再将原来的死信消息再次进行投递消费

死信消息的处理

死信队列

4.应用