架构决策之消息中间件MQ系列五-RocketMQ

- 日理万妓 2023-02-11 15:27 14阅读 0赞

# 一、前言 #

RocketMq是阿里贡献Apache消息中间件项目，采用java语言开发，经过阿里历年双十一流量洪峰的洗礼，并发性和可靠性经过了充分的验证，且支持的功能丰富，是活跃度较高的中间件之一，特别在国内市场。

# 二、架构 #

RocketMq由四部分组成，如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70][]

**NameServer**，名字服务，每个NameServer维护全量的broker，topic路由的相关信息。NameServer可集群化部署，本身是无状态的（元数据存在内存中，不落盘），相互之间独立。为客户端(生产者和消费者)提供路由发现，同时接受broker的心跳信息，维护broker在线状态。其作为类似kafka的zookeeper。

**Producer cluster**，生产者集群，包含一组生产者，负责生产消息，通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递。提供多种发送方式，同步发送、异步发送、单向发送。投递的过程支持快速失败并且低延迟。

**Broker cluster**，节点服务器集群，接受生产者发送的消息，将消息持久化(commitlog文件)，并发送给消费者消费。节点服务器也存储相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。broker节点间采用主从模式，实现高可用。

**Consumer cluster**，消费者集群，包含一组消费者，提供拉和推两种方式，从broker获取消息并提供给应用程序消费。同时也支持集群方式和广播方式的消费。

除了这几个部分外，还需要了解一下两个概念

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

**Message Queue**，消息队列，消息逻辑存储单位，一个Topic下的消息，顺序保存多个消息队列中(默认为创建4个读，4个写队列，这个类似于kafka的分区)

# 三、生产消息 #

## 1、创建消息 ##

消息由消息主题(Topic)，消息Flag，消息属性，消息体组成。

*  主题，该消息所属的topic。
 *  消息Flag，RocketMq不做处理。
 *  消息属性，包括tag，用于消息的过滤；key，消息的索引，多个用空格隔开，用于消息的快速索引或者追踪；waitStoreMsgOk，消息发送是否等消息存储完成后再返回；delayTimeLevel,消息延迟级别，用于定时消息或者消息重试。
 *  消息体，消息实际内容，字节数组形式存储。

对消息采用自定义的编解码格式，如下：

![20200608224643945.png][]

(1) 消息长度：总长度，四个字节存储，占用一个int类型；

(2) 序列化类型&消息头长度：同样占用一个int类型，第一个字节表示序列化类型，后面三个字节表示消息头长度；

(3) 消息头数据：经过序列化后的消息头数据；

(4) 消息主体数据：消息主体的二进制字节数据内容；

## 2、选择消息队列 ##

消息的逻辑存储单位是消息队列，一个Topic下有多个MessageQueue，并分布在不同的broker上。所以发送消息前，需要获取Topic路由信息。如图，TopicA主题下有8个消息队列(需要注意是，broker-a的queue0与broker-b的queue-0是两个队列，并不是备份)。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 1][]

首先生产者查看本地是否缓存了Topic的路由信息(主要是MessageQueue列表，每个MessageQueue对象包括brokerName，queueId等信息)；如果本地没有查到，则需要向NameServer查询，并本地组装该Topic的路由信息。

下一步就是要选择往哪个消息队列发送了，RocketMq有两种负载均衡测策略，**默认策略**和**集群超时容忍策略**。默认策略实际是一种轮询策略，通过自增随机数对MessageQueue列表大小取余，并获取MessageQueue的位置信息，但获得的MessageQueue所在的集群不能是上次的失败集群；集群超时容忍策略，先随机选择一个MessageQueue，如果因为超时等异常发送失败，会优先选择该broker集群下其他的MessageQueue进行发送，是一种高可用处理策略。

当然，也可以自定义发送规则，选择特定的MessageQueue发送，比如sharding key 进行区块分区，发送到对应的MessageQueue。

## 3、发送消息 ##

消息创建完成，并且选择了待发送的队列，万事俱备，只待发送。RocketMq发送模式有三种，同步发送，异步发送，单向发送，类似ActiveMQ。

**同步发送**，发送后，阻塞当前的线程，等待broker的响应。这种模式，可靠性高，但是会影响吞吐量。

**异步发送**，配置回调方法，发送后立即返回，broker调用回调方法响应发送结果，如同步模式相比，性能显著提高。

**单向发送**，仅将消息写入socket即可，不关注发送是否成功，也不会有重试机制。这种吞吐量最高，但是会有丢失消息的风险。

除了支持单条消息的发送，RocketMq还支持批量消息的发，多条消息的消息体合并一个。

## 4、消息重试 ##

在同步和异步模式下，如果消息发送异常，可以进行重试，重试的策略课设置如下：

*  retryTimesWhenSendFailed:同步发送失败重投次数，默认为2，因此生产者会最多尝试发送retryTimesWhenSendFailed + 1次。不会选择上次失败的broker，尝试向其他broker发送，最大程度保证消息不丢。超过重投次数，抛出异常，由客户端保证消息不丢。当出现RemotingException、MQClientException和部分MQBrokerException时会重投。
 *  retryTimesWhenSendAsyncFailed:异步发送失败重试次数，异步重试不会选择其他broker，仅在同一个broker上做重试，不保证消息不丢。
 *  retryAnotherBrokerWhenNotStoreOK:消息刷盘（主或备）超时或slave不可用（返回状态非SEND\_OK），是否尝试发送到其他broker，默认false。十分重要消息可以开启。

# 二、存储消息 #

# 1、刷盘与主从复制 #

RocketMQ采用主从架构，我们看下消息的数据流走向，如图。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 2][]

生产者生产消息后，发送到消息队列所在Master节点(仅在master节点写)的页面缓存中，一方面，master节点上数据继续落盘到磁盘上，实现持久化；另一方面，数据同步到Slave节点上，然后落盘到Slave磁盘上。

前面我们介绍在同步和异步发送的模式下，broker会返回写入响应结果。从上图看，返回的时机有以下几种可能

*  写入Master节点页面缓存
 *  Master节点落盘
 *  同步到Slave节点的页面缓存
 *  同步到Slave节点，并落盘。

其中第一种效率最高，但是可靠性较差，第四种效率是最低的，但是可靠性最强。

RocketMq不支持对单个消息进行策略设置，而是在Broker节点上进行全局设置。

*  flushDiskType，可选值为SYNC\_FLUSH,ASYNC\_FLUSH。SYNC\_FLUSH表示同步刷盘，即等待刷盘完成后，再通知返回；ASYNC\_FLUSH表示异步刷盘，写入页面缓存后即可返回，当缓存的消息积累到一定程度，统一刷盘。
 *  brokerRole，可选值为ASYNC\_MASTER,SYNC\_MASTER,SLAVE。其中ASYNC\_MASTER表示异步复制到从节点，master节点写入成功，即可返回；SYNC\_MASTER表示同步复制，Slaver写入成功后才返回；

鉴于性能和可靠性权衡考虑，一般**建议flushDiskType选择ASYNC\_FLUSH，brokerRole选择SYNC\_MASTER**。

## 2、存储模式 ##

消息最终要落盘到磁盘上，那么是怎么存储这些消息的呢？

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 3][]

如图所示，与消息相关的文件有三个，分别是Commitlog，ConsumerQueue(注意，不是MessageQueue)，indexFile。

*  **commitLog**

broker上所有topic的消息都会保存在commitLog文件中，该文件的路径为$HOME/store/$\{commitlog\}/$\{fileName\}，单个文件的大小为1G，fileName文件名长度为20，左边补零，剩余为起始偏移量，比如00000000001073741824，表示该文件保存偏移量从1073741824起始的消息。

还记得kafka是以每个partition作为一个文件，这个弊端就是打开文件的句柄会很多，从而限制了每台broker上partition的个数。

RocketMq从设计之初，就避免了这个"坑",让所有的消息顺序写入一个文件，同时也导致了一个问题，如果从这个大文件中查找某个消息，那将非常耗性能。为了解决这个问题，RocketMq引入了consumerQueue和indexFile。

*  **consumerQueue**

为了快速的查到并读出消息，在消息到达commitlog后，会异步转发到消息队列，进行消息检索。消息队列中并不保存消息的实际内容，而是消息的索引，每个消息包含20个字节，其中commitlog offset为8个字节，消息长度4个字节，消息的tag hashcode为8个字节。消息队列文件保存在HOME/store/consumequeue/\{topic\}/\{queueId\}/\{fileName\}，单个文件保存30W个条目。通过消息队列的索引，就能快速定位到消息在commitlog文件的位置。

consumerQueue并不持有消息的内容，仅保存消息的索引信息，注意与MessageQueue的区别，可以看做是MessageQueue的索引队列。

*  **indexFile**

indexFile保存了key与offset的关系，通过key快速检索出对应的消息。indexFile文件的保存位置$HOME \\store\\index$\{fileName\}，文件名fileName是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现HashMap结构，故RocketMq的索引文件其底层实现为hash索引。

消息最先存入commitLog，然后将索引写入consumerQueue。

## 3、消息删除机制 ##

消息不会永久保存在broker服务器上，对于commitLog和consumerQueue文件，一旦文件达到大小的阀值，就会创建新的文件，那么老的文件就不会再有更新，这些老文件都是可以被认为是过期文件，理论上都是可以被删除的。RocketMq可以通过配置过期时间(fileReservedTime，默认为72小时)，一旦达到这个过期时间，就是允许删除，但不是马上删除，而是在以下几个场景中。

*  指定删除文件的时间点，RocketMq通过deleteWhenRocketMQ 通过delete When 设置一天的固定时间执行一次删除过期文件操作， 默认为凌晨4 点。
 *   磁盘空间是否充足，如果磁盘空间不充足，则返回true ，表示应该触发过期文件删除操作。
 *  预留，手工触发，可以通过调用excuteDeleteFilesManualy 方法手工触发过期文件删除，

需要注意的是，删除消息文件时，不会判断这些消息有没有被消费。

# 三、消费消息 #

## 1、拉/推模式请求消息 ##

与RabbitMq类似，RocketMq的消费端也支持pull和push两种模式

*  **拉(pull)**

pull(拉)模式是消费端主动从broker上获取消息，获取请求的间隔时间，每次获取的条数，偏移量等都是有消费端决定，消费端可以根据自身的消费能力进行控制。

pull模式客户端需要实现消息队列的遍历，以及每个队列offset的保存，控制代码较为复杂，实际应用的较少。

*  **推(push)**

事实上，RocketMq并没有实现真正意义上的推模式，而是通过pull的"长轮询"实现。消费端不间断的发起请求，当broker有消息堆积时，则返回消费端，当没有新消息时，并不着急返回，broker会将这个请求"挂起"，通过线程不断扫描消息队列(请求的偏移量小于消息队列的偏移量，则表示有新消息)，直到有新消息，再返回给消费端。

无论pull还是push模式，获取到消息，都会先放入到消费端的缓存，再进行消费处理。

## 2、消息发送 ##

Broker是主从模式，主从节点都可读(这个和kafka以及RabbitMq不同)。请求到达broker后，根据consumerQueue索引消息的位置，然后再从commitLog文件中获取实际消息内容。

我们知道consumerQueue以及commitLog都是以文件的模式存储，并且commitLog的文件大小为1G，传统的I/O操作，需要经过两次拷贝操作，如图。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 4][]

对于大文件来说，两次拷贝的性能低下，影响读写的吞吐量。RocketMq采用内存映射机制，通过将应用程序的逻辑内存地址直接映射到Linux操作系统的内核缓冲区，减少一次拷贝。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 5][]

内存映射的基本原理，RocketMq通过MappedByteBuffer的map()函数将文件映射到虚拟内存，读取操作时，如果文件已经加载到页面缓存，则直接从内存中读取；如果没有，需要发生一次缺页中断重新拷贝到内存页中，再被读取。应用程序通过读写自己的逻辑内存，达到实际操作操作系统内核缓冲区的效果。

内存映射机制适合大文件的读写操作，整体性能提升2-3倍。

## 3、消息确认(ACK) ##

消息消费完成后，需要进行确认，以便broker知道消息队列的消费的进度，并进行管理。再均衡时，知道从哪个offset继续消费。

*  消费成功，消息处理完成后，返回ConsumeConcurrentlyStatus.CONSUME\_SUCCESS,表示消息成功。broker记录下offset。
 *  消费失败，如果由于数据库异常，余额不足等情况，返回ConsumeConcurrentlyStatus.RECONSUME\_LATER，表示消息失败，那么重新发回broker，构建RETRY topic，延迟一定时间(业务可配置，默认10s)，重新发送，当重试一定次数后(可配置，默认16次)，则投递到死信队列，人工干预。

消息是通过批量获取，且批量确认的，存在一定的重复消费风险。比如100条消息，其中99条已消费，还有一条没有消费，此时消费端宕机或者故障，没有来得及ack，broker无法知晓实际的消费情况。当重新负载后，这99条将会重新投递消费。这种情况，需要消费端做好幂等性保护(kafka,rabbitmq都有类似的情况)。

需要注意的是，消息确认对于push下的集群模式有效。

## 4、消费者负载均衡 ##

与kafka类似，消费者也是集群模式，集群中的消息者通讯模型分为两种，广播模式和集群模式。

*  **广播模式，**相同消费集群的每个Consumer实例都接收全量的消息(发布-订阅模式)。
 *  **集群模式**，相同消费集群的每个Consumer实例平均分摊消息(队列模式)。

集群模式下，一个消息消费队列在同一时间只允许被同一消费组内的一个消费者消费，一个消息消费者能同时消费多个消息队列(与kafka的设计理念一致)。

当消费集群中扩容或者缩容消费者实例，或者消费队列所在主从broker宕机，都需要进行重新负载，使得流量重新分配。我们看下再均衡的过程。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 6][]

(1)消费者通过定时任务，每个一段时间，向所有的broker发送心跳包，包含消息消费分组名称、订阅关系集合、消息通信模式和客户端id的值等信息。broker在接受到心跳包后，本地保存这些信息，为后面的再均衡提供数据准备。

(2)消费者启动再均衡线程，每隔20s执行一次。首先根据Topic向broker获取消费队列集合(mqSet)。

(3)根据消费者，Topic向broker获取消费者id列表(comsumerIdList，第一步broker已经保存了这些信息)。

(4)根据mqSet，consumerIdList，通过均衡算法，计算当前消费者分配到的消费队列集合。比如mqSet(q0,q1,q2,q3,q4,q5,q6,q7),consumerIdList(c0,c1,c2)。常用有以下两种算法。

*  平均分配,结果如下：

c0:q0,q1,q2

c1:q3,q4,q5

c2:q6,q7

*  平均轮询分配,结果如下；

c0:q0,q3,q6

c1:q1,q4,q7

c2:q2,q5,

(5)新分配的队列集合，与当前消费者原有的队列集合进行比较。

如果新队列集合中不包含原有的队列，则停止原有队列消息并移除。

如果原有队列集合中不包含新分配的队列集合，则创建pullquest，根据offset，开始从该消费队列消费消息。

如果新队列集合与原有集合队列重合，则保持消费。

# 四、NameServer #

我们先来考虑以下几个问题：

1、生产者和消费者如何知道所要操作的Topic有哪些队列，这些队列都分布在哪些broker上？

2、当broker出现故障，或者topic以及消费队列发生变化，生产者和消费者如何能快速的获悉。

3、broker之间如何了解彼此的信息，如topic，队列，主从等。

这些都是NameServer(协调者)需要完成的工作，它好比RocketMq的大脑，统一协调各部位的工作。

## 1、状态存储结构 ##

NameServer保存Broker集群状态主要有以下5个HashMap变量

*  topicQueueTable< String/ \* topic \*/， List<QueueData>>

key值是topic名称，value为List<QueueData>,QueueData 里存储着Broker 的名称、读写queue 的数量、同步标识等。

*  BrokerAddrTable< String/\* BrokerName\*/， BrokerData>

key值为BrokerName，相同名称的Broker可能存在多个broker节点，包括一个Master和多个slaver。brokerData包含brokerName，所属集群，主备Broker地址等信息。

*  ClusterAddrTable< String/\* ClusterName\*/， Set<String/\* BrokerName \*/>>

key值为集群名，value为集群下包含所有节点的brokerName。

*  BrokerLiveTable< String/ \* BrokerAddr \*/， BrokerLivelnfo >

key值为broker的地址，表示一台服务器，value为该节点服务器的实时状态，包括更新状态的时间。

*  filterServerTable<String/\* BrokerAddr \*/，List <String >/\* FilterServer\*/>

key值为broker的地址，filterServer是与这个broker关联的多个filterServer的地址。

所有的信息都可以由这5个变量组合而成。

## 2、状态维护 ##

broker与nameServer间建立长链，心跳保活，nameServer每隔10s检查一次，如果超过2分钟没有更新，则认为broker失效。从BrokerLivelnfo中剔除。注意：这样会导致生产者和消费者最长120s后发现broker不可用。

broker新增和删除，topic新增和删除，消费队列的变化，都会向NameServer进行注册，及时维护状态。

NameServer是无状态的，就意味着存储的变量不会落盘保存，只会保存在内存中。broker和所有的NameServer进行注册，每个NameServer保持全量的数据，NameServer之间相互独立，无需同步数据。生产者和消费者可以配置多个NameServer，当连接的NameServer故障后，转移另一台NameServer上。

# 五、可靠性 #

我们从消息发送，存储，消费三个阶段来分析下RocketMq的可靠性。

## 1、发送阶段 ##

前面我们介绍了消费发送时，会有三种模式，同步和异步模式下，都会返回发送的状态，如果发送失败，会进行消息重发，减少了在发送阶段的消息丢失。我们看下在发送阶段的其他的可靠性保障措施。

## (1)容错机制 ##

当某个broker节点宕机后，由于心跳检查的延后性(前面介绍，最长需要120s)，生产者还认为该broker可用，还会向该broker发送消息，发送的结果肯定失败。失败后会触发重发机制，按照前面讲的两种策略进行负载，无论哪一种都需要"过滤"掉失败的broker。

RocketMq设计了latencyFaultTolerance机制，sendLatencyFaultEnable开关开启，对之前失败的，按一定的时间做退避。例如，如果上次请求的latency超过550ms，就退避3000ms；超过1000ms，就退避60000ms。

### (2) 事务消息 ###

与RabbitMq类似，RocketMq也支持事务消息，采用的是两阶段提交方式。

1)生产者发送"待确认"消息，RocketMq接受到后，放入到特定的topic，该topic对消息者不可见。

2)Rocketmq回复发送成功，第一阶段结束。

3)生产者开始执行本地的逻辑。

4)生产者根据本地逻辑的执行结果，向RocketMq发送commit或者rollback消息。

5)RocketMq接受commit消息后，将消息放入原有的topic，订阅方将能够接受到消息；如果接受到rollback消息，则删除第一阶段的消息，订阅方无法接受消息。

6) 如果出现异常情况，没有接受到4)的消息，经过固定的时间后，对"待确认"消息进行回查(不一定是原来的生产者实例，同组的即可)，根据回查的结果按照5)处理。

## 2、存储阶段 ##

前面介绍，对于单台broker的flushDiskType有同步落盘和异步落盘两种配置，异步落盘的可靠性高。同时，RocketMq的broker是主从模式，主从之间的brokerRole可设置异步复制和同步复制，同步复制的可靠性高。我们重点看下主从和切换。

### (1)主从同步 ###

RocketMq的Master-Slaver模式是高可用的保障之一，支持读写分离，master可读可写，slaver只读。当slaver启动后，就会从master同步信息，注意是"信息"，不仅指的是消息体(commitlog)，还还包括一些元数据，如consumerOffset,SubscriptionGroupConfig等。对于这两种数据，采用不同的同步策略。

对于元数据，采用的是定时同步，因为元数据的实时性和可靠性相比较而言，没有太高要求。

对于消息体，创建TCP连接，不间断发送同步请求。消息内容一旦没有同步，master磁盘故障后，可能会导致消息的彻底丢失。所以对于消息体，实时性和可靠性要求高。其同步的流程图如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 7][]

### (2)主从切换 ###

对于Master-slave架构，由于Master负责写入，如果某台slave宕机后，不会影响消息的写入，但是会影响从该slave上读取消息，自动转移到其他的节点。

当Master节点宕机后，该节点上的消息队列将无法写入，但不会影响其他slave节点的消息读取，连接这台master节点的生产者通过容错机制，将选取其他的消息队列进行写入。

slave无法自动切换成master，需要手动干预。

RocketMq on Dledger版本可以基于raft算法进行leader节点选举，从而实现自动切换。

## 3、消费阶段 ##

前面介绍，通过消息确认机制，确保消息的不丢失。我们再来研究下消费阶段其他的可靠性场景。

### (1)顺序消息 ###

有些场景中，比如订单的生成，付款，发货3个消息，需要确保消息的顺序。顺序消息又分为全局顺序消息和部分顺序消息。无论哪种，都需要生产和消费配合。

*  全局顺序消费

默认的情况下，RocketMq的一个topic，会创建8个写队列，8个读队列，每个消息可能被写入任意一个队列里，而消费者有多个，每个消费者可能启动多个线程并行处理，也无法保证顺序。

如果要实现全局顺序消费，需要把Topic的读写队列数设置为1，然后Producer和Consumer的并发也设置为1，此时虽然能达到全局顺序消息的目的，但是会牺牲高吞吐，高并发的特性。

*  局部顺序消费

对于指定的Topic，根据业务的ID或者sharding key，分发到对应的MessageQueue。在消费的过程，通过加锁的方式，控制并发消费，确保顺序消费。

### (2)定时消息 ###

RocketMq支持定时任务的发送，考虑到性问题，RocketMq不支持任意精读的延迟时间，仅支持按特定级别的延迟消息,默认为”1s 5s 10s 30s lm 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，delayLevel=1表示延迟1s，delayLevel=2表示延迟5s，依次类推。

当delayLevel为1的消息到broker后，我们看下处理流程：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 8][]

1）改变topic为SCHEDULE\_TOPIC\_XXXX，备份原有的主题和消息队列到消息属性中，放入到消息队列为delay减1(本例为queue0)。改变topic是RocketMq常用的方式，前面介绍的事务提交也是如此。

2）每个延迟级别对应一个消息队列，并启动一个定时任务，按照延迟级别对应的延迟时间进行扫描，比如delay=1，为1s扫描一次。

3) 根据上次拉取的偏移量从消费队列中获取所有的信息。

4）根据消息的物理偏移量和消息大小，从commitlog拉取消息。

5）消息重新创建，恢复消息原有的主题，消息队列，清除delaylevel属性，存入commitlog文件。

6）转发到原有主题的消息 队列，供消费者消费。

由于定时任务按照延迟级别对应的延迟时间定期扫描，所以无法做到精确的延迟。

### (3)消息回溯 ###

消息被消费后，并不立即被删除，根据配置(fileReservedTime，默认为72小时)会保留一定的时间。在这段时间内，如果消费端下游系统(比如数据库），由于故障，导致消息丢失，可以通过时间维度进行消息回溯，重新消费。但此时可能会有重复，需要做好幂等性保护。

### (4)消息过滤 ###

RocketMq采用的是服务端过滤模式，避免不必要的消息传递到消费端。提供了三种方式进行消息过滤

*  Tag过滤

前面我们介绍过，在消息属性中包含了Tag，消息队列的后8个字节是Taghashcode。

![20200607094641477.png][]

根据Tag可以进行简单的过滤，不需要读取commitlog的消息体内容，性能最高。存储taghashcode而不是tag，是为了保持定长。

*  SQL表达式过滤

Tag虽然高效，但是支持的逻辑较简单，有时需要采用比较复杂的过滤逻辑，RocketMq提供了类似SQL表达式的方式进行过滤。这种方式需要读取commitlog的内容，会增加磁盘的读取压力，效率较低。

*  Filter Server过滤

Filter Server是比SQL更灵活的过滤方式，用户通过自定义java函数，根据java函数的逻辑对消息进行过滤。

Filter Server类似一个comsumer进程，从本机的Broker获取消息，在根据用户上传过来的java函数进行过滤，过滤后的消息在传给远端的comsumer。这种方式虽然灵活，但是会占用很多的CPU资源，要根据实际情况选择。

# 六、高吞吐 #

高吞吐和高可靠性在多数场景下是一对矛盾体，这就需要我们根据应用的场景进行不同的取舍选择。与可靠性一样，我们也按照发送，存储，消费三个阶段来进行分析。

## 1、发送阶段 ##

生产者发送消息有三种模式，其中单向模式，将消息发送socket后立即返回，不关心消息是否真正到达borker，这种模式的吞吐量是最高的，但是会导致消息的丢失，在某些场景下，比如日志收集是可以使用的。

另一种提高发送速度的方法就是增加Producer并发量，使用多个Producer实例进行发送，由于是顺序写入commitlog，所以能保持较高的写入性能。

## 2、存储阶段 ##

数据索引和数据实体分离，既能确保较高的写入性能，又能实现快速的读取。内存映射机制，减少一次拷贝，提升了写入和读取的性能。另外，I/O的调度算法推荐使用deadline。

## 3、消费阶段 ##

某些业务场景下，多条消息同时处理的时间会大大小于逐个处理的时间总和，比如消费消息中涉及update 某个数据库， 一次update IO 条的时间会大大小于十次updatel 条数据的时间。这时可以通过批量方式消费来提高消费的吞吐量。批量消费消息的个数可以通过设置Consumer 的consumeMessageBatchMaxSize 这个参数。

另一种方式通过增加consumer的并行量，不过在集群模式下，comsumer的个数不能超过Topic下read queue的总量。

# 七、总结 #

RocketMq由Producer，consumer，NameService，broker四部分组成。其中Producer为生产者，Consumer为消费者，NameService作为协调器，管理broker，topic相关信息；broker存储消息，并提供读写服务。

Producer从NameService获取topic的路由信息，封装消息后，通过负载均衡策略选择相应的消息队列，发送消息。发送的模式有三种，分别为同步，异步，单向。

broker采用主从模式，主节点读写，从节点只读。发送的消息到达主节点后，可选择异步或者同步刷盘方式，同时通过同步或者异步复制到从节点，确保主从一致。同一个broker节点上所有Topic信息写入到同一个commitlog，采用数据和索引分别保存，保证写入和读取的速度。

Consumer通过pull或者push的方式获取消息，消息处理后，进行消费ACK，确认消费成功；对于消费不成功的消息，可进行消息重试。

RocketMq支持事务消息，顺序消息，定时消息，消息回溯，消息重试，消息过滤等功能。

附录:

[架构决策之消息中间件MQ系列一-开篇][MQ_-]

[架构决策之消息中间件MQ系列二-ActiveMQ][MQ_-ActiveMQ]

[架构决策之消息中间件MQ系列三-RabbitMQ][MQ_-RabbitMQ]

[架构决策之消息中间件MQ系列四-Kafka][MQ_-Kafka]

[架构决策之消息中间件MQ系列五-RocketMQ][MQ_-RocketMQ]

[架构决策之消息中间件MQ系列六-Pulsar][MQ_-Pulsar]

[架构决策之消息中间件MQ系列七-总结][MQ_- 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200524121511228.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz,size_16,color_FFFFFF,t_70
[20200608224643945.png]: https://img-blog.csdnimg.cn/20200608224643945.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200524232508886.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200525224328891.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200527221525437.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200530225614788.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200530225623571.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200531104234298.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20200601204350892.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20200606234117927.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RjeTgz,size_16,color_FFFFFF,t_70
[20200607094641477.png]: https://img-blog.csdnimg.cn/20200607094641477.png
[MQ_-]: https://blog.csdn.net/tcy83/article/details/107899399
[MQ_-ActiveMQ]: https://blog.csdn.net/tcy83/article/details/107306096
[MQ_-RabbitMQ]: https://blog.csdn.net/tcy83/article/details/105977528
[MQ_-Kafka]: https://blog.csdn.net/tcy83/article/details/105619023
[MQ_-RocketMQ]: https://blog.csdn.net/tcy83/article/details/106311310
[MQ_-Pulsar]: https://blog.csdn.net/tcy83/article/details/106731392
[MQ_- 1]: https://blog.csdn.net/tcy83/article/details/108019907