RocketMq

1.消息中间件的使用场景

1.1 异步与解耦

1.2 流量消峰

1.3 数据分发

---

2.RocketMq中的角色

2.1 nameSpace

NameServer是整个RocketMQ的“大脑”，它是RocketMQ的服务注册中心,所以RocketMQ需要先启NameServer再启动Rocket中的Broker。
Broker在启动时向所有NameServer注册（主要是服务器地址等），生产者在发送消息之前先从NameServer获取Broker服务器地址列表（消费者一样），然后根据负载均衡算法从列表中选择一台服务器进行消息发送。

2.2 主机（broker）

RocketMQ的核心，用于暂存和传输消息。

2.3 消息（Message）

消息：生产或消费的数据，对于RocketMQ来说，消息就是字节数组。

2.4 生产者（producer）

生产者：也称为消息发布者，负责生产并发送消息至RocketMQ。

2.5 消费者（consumer）

消费者：也称为消息订阅者，负责从RocketMQ接收并消费消息

3.使用RocketMQ的核心概念

3.1 分组(Group)

生产者：标识发送同一类消息的Producer，通常发送逻辑一致。发送普通消息的时候，仅标识使用，并无特别用处。主要作用用于事务消息：
消费者：标识一类Consumer的集合名称，这类Consumer通常消费一类消息（也称为Consumer Group），且消费逻辑一致。同一个Consumer Group下的各个实例将共同消费topic的消息，起到负载均衡的作用。

3.2 主题（Topic）

标识RocketMQ中一类消息的逻辑名字，消息的逻辑管理单位。无论消息生产还是消费，都需要指定Topic。主题主要用于区分消息的种类：一个生产者可以发送消息给一个或者多个Topic，消息的消费者也可以订阅一个或者多个Topic消息。

3.3 标签(Tag)

RocketMQ支持给在发送的时候给消息打tag，同一个topic的消息虽然逻辑管理是一样的。但是消费同一个topic时，如果你消费订阅的时候指定的是tagA，那么tagB的消息将不会投递。

3.4 消息队列（Message Queue）

简称Queue或Q。消息物理管理单位。一个Topic将有若干个Q。若一个Topic创建在不同的Broker，则不同的broker上都有若干Q，消息将物理地存储落在不同Broker结点上，具有水平扩展的能力。
无论生产者还是消费者，实际的生产和消费都是针对Q级别。例如Producer发送消息的时候，会预先选择（默认轮询）好该Topic下面的某一条Q发送；Consumer消费的时候也会负载均衡地分配若干个Q，只拉取对应Q的消息。
每一条message queue均对应一个文件，这个文件存储了实际消息的索引信息。并且即使文件被删除，也能通过实际纯粹的消息文件（commit log）恢复回来。

3.5 偏移量(Offset)

RocketMQ中，有很多offset的概念。一般我们只关心暴露到客户端的offset。不指定的话，就是指Message Queue下面的offset。
Message queue是无限长的数组。一条消息进来下标就会涨1,而这个数组的下标就是offset，Message queue中的max offset表示消息的最大offset。
Consumer offset可以理解为标记Consumer Group在一条逻辑Message Queue上，消息消费到哪里即消费进度。但从源码上看，这个数值是消费过的最新消费的消息offset+1，即实际上表示的是下次拉取的offset位置。

4.普通消息

先使用RocketMQ提供的原生客户端的API，当然除了原生客户端外，SpringBoot、SpringCloudStream也进行了集成，但本质上这些也是基于原生API的封装，所以只需掌握原生API，其他的也会水到渠成。
Java代码中使用普通消息的整体流程如下

• 导入MQ客户端依赖

<dependency>
	<groupId>org.apache.rocketmq</groupId>
	<artifactId>rocketmq-client</artifactId>
	<version>4.8.0</version>
</dependency>
1
2
3
4
5

• 消息发送者步骤
  1.创建消息生产者producer，并指定生产者组名
  2.指定Nameserver地址
  3.启动producer
  4.创建消息对象，指定Topic、Tag和消息体
  5.发送消息
  6.关闭生产者producer

• 消息消费者步骤
  1.创建消费者Consumer，指定消费者组名
  2.指定Nameserver地址
  3.订阅主题Topic和Tag
  4.设置回调函数，处理消息
  5.启动消费者consumer

4.1 三种消息发送方式

4.1.1 发送同步消息

同步发送是指消息发送方发出数据后，同步等待，直到收到接收方发回响应之后才发下一个请求。这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。

代码演示:

发送结果分析:

• msgId
  消息的全局唯一标识（RocketMQ的ID生成是使用机器IP和消息偏移量的组成），由消息队列 MQ 系统自动生成，唯一标识某条消息。
• sendStatus
  发送的标识：成功，失败等
• queueId
  queueId是Topic的分区；Producer发送具体一条消息的时，对应选择的该Topic下的某一个Queue的标识ID。
• queueOffset
  Message queue是无限长的数组。一条消息进来下标就会涨1,而这个数组的下标就是queueOffset，queueOffset是从0开始递增。

4.1.2 发送异步消息

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。消息发送方在发送了一条消息后，不等接收方发回响应，接着进行第二条消息发送。发送方通过回调接口的方式接收服务器响应，并对响应结果进行处理。

代码演示

发送结果分析跟发送同步消息相同。

4.1.3 单向发送

这种方式主要用在不特别关心发送结果的场景，例如日志发送。单向（Oneway）发送特点为发送方只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答。此方式发送消息的过程耗时非常短，一般在微秒级别。

代码演示

4.1.4 三种消息发送的比较

4.2 两种消费类型（消费者从broker中获取消息的两种消息方式）

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

4.2.1 拉取式消费

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

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

4.2.2 推送式消费

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

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

4.2.3 两种消费类型比较

• pull：需要应用去实现对关联Queue的遍历，实时性差；但便于应用控制消息的拉取-
• push：封装了对关联Queue的遍历，实时性强，但会占用较多的系统资源

4.3 两种消息消费方式（消费模式）

4.3.1 负载均衡模式（集群消费）

消费者采用负载均衡方式消费消息，一个分组(Group)下的多个消费者共同消费队列消息，每个消费者处理的消息不同。一个Consumer Group中的各个Consumer实例分摊去消费消息，即一条消息只会投递到一个Consumer Group下面的一个实例。例如某个Topic有3个队列，其中一个Consumer Group 有 3 个实例，那么每个实例只消费其中的1个队列。集群消费模式是消费者默认的消费方式。

代码演示：

4.3.2 广播消费

广播消费模式中消息将对一个Consumer Group下的各个Consumer实例都投递一遍。即使这些 Consumer属于同一个Consumer Group，消息也会被Consumer Group 中的每个Consumer都消费一次。实际上，是一个消费组下的每个消费者实例都获取到了topic下面的每个Message Queue去拉取消费。所以消息会投递到每个消费者实例。

代码演示：

4.3.3 两种消费模式比对

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

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

• 广播消费模式下不支持顺序消息、不支持重置消费位点。

5 顺序消息

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

5.1 全局有序

• 全局有序
  全局有序比较简单，主要控制在于创建Topic指定只有一个队列，同步确保生产者与消费者都只有一个实例进行即可。

5.2 分区有序

• 分区有序
  在电商业务场景中，一个订单的流程是：创建、付款、推送、完成。在加入RocketMQ后，一个订单会分别产生对于这个订单的创建、付款、推送、完成等消息，如果我们把所有消息全部送入到RocketMQ中的一个主题中，这里该如何实现针对一个订单的消息顺序性呢！如下图：
  
  要完成分区有序性，在生产者环节使用自定义的消息队列选择策略，确保订单号尾数相同的消息会被先后发送到同一个队列中(案例中主题有3个队列，生产环境中可设定成10个满足全部尾数的需求)，然后再消费端开启负载均衡模式，最终确保一个消费者拿到的消息对于一个订单来说是有序的。
  代码案例
  生产者代码
  
  

发送日志

消费者代码
消费时，同一个OrderId获取到的肯定是同一个队列。从而确保一个订单中处理的顺序。

注意事项

使用顺序消息：首先要保证消息是有序进入MQ的，消息放入MQ之前，对id等关键字进行取模，放入指定messageQueue，同时consume消费消息失败时，不能返回reconsume——later，这样会导致乱序，所以应该返回suspend_current_queue_a_moment,意思是先等一会，一会儿再处理这批消息，而不是放到重试队列里。

6 延时消息

概念
延时消息：Producer 将消息发送到消息队列 RocketMQ 服务端，但并不期望这条消息立马投递（被消费者消费），而是延迟一定时间后才投递到 Consumer 进行消费，该消息即延时消息。

适用场景
消息生产和消费有时间窗口要求：比如在电商交易中超时未支付关闭订单的场景，在订单创建时向RocketMQ发送一条延时消息。这条消息将会在 30 分钟以后投递给消费者，消费者收到此消息后需要判断对应的订单是否已完成支付。 如支付未完成，则关闭订单。如已完成支付则忽略。

使用案例
Apache RocketMQ目前只支持固定精度的定时消息，因为如果要支持任意的时间精度，在 Broker 层面，必须要做消息排序，如果再涉及到持久化，那么消息排序要不可避免的产生巨大性能开销。（RocketMQ的商业版本Aliware MQ提供了任意时刻的定时消息功能，Apache的RocketMQ并没有,阿里并没有开源）

Apache RocketMQ发送延时消息是设置在每一个消息体上的，在创建消息时设定一个延时时间长度，消息将从当前发送时间点开始延迟固定时间之后才开始投递。
延迟消息的level，区分18个等级：level为1，表示延迟1秒后消费；level为2表示延迟5秒后消费；level为3表示延迟10秒后消费；以此类推；最大level为18表示延迟2个小时消费。具体标识如下：

是这生产消息跟普通的生产消息类似，只需要在消息上设置延迟队列的level即可。消费消息跟普通的消费消息一致。
生产者

消费者

查看消费者消息信息，打印消费延迟与生产时设定符合。

7. 批量消息

在高并发场景中，批量发送消息能显著提高传递消息发送时的性能（减少网络连接及IO的开销）。使用批量消息时的限制是这些批量消息应该有相同的topic，相同的waitStoreMsgOK（集群时会细讲），且不能是延时消息。
在发送批量消息时先构建一个消息对象集合，然后调用send(Collection msg)系列的方法即可。由于批量消息的4MB限制，所以一般情况下在集合中添加消息需要先计算当前集合中消息对象的大小是否超过限制，如果超过限制也可以使用分割消息的方式进行多次批量发送。

使用案例
一般批量发送(不考虑消息分割)
因为批量消息是一个Collection，所以送入消息可以是List，也可以使Set，这里为方便起见，使用List进行批量组装发送。

批量切分发送
如果消息的总长度可能大于4MB时，这时候最好把消息进行分割，案例中以1M大小进行消息分割。
我们需要发送10万元素的数组，这个量很大，怎么快速发送完。使用批量发送，同时每一批控制在1M左右确保不超过消息大小限制。

8. 消息的过滤

在实际的开发应用中，对于一类消息尽可能使用一个Topic进行存储，但在消费时需要选择您想要的消息，这时可以使用RocketMQ的消息过滤功能，具体实现是利用消息的Tag和Key。
Key一般用于消息在业务层面的唯一标识。对发送的消息设置好 Key，以后可以根据这个 Key 来查找消息。比如消息异常，消息丢失，进行查找会很方便。RocketMQ 会创建专门的索引文件，用来存储 Key与消息的映射，由于底层实现是 Hash 索引，应尽量使 Key唯一，避免潜在的哈希冲突。
Tag可以理解为是二级分类。以淘宝交易平台为例，订单消息和支付消息属于不同业务类型的消息，分别创建OrderTopic 和PayTopic，其中订单消息根据不同的商品品类以不同的 Tag 再进行细分，如手机类、家电类、男装类、女装类、化妆品类，最后它们都被各个不同的系统所接收。通过合理的使用 Topic 和 Tag，可以让业务结构清晰，更可以提高效率。
Key和Tag的主要差别是使用场景不同，Key主要用于通过命令行命令查询消息，而Tag用于在消息端的代码中，用来进行服务端消息过滤。
使用Key一般使用mqadmin管理工具，具体位置在RocketMQ/bin目录下。具体文档见：https://github.com/apache/rocketmq/blob/master/docs/cn/operation.md

8.1 Tag过滤

使用Tag过滤的方式是在消息生产时传入感兴趣的Tag标签，然后在消费端就可以根据Tag来选择您想要的消息。具体的操作是在创建Message的时候添加，一个Message只能有一个Tag。

8.2 Sql过滤

SQL特性可以通过发送消息时的属性来进行消息的过滤计算。具体的操作是使用SQL92标准的sql语句，前提是只有使用push模式的消费者才能用（消费的模式就是push）
SQL基本语法
数值比较：比如：>，>=，<，<=，BETWEEN，=；
字符比较：比如：=，<>，IN；
IS NULL 或者 IS NOT NULL；
逻辑符号：AND，OR，NOT；
常量支持类型为：
数值，比如：123，3.1415；
字符，比如：‘abc’，必须用单引号包裹起来；
NULL，特殊的常量
布尔值，TRUE 或 FALSE
注意事项
Sql过滤需要Broker开启这项功能（如果消费时使用SQL过滤抛出异常错误，说明Sql92功能没有开启），需要修改Broker.conf配置文件。加入enablePropertyFilter=true 然后重启Broker服务。
使用案例
消息生产者，发送消息时加入消息属性，你能通过putUserProperty来设置消息的属性，以下案例中生产者发送10条消息，除了设置Tag之外，另外设置属性a的值。

用MessageSelector.bySql来使用sql筛选消息

消费结果：按照Tag和SQL过滤消费3条消息。

9. 分布式事务消息

分布式事务的简图

业务场景：用户A转账100元给用户B，这个业务比较简单，具体的步骤：
1、用户A的账户先扣除100元
2、再把用户B的账户加100元
如果在同一个数据库中进行，事务可以保证这两步操作，要么同时成功，要么同时不成功。这样就保证了转账的数据一致性。
但是在微服务架构中因为各个服务都是独立的模块，都是远程调用，没法在同一个事务中，所以就会遇到分布式事务问题。
RocketMQ中的处理方案

RocketMQ分布式事务方式：把扣款业务和加钱业务异步化，扣款成功后，发送“扣款成功消息”到RocketMQ，加钱业务订阅“扣款成功消息”，再对用户B加钱（扣款消息中包含了源账户和目标账户ID，以及金额）
对于扣款业务来说，需规定是先扣款还是先向MQ发消息
场景一：先扣款后向MQ发消息
先扣款再发送消息，万一发送消息失败了，那用户B就没法加钱

场景二：先向MQ发像消息，后扣款
扣款成功消息发送成功，但用户A扣款失败，可加钱业务订阅到了消息，用户B加了钱
问题所在，也就是没法保证扣款和发送消息，同时成功，或同时失败；导致数据不一致。
为了解决以上问题，RocketMq把消息分为两个阶段：半事务阶段和确认阶段

• 半事务阶段:
  该阶段主要发一个消息到rocketmq，但该消息只储存在commitlog中，但consumeQueue中不可见，也就是消费端（订阅端）无法看到此消息
• 确认阶段（commit/rollback）:
  该阶段主要是把半事务消息保存到consumeQueue中，即让消费端可以看到此消息，也就是可以消费此消息。如果是rollback就不保存。

整个流程：
1、A在扣款之前，先发送半事务消息
2、发送预备消息成功后，执行本地扣款事务
3、扣款成功后，再发送确认消息
4、B消息端（加钱业务）可以看到确认消息，消费此消息，进行加钱
注意：上面的确认消息可以为commit消息，可以被订阅者消费；也可以是Rollback消息，即执行本地扣款事务失败后，提交rollback消息，即删除那个半事务消息，订阅者无法消费。这样就可以解决以下异常问题：
异常1：如果发送半事务消息失败，下面的流程不会走下去，这个是正常的。
异常2：如果发送半事务消息成功，但执行本地事务失败。这个也没有问题，因为此半事务消息不会被消费端订阅到，消费端不会执行业务。
异常3：如果发送半事务消息成功，执行本地事务成功，但发送确认消息失败；这个就有问题了，因为用户A扣款成功了，但加钱业务没有订阅到确认消息，无法加钱。这里出现了数据不一致。

10. Request-Reply消息

什么是Request-Reply？
RocketMQ 中"Request-Reply"模式允许Producer发出消息后，以同步或异步的形式等Consumer消费并返回一个响应消息，达到类似RPC的调用过程。
RocketMQ从4.6.0版本开始支持这种模式。这种模式的流程如下图：

与RPC的不同
RocketMQ的这种调用方式跟dubbo之类的RPC调用非常类似，那为什么不使用dubbo？而要使用RocketMQ的这种RPC调用呢？原因如下：

• 基于RocketMQ来实现RPC可以快速搭建服务的消息总线，实现自己的RPC框架。
• 基于RocketMQ来实现RPC可以方便的收集调用的相关信息，能够实现调用链路追踪和分析。
• 基于RocketMQ来实现RPC既可以解耦两个系统之间的依赖，也可以实现跨网络区域实现系统间的同步调用，这里RocketMQ扮演的是一个类似于网关的角色。
  Request-Reply的实现逻辑
  

在以上图中，可以看到，使用RPC模式还是三方：Producer、Broker、Consumer。
在Producer中进行消息的发送时，可以随便指定Topic，但是需要送入reply_to_client、correlation_id两个关键信息，reply_to_client记录着请求方的clientlD（用于Broker响应时确定client端）。而correlation_id是标识每次请求的，用于响应消息与请求的配对。而在进行发送消息时，也有两种模式，一种同步阻塞，另外一种异步非阻塞，这些跟之前普通消息的三种发送方式类似。
Broker端除了Producer发送时指定的Topic之外，还有一个Reply_Topic,这个以集群名_REPLY_TOPIC命名（不管RPC生产者主题有多少，这个在一个集群中只有一个），主要用于Consumer响应RPC消息的路由发现。
Consumer端除了消费监听之外，还需要加入一个消息的生产（用于RPC的响应消息），必须使用客户端提供的MessageUtil进行消息的包装，防止关键信息丢失从而导致Producer不能收到RPC消息响应。
代码案例
生产者向RequestTopic主题发送RPC消息，使用同步阻塞方式。发送方法也不是send方法，而是request方法（该方法会封装reply_to_client、correlation_id等关键信息），同时方法也提供了Message的返回值。

消费者接受主题消息发送RPC响应。收到响应后需要再做一次生产，使用工具类MessageUtil封装消息后进行响应消息发送。

RPC案例消息：
生产者打印如下，对比普通消息多了reply_to_client、correlation_id两个关键信息，reply_to_client记录着请求方的clientlD（用于Broker响应时确定client端）。而correlation_id是标识每次请求的，用于响应消息与请求的配对。

消费打如下，消费者同时需要响应RPC，对应的主题是DefaultCluster_REPLY_TOPIC。

在PRC方式中，生产者也可以使用异步方式发起，代码如下：

11. RocketMQ的存储设计

11.1 Domain Model

领域模型（Domain Model）是对领域内的概念类或现实世界中对象的可视化表示。又称概念模型、领域对象模型、分析对象模型。它专注于分析问题领域本身，发掘重要的业务领域概念，并建立业务领域概念之间的关系。

---

---

11.1.1 Message

Message是RocketMQ消息引擎中的主体。messageId是全局唯一的。MessageKey是业务系统（生产者）生成的，所以如果要结合业务，可以使用MessageKey作为业务系统的唯一索引。

另外Message中的equals方法和hashCode主要是为了完成消息只处理一次（Exactly-Once）。
Exactly-Once是指发送到消息系统的消息只能被消费端处理且仅处理一次，即使生产端重试消息发送导致某消息重复投递，该消息在消费端也只被消费一次。

11.1.2 Topic

Tags是在同一Topic中对消息进行分类
subTopics==Message Queue,其实在内存逻辑中，subTopics是对Topics的一个拓展，尤其是在MQTT这种协议下，在Topic底下会有很多subTopics。

11.1.3 Queue

Queue是消息物理管理单位，比如在RocketMQ的控制台中，就可以看到每一个queue中的情况（比如消息的堆积情况、消息的TPS、QPS）

11.1.4 Offset

对于每一个Queue来说都有Offset,这个是消费位点。

11.1.5 Group

业务场景中，如果有一堆发送者，一堆消费者，所以这里使用Group的概念进行管理。

11.1.6 对应关系

Message与 Topic是多对一的关系，一个Topic可以有多个Message.
Topic到Queue是一对多的关系，这个也是方便横向拓展，也就是消费的时候，这里可以有很多很多的Queue.
一个Queue只有一个消费位点(Offset)，所以Topic和Offset也是一对多的关系
Topic和Group也是多对多的关系。

11.1.7 消费并发度

从上面模型可以看出，要解决消费并发，就是要利用Queue,一个Topic可以分出更多的queue,每一个queue可以存放在不同的硬件上来提高并发。

11.1.8 热点问题(顺序、重复)

前面讲过要确保消息的顺序，生产者、队列、消费者最好都是一对一的关系。但是这样设计，并发度就会成为消息系统的瓶颈（并发度不够），RocketMQ不解决这个矛盾的问题。理由如下：
1、乱序的应用实际大量存在
2、队列无序并不意味着消息无序
另外还有消息重复，造成消息重复的根本原因是：网络不可达（网络波动）。所以如果消费者收到两条一样的消息，应该是怎么处理？
RocketMQ不保证消息不重复，如果你的业务要严格确保消息不重复，需要在自己的业务端进行去重。
1、消费端处理消息的业务逻辑保持幂等性
2、确保每一条消息都有唯一的编号且保证消息处理成功与去重表的日志同时出现

11.2 消息存储结构

RocketMQ因为有高可靠性的要求(宕机不丢失数据)，所以数据要进行持久化存储。所以RocketMQ 采用文件进行存储。

11.2.1 存储文件

- commitLog：消息存储目录
- config：运行期间一些配置信息
- consumerqueue：消息消费队列存储目录
- index：消息索引文件存储目录
- abort：如果存在改文件则Broker非正常关闭
- checkpoint：文件检查点，存储CommitLog文件最后一次刷盘时间戳、consumerqueue最后一次刷盘时间，index索引文件最后一次刷盘时间戳。
1
2
3
4
5
6

11.2.2 消息存储结构

RocketMQ消息的存储是由ConsumeQueue和CommitLog配合完成 的，消息真正的物理存储文件是CommitLog，ConsumeQueue是消息的逻辑队列，类似数据库的索引文件，存储的是指向物理存储的地址。每 个Topic下的每个Message Queue都有一个对应的ConsumeQueue文件。

• CommitLog：存储消息的元数据
• ConsumerQueue：存储消息在CommitLog的索引
• IndexFile：为了消息查询提供了一种通过key或时间区间来查询消息的方法，这种通过IndexFile来查找消息的方法不影响发送与消费消息的主流程

---

---

11.2.3 CommitLog

CommitLog 以物理文件的方式存放，每台 Broker 上的 CommitLog 被本机器所有 ConsumeQueue 共享，文件地址：$ {user.home} \store$ { commitlog} \ $ { fileName}。在CommitLog 中，一个消息的存储长度是不固定的， RocketMQ采取一些机制，尽量向CommitLog 中顺序写 ，但是随机读。commitlog 文件默认大小为lG ，可通过在 broker 置文件中设置 mappedFileSizeCommitLog属性来改变默认大小。

---

---

Commitlog文件存储的逻辑视图如下，每条消息的前面4个字节存储该条消息的总长度。但是一个消息的存储长度是不固定的。

每个 CommitLog 文件的大小为 1G，一般情况下第一个 CommitLog 的起始偏移量为 0，第二个 CommitLog 的起始偏移量为 1073741824 （1G = 1073741824byte）。

每台Rocket只会往一个commitlog文件中写，写完一个接着写下一个。
indexFile 和 ComsumerQueue 中都有消息对应的物理偏移量，通过物理偏移量就可以计算出该消息位于哪个 CommitLog 文件上。

11.2.4 ConsumeQueue

ConsumeQueue 是消息的逻辑队列，类似数据库的索引文件，存储的是指向物理存储的地址。每个Topic下的每个 Message Queue 都有一个对应的 ConsumeQueue 文件， 文件地址在$ {$storeRoot} \consumequeue$ {topicName} $ { queueld} $ {fileName}。

ConsumeQueue中存储的是消息条目，为了加速 ConsumeQueue 消息条目的检索速度与节省磁盘空间，每一个 Consumequeue条目不会存储消息的全量信息，消息条目如下：

ConsumeQueue 即为Commitlog 文件的索引文件， 其构建机制是 当消息到达 Commitlog 文件后 由专门的线程产生消息转发任务，从而构建消息消费队列文件（ConsumeQueue ）与下文提到的索引文件。
存储机制这样设计有以下几个好处：

• （1)、CommitLog 顺序写 ，可以大大提高写入效率。
  （实际上，磁盘有时候会比你想象的快很多，有时候也比你想象的慢很多，关键在如何使用，使用得当，磁盘的速度完全可以匹配上网络的数据传输速度。目前的高性能磁盘，顺序写速度可以达到600MB/s ，超过了一般网卡的传输速度，这是磁盘比想象的快的地方 但是磁盘随机写的速度只有大概lOOKB/s,和顺序写的性能相差 6000 倍！）
• （2)、虽然是随机读，但是利用操作系统的 pagecache 机制，可以批量地从磁盘读取，作为 cache 存到内存中，加速后续的读取速度。
• （3)、为了保证完全的顺序写，需要 ConsumeQueue 这个中间结构 ，因为ConsumeQueue 里只存偏移量信息，所以尺寸是有限的，在实际情况中，大部分的 ConsumeQueue 能够被全部读入内存，所以这个中间结构的操作速度很快，可以认为是内存读取的速度。此外为了保证 CommitLog和ConsumeQueue 的一致性， CommitLog 里存储了 Consume Queues 、Message Key、 Tag 等所有信息，即使 ConsumeQueue 丢失，也可以通过 commitLog 完全恢复出来。

11.2.5 IndexFile

RocketMQ还支持通过MessageID或者MessageKey来查询消息；使用ID查询时，因为ID就是用broker+offset生成的（这里msgId指的是服务端的），所以很容易就找到对应的commitLog文件来读取消息。但是对于用MessageKey来查询消息，RocketMQ则通过构建一个index来提高读取速度。
index 存的是索引文件，这个文件用来加快消息查询的速度。消息消费队列 RocketMQ 专门为消息订阅构建的索引文件 ，提高根据主题与消息检索消息的速度 ，使用Hash索引机制，具体是Hash槽与Hash冲突的链表结构。（这里不做过多解释）

11.2.6 Config

config 文件夹中 存储着Topic和Consumer等相关信息。主题和消费者群组相关的信息就存在在此。
topics.json : topic 配置属性
subscriptionGroup.json :消息消费组配置信息。
delayOffset.json ：延时消息队列拉取进度。
consumerOffset.json ：集群消费模式消息消进度。
consumerFilter.json ：主题消息过滤信息。

11.2.7 其他

abort ：如果存在 abort 文件说明 Broker 非正常闭，该文件默认启动时创建，正常退出之前删除
checkpoint ：文件检测点，存储 commitlog 文件最后一次刷盘时间戳、 consumequeue最后一次刷盘时间、 index 索引文件最后一次刷盘时间戳。

11.3 过期文件删除

由于 RocketMQ 操作 CommitLog，ConsumeQueue文件是基于内存映射机制并在启动的时候会加载 commitlog，ConsumeQueue 目录下的所有文件，为了避免内存与磁盘的浪费，不可能将消息永久存储在消息服务器上，所以需要引入一种机制来删除己过期的文件。
删除过程分别执行清理消息存储文件（ Commitlog ）与消息消费 队列文件（ ConsumeQueue 文件）， 消息消费队列文件与消息存储文件（ Commitlog ）共用一套过期文件机制。
RocketMQ 清除过期文件的方法是 ：如果非当前写文件在一定时间间隔内没有再次被更新，则认为是过期文件，可以被删除， RocketMQ 不会关注这个文件上的消息是否全部被消费。默认每个文件的过期时间为 42小时（不同版本的默认值不同，这里以4.4.0为例） ，通过在 Broker 配置文件中设置 fileReservedTime 来改变过期时间，单位为小时。
触发文件清除操作的是一个定时任务，而且只有定时任务，文件过期删除定时任务的周期由该删除决定，默认每10s执行一次。

11.3.1 过期判断

文件删除主要是由这个配置属性：fileReservedTime：文件保留时间。也就是从最后一次更新时间到现在，如果超过了该时间，则认为是过期文件， 可以删除。
另外还有其他两个配置参数：
deletePhysicFilesInterval：删除物理文件的时间间隔（默认是100MS），在一次定时任务触发时，可能会有多个物理文件超过过期时间可被删除，因此删除一个文件后需要间隔deletePhysicFilesInterval这个时间再删除另外一个文件，由于删除文件是一个非常耗费IO的操作，会引起消息插入消费的延迟（相比于正常情况下），所以不建议直接删除所有过期文件。
destroyMapedFileIntervalForcibly：在删除文件时，如果该文件还被线程引用，此时会阻止此次删除操作，同时将该文件标记不可用并且纪录当前时间戳destroyMapedFileIntervalForcibly这个表示文件在第一次删除拒绝后，文件保存的最大时间，在此时间内一直会被拒绝删除，当超过这个时间时，会将引用每次减少1000，直到引用 小于等于 0为止，即可删除该文件.

11.3.2 删除条件

1）指定删除文件的时间点， RocketMQ 通过 deleteWhen 设置一天的固定时间执行一次。删除过期文件操作， 默认为凌晨4点。
2）磁盘空间是否充足，如果磁盘空间不充足(DiskSpaceCleanForciblyRatio。磁盘空间强制删除文件水位。默认是85)，会触发过期文件删除操作。
另外还有RocketMQ的磁盘配置参数：
1:物理使用率大于diskSpaceWarningLevelRatio（默认90%可通过参数设置）,则会阻止新消息的插入。
2:物理磁盘使用率小于diskMaxUsedSpaceRatio(默认75%) 表示磁盘使用正常。

11.4 零拷贝与MMAP

11.4.1 什么是零拷贝?

零拷贝(英语: Zero-copy) 技术是指计算机执行操作时，CPU不需要先将数据从某处内存复制到另一个特定区域。这种技术通常用于通过网络传输文件时节省CPU周期和内存带宽。
➢零拷贝技术可以减少数据拷贝和共享总线操作的次数，消除传输数据在存储器之间不必要的中间拷贝次数，从而有效地提高数据传输效率
➢零拷贝技术减少了用户进程地址空间和内核地址空间之间因为上:下文切换而带来的开销
可以看出没有说不需要拷贝，只是说减少冗余[不必要]的拷贝。
下面这些组件、框架中均使用了零拷贝技术：Kafka、Netty、Rocketmq、Nginx、Apache。

11.4.2 传统数据传送机制

比如：读取文件，再用socket发送出去，实际经过四次copy。
伪码实现如下：
buffer = File.read()
Socket.send(buffer)
1、第一次：将磁盘文件，读取到操作系统内核缓冲区；
2、第二次：将内核缓冲区的数据，copy到应用程序的buffer；
3、第三步：将application应用程序buffer中的数据，copy到socket网络发送缓冲区(属于操作系统内核的缓冲区)；
4、第四次：将socket buffer的数据，copy到网卡，由网卡进行网络传输。

分析上述的过程，虽然引入DMA来接管CPU的中断请求，但四次copy是存在“不必要的拷贝”的。实际上并不需要第二个和第三个数据副本。应用程序除了缓存数据并将其传输回套接字缓冲区之外什么都不做。相反，数据可以直接从读缓冲区传输到套接字缓冲区。
显然，第二次和第三次数据copy 其实在这种场景下没有什么帮助反而带来开销(DMA拷贝速度一般比CPU拷贝速度快一个数量级)，这也正是零拷贝出现的背景和意义。
打个比喻：200M的数据，读取文件，再用socket发送出去，实际经过四次copy（2次cpu拷贝每次100ms ，2次DMS拷贝每次10ms）
传统网络传输的话：合计耗时将有220ms
同时，read和send都属于系统调用，每次调用都牵涉到两次上下文切换：

总结下，传统的数据传送所消耗的成本：4次拷贝，4次上下文切换。
4次拷贝，其中两次是DMA copy，两次是CPU copy。

11.4.2 mmap内存映射

硬盘上文件的位置和应用程序缓冲区(application buffers)进行映射（建立一种一一对应关系），由于mmap()将文件直接映射到用户空间，所以实际文件读取时根据这个映射关系，直接将文件从硬盘拷贝到用户空间，只进行了一次数据拷贝，不再有文件内容从硬盘拷贝到内核空间的一个缓冲区。
mmap内存映射将会经历：3次拷贝: 1次cpu copy，2次DMA copy；
打个比喻：200M的数据，读取文件，再用socket发送出去，如果是使用MMAP实际经过三次copy（1次cpu拷贝每次100ms ，2次DMS拷贝每次10ms）合计只需要120ms
从数据拷贝的角度上来看，就比传统的网络传输，性能提升了近一倍。
以及4次上下文切换

mmap()是在 <sys/mman.h> 中定义的一个函数，此函数的作用是创建一个新的 虚拟内存 区域，并将指定的对象映射到此区域。 mmap 其实就是通过 内存映射 的机制来进行文件操作。
Windows操作系统上也有虚拟机内存，如下图：

11.4.2 代码

RocketMQ中MMAP运用
如果按照传统的方式进行数据传送，那肯定性能上不去，作为MQ也是这样，尤其是RocketMQ，要满足一个高并发的消息中间件，一定要进行优化。所以RocketMQ使用的是MMAP。
RocketMQ一个映射文件大概是，commitlog 文件默认大小为lG。
这里需要注意的是，采用MappedByteBuffer这种内存映射的方式有几个限制，其中之一是一次只能映射1.5~2G 的文件至用户态的虚拟内存，这也是为何RocketMQ默认设置单个CommitLog日志数据文件为1G的原因了。
MMAP文件对应

11.5 RocketMQ源码中的MMAP运用

RocketMQ源码中，使用MappedFile这个类类进行MMAP的映射

如果按照传统的方式进行数据传送，那肯定性能上不去，作为MQ也是这样，尤其是RocketMQ，要满足一个高并发的消息中间件，一定要进行优化。所以RocketMQ使用的是MMAP。
RocketMQ一个映射文件大概是，commitlog 文件默认大小为lG。
这里需要注意的是，采用MappedByteBuffer这种内存映射的方式有几个限制，其中之一是一次只能映射1.5~2G 的文件至用户态的虚拟内存，这也是为何RocketMQ默认设置单个CommitLog日志数据文件为1G的原因了。

11.5.1 MMAP文件对应

---

---

11.5.2 RocketMQ源码中的MMAP运用

RocketMQ源码中，使用MappedFile这个类类进行MMAP的映射

11.6 RocketMQ存储整体设计总结

11.6.1 消息生产与消息消费相互分离

Producer端发送消息最终写入的是CommitLog（消息存储的日志数据文件），Consumer端先从ConsumeQueue（消息逻辑队列）读取持久化消息的起始物理位置偏移量offset、大小size和消息Tag的HashCode值，随后再从CommitLog中进行读取待拉取消费消息的真正实体内容部分；

11.6.2 RocketMQ的CommitLog文件采用混合型存储

所有的Topic下的消息队列共用同一个CommitLog的日志数据文件，并通过建立类似索引文件—ConsumeQueue的方式来区分不同Topic下面的不同MessageQueue的消息，同时为消费消息起到一定的缓冲作用（异步服务线生成了ConsumeQueue队列的信息后，Consumer端才能进行消费）。这样，只要消息写入并刷盘至CommitLog文件后，消息就不会丢失，即使ConsumeQueue中的数据丢失，也可以通过CommitLog来恢复。

11.6.3 RocketMQ每次读写文件的时候真的是完全顺序读写吗？

发送消息时，生产者端的消息确实是顺序写入CommitLog；订阅消息时，消费者端也是顺序读取ConsumeQueue，然而根据其中的起始物理位置偏移量offset读取消息真实内容却是随机读取CommitLog。 所以在RocketMQ集群整体的吞吐量、并发量非常高的情况下，随机读取文件带来的性能开销影响还是比较大的，RocketMQ怎么优化的，源码解读部分进行讲解。

11.7 消息发送时的重要方法/属性

11.7.1 属性

org.apache.rocketmq.example.details. ProducerDetails类中

• producerGroup：生产者所属组
• defaultTopicQueueNums：默认主题在每一个Broker队列数量
• sendMsgTimeout：发送消息默认超时时间，默认3s
• compressMsgBodyOverHowmuch：消息体超过该值则启用压缩，默认4k
• retryTimesWhenSendFailed：同步方式发送消息重试次数，默认为2，总共执行3次
• retryTimesWhenSendAsyncFailed：异步方法发送消息重试次数，默认为2
• retryAnotherBrokerWhenNotStoreOK：消息重试时选择另外一个Broker时，是否不等待存储结果就返回，默认为false
• maxMessageSize：允许发送的最大消息长度，默认为4M

11.7.2 方法

org.apache.rocketmq.example.details. ProducerDetails类中
//启动
void start() throws MQClientException;
//关闭
void shutdown();
//查找该主题下所有消息队列
List fetchPublishMessageQueues(final String topic) throws MQClientException;

11.7.2.1 单向发送

//发送单向消息
void sendOneway(final Message msg) throws MQClientException, RemotingException,
    InterruptedException;
    
//选择指定队列单向发送消息
void sendOneway(final Message msg, final MessageQueue mq) throws MQClientException,
    RemotingException, InterruptedException;
1
2
3
4
5
6
7

11.7.2.2 同步发送

//同步发送消息
SendResult send(final Message msg) throws MQClientException, RemotingException, MQBrokerException,InterruptedException;

//同步超时发送消息
SendResult send(final Message msg, final long timeout) throws MQClientException,
        RemotingException, MQBrokerException, InterruptedException;
 
//选择指定队列同步发送消息
SendResult send(final Message msg, final MessageQueue mq) throws MQClientException,
    RemotingException, MQBrokerException, InterruptedException;

1
2
3
4
5
6
7
8
9
10
11

11.7.2.3 异步发送

//异步发送消息
void send(final Message msg, final SendCallback sendCallback) throws MQClientException,
        RemotingException, InterruptedException;
        
//异步超时发送消息
void send(final Message msg, final SendCallback sendCallback, final long timeout)
    throws MQClientException, RemotingException, InterruptedException;
    
//选择指定队列异步发送消息
void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback)
    throws MQClientException, RemotingException, InterruptedException;

1
2
3
4
5
6
7
8
9
10
11
12

11.8 消息消费时的重要方法/属性

org.apache.rocketmq.example.details. ComuserDetails类中

11.8.1 属性

//消费者组
private String consumerGroup;

//消息消费模式
private MessageModel messageModel = MessageModel.CLUSTERING;

//指定消费开始偏移量（最大偏移量、最小偏移量、启动时间戳）开始消费
private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
//ConsumeFromTimestamp模式下只会在订阅组（消费者群组）第一次启动的时候，过滤掉小于当前系统时间戳的消息，后续如果进程停掉或者崩溃，但是又生产了新消息。下次启动消费者时，会继续消费停掉期间新生产的消息。后续行为和ConsumeFromLastOffset类似

//消费者最小线程数量
private int consumeThreadMin = 20;

//消费者最大线程数量
private int consumeThreadMax = 20;

//推模式下任务间隔时间
private long pullInterval = 0;

//推模式下任务拉取的条数,默认32条
private int pullBatchSize = 32;

//消息重试次数,-1代表16次
private int maxReconsumeTimes = -1;

//消息消费超时时间
private long consumeTimeout = 15;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

11.8.2 方法

//订阅消息，并指定队列选择器
void subscribe(final String topic, final MessageSelector selector);

//取消消息订阅
void unsubscribe(final String topic);

//获取消费者对主题分配了那些消息队列
Set<MessageQueue> fetchSubscribeMessageQueues(final String topic);

//注册并发事件监听器
void registerMessageListener(final MessageListenerConcurrently messageListener);

//注册顺序消息事件监听器
void registerMessageListener(final MessageListenerOrderly messageListener);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

注册顺序消息事件监听器
void registerMessageListener(final MessageListenerOrderly messageListener);

1
2
3
4

11.8.3 消费确认(ACK)

业务实现消费回调的时候，当且仅当此回调函数返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS，RocketMQ才会认为这批消息（默认是1条）是消费完成的中途断电，抛出异常等都不会认为成功——即都会重新投递。
返回ConsumeConcurrentlyStatus.RECONSUME_LATER，RocketMQ就会认为这批消息消费失败了。
如果业务的回调没有处理好而抛出异常，会认为是消费失败ConsumeConcurrentlyStatus.RECONSUME_LATER处理。
为了保证消息是肯定被至少消费成功一次，RocketMQ会把这批消息重发回Broker（topic不是原topic而是这个消费组的RETRY topic），在延迟的某个时间点（默认是10秒，业务可设置）后，再次投递到这个ConsumerGroup。而如果一直这样重复消费都持续失败到一定次数（默认16次），就会投递到DLQ死信队列。应用可以监控死信队列来做人工干预。
另外如果使用顺序消费的回调MessageListenerOrderly时，由于顺序消费是要前者消费成功才能继续消费，所以没有RECONSUME_LATER的这个状态，只有SUSPEND_CURRENT_QUEUE_A_MOMENT来暂停队列的其余消费，直到原消息不断重试成功为止才能继续消费。

当消费者成功消费消息会返回Broker一个ACK,告之Broker更新offset，但是RocketMQ并不是按一条一条消息来做ack，而是根据一次拉取批量来做消息ack.
消费者可以返回ACK的值：

• ConsumeConcurrentlyStatus.CONSUME_SUCCESS 消费成功
• ConsumeConcurrentlyStatus.RECONSUME_LATER 消费失败，需要稍后重新消费

如果消息消费失败，我们可以返回（ConsumeConcurrentlyStatus.RECONSUME_LATER），通知mq消息处理失败，重新发送;
若消费一直失败，rockmq默认会重推16次（默认），就会投递到DLQ死信队列。应用可以监控死信队列来做人工干预。当然我们也可以判断超过一定次数后可以不再消费。
if(messageExt.getReconsumeTimes()==3)

如果消息消费成功，我们可以返回（ConsumeConcurrentlyStatus.CONSUME_SUCCESS），表示成功消费。