01｜基本概念

1.1｜Message Model（消息模型）

RocketMQ主要由Producer、Broker、Consumer三部分组成，其中Producer负责生成消息，Consumer负责消费消息，Broker负责存储消息。Broker在实际部署过程中对应一台服务器，每个Broker可以存储多个Topic的消息，每个Topic中的消息可以分片存储与不同的Broker。Message Queue用于存储消息的物理地址。每个Topic的消息地址存储于多个Message Queue中。ConsumerGroup由多个Consumer实例构成。

1.2｜Producer（消息生产者）

负责生产消息，将消息发送到Broker服务器。发送方式分为：

  • 同步发送
  • 异步发送
  • 顺序发送
  • 单向发送

同步和异步都需要Broker返回确认消息，单向发送不需要。

1.3｜Consumer（消息消费者）

负责消费消息，消费者会从 Broker服务器拉取消息。RocketMQ提供来两种消费模式：

  • 拉取（PULL）

  • 推送（PUSH）

    1.4｜TOPIC（主题）

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

    1.5｜Broker Server（代理服务器）

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

    1.6｜Name Server（命名服务器）

    NameServer充当路由消息的提供者，生产者和消费者能够通过NameServer查找各主题相应的Broker IP列表。多个NameServer实例组成集群，但互相独立，没有信息交换（通过多写保证高可用）。

    1.7｜Producer Group（生产者组）

    同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。如果发送的是事务消息且原始生产者在发送之后崩溃，则Broker服务器会联系同一生产者组的其他生产者实例提交或回滚。

    1.8｜Consumer Group（消费者组）

    同一类Consumer的集合，这类Consumer通常消费同一类消息且消费逻辑一致。用于实现消息的负载均衡和容错。消费者组中的消费者实例必须订阅完全相同的Topic。RocketMQ支持两种消息消费模式：

  • 集群消费（Clustering）：消费组中的实例平均分摊消息

  • 广播消费（Broadcasting）：消费组中的实例全量接收消息

    1.9｜Message

    消息系统锁传输的物理载体，生产和消费数据的最小单位，每条消息必须属于一个Topic。RocketMQ中每个消息用于唯一的Message ID，且可以携带具有业务标识的Key。系统提供了通过Message ID和Key查询消息的功能。

    1.10｜Tag

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

    02｜特性

    2.1｜顺序消费

    RocketMQ顺序消费分为全局顺序消费和分区顺序消费：

  • 全局顺序：对于指定的一个 Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。 适用场景：性能要求不高，所有的消息严格按照 FIFO 原则进行消息发布和消费的场景

  • 分区顺序：对于指定的一个 Topic，所有消息根据 sharding key 进行区块分区。 同一个分区内的消息按照严格的 FIFO 顺序进行发布和消费。 Sharding key 是顺序消息中用来区分不同分区的关键字段，和普通消息的 Key 是完全不同的概念。 适用场景：性能要求高，以 sharding key 作为分区字段，在同一个区块中严格的按照 FIFO 原则进行消息发布和消费的场景。

    2.2｜消息过滤

    RocketMQ的消费者可以根据Tag进行消息过滤，也支持自定义属性过滤。消息过滤目前版本在Broker端实现，其优点是减少了对于Consumer无用消息的网络传输，缺点是增加了Broker的负担。

    2.3｜消息可靠性

    影响消息可靠性的几种情况：

    1. Broker非正常关闭
    1. Broker异常Crash
    1. OS Crash
    1. 机器掉电
    1. 机器无法开机
    1. 硬盘设备损坏

1、2、3、4属于硬件资源可立即恢复的情况，RocketMQ能保证在这四种情况下消息不丢，或者丢失少量数据，取决于刷盘方式是同步还是异步。
5、6属于单点故障，且无法恢复，一旦发送，在此单点上的消息全部丢失。RocketMQ通过异步复制，可以保证99%的消息不丢失，但是仍然会有少量消息可能丢失。可通过同步双写完全避免单点问题，但会降低MQ性能，适合对消息可靠性要求极高的场景

2.4｜At Least Once

At Least Once指每个消息必须投递一次。Consumer先Pull消息到本地，消费完成后，才向服务器返回ACK

2.5｜回溯消费

回溯消费是指Consumer已经消费成功的消息，由于业务上需求需要重新消费，要支持此功能，Broker在向Consumer投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度，例如由于Consumer系统故障，恢复后需要重新消费1小时前的数据，那么Broker要提供一种机制，可以按照时间维度来回退消费进度。RocketMQ支持按照时间回溯消费，时间维度精确到毫秒。

2.6｜事务消息

RocketMQ事务消息（Transactional Message）是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致。

2.7｜定时消息

定时消息（延迟队列）是指发送消息到Broker后，不会立即被消费，等待特定时间投递给真正的topic。

2.8｜消息重试

2.9｜消息重投

2.10｜流控

2.11｜死信队列

03｜架构设计

3.1｜技术架构

image.png

RocketMQ架构主要分为四部分，如图所示：

  • 1. Producer：生产者，支持分布式集群部署。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程能够保证低延迟的同时并且支持快速失败。Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic 服务的Master建立长连接，且定时向Master发送心跳。
  • 2. Consumer：消费者，支持分布式集群部署。支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，消费者在向Master拉取消息时，Master服务器会根据拉取偏移量与最大偏移量的距离（判断是否读老消息，产生读I/O），以及从服务器是否可读等因素建议下一次是从Master还是Slave拉取。
  • 3. NameServer：NameServer是一个非常简单的Topic路由注册中心，其角色类似Dubbo中的zookeeper，支持Broker的动态注册与发现。主要包括两个功能：
    • Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据，并且提供心跳检测，检查Broker是否存活；
    • 路由信息管理，每个NameServer将保存关于Broker集群的整个路由信息和用户客户端查询的队列信息，Producer和Consumer 通过NameServer就可以获取到整个Broker集群的路由信息，从而进行消息的投递和消费；

NameServer通常也是集群的方式部署，但个实例间互相不进行信息通讯。Broker会向集群中的每一台NameServer注册自己的路由信息，集群中的每一个NameServer实例上都保存了一份完整的路由信息。当某个NameServer因某种原因下线了，Broker仍然可以向其他NameServer同步路由信息，Producer，Consumer仍然可以动态感知Broker的路由信息。

  • 4. BrokerServer：Broker主要负责消息的存储、投递、查询以及服务的高可用保证，为了保证这些功能，Broker包含了以下几个重要的子模块：
      1. Remoting Module：整个Broker的实体，负责处理来自clients端的请求
      1. Client Manager：负责管理客户端（Producer/Consumer）和维护Consumer的Topic订阅信息
      1. Store Service：提供方便简单的API接口处理消息存储到物理硬盘和查询功能
      1. HA Service：高可用服务，提供Master Broker和Slave Broker之间的数据同步功能
      1. Index Service：根据特定的Message Key对投递到Broker的消息进行索引服务，以提供消息的快速查询

image.png
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的从服务器才会参与消息的读负载。

04｜设计

4.1｜消息存储

image.png

消息存储整体架构

消息存储架构图中主要有下面三个跟消息存储相关的文件构成。

  • 1. CommitLog：消息以及元数据的存储主体，存储Producer端写入的消息主体内容，消息内容不是定长的。单个文件大小默认1G，文件名长度为20位，左边补零，剩余为起始偏移量。消息顺序写入日志文件，当文件写满，则写入下一个文件；
  • 2. ConsumeQueue：消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。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；
  • 3. IndexFile：IndexFile提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是：$HOME \store\index${fileName}，文件名fileName是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现HashMap结构，故rocketmq的索引文件其底层实现为hash索引。

RocketMQ采用的是混合型的存储结构，Broker单个实例下所有的队列共用一个日志数据文件（CommitLog）来存储。针对Producer和Consumer则采用来数据和索引部分相分离的存储结构。（不同的Topic为何不分不同的CommitLog？）

页缓存与内存映射

页缓存（PageCache）是OS对文件的缓存，用于加速对文件的读写。在RocketMQ中，ConsumerQueue逻辑消费队列存储的数据较少，并且是顺序读取，在PageCache机制的预读作用下，ConsumeQueue文件的读性能几乎接近读内存，即使在有消息堆积情况下页不会影响性能。而对于CommitLog消息存储的日志文件来说，读取消息内容时会产生随机IO，从而影响性能。
另外，RocketMQ主要通过MappedByteBuffer对文件进行读写操作。其中，利用了NIO中的FileChannel模型将磁盘上的物理文件直接映射到用户态的内存地址中（这种Mmap的方式减少了传统IO将磁盘文件数据在操作系统内核地址空间的缓冲区和用户应用程序地址空间的缓冲区之间来回进行拷贝的性能开销），将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率（正因为需要使用内存映射机制，故RocketMQ的文件存储都使用定长结构来存储，方便一次将整个文件映射至内存）。

消息刷盘

  • 同步刷盘：只有在消息真正持久化至磁盘后RocketMQ的Broker端才会真正返回给Producer端一个成功的ACK响应。同步刷盘对MQ消息可靠性来说是一种不错的保障，但是性能上会有较大影响，一般适用于金融业务应用该模式较多。

  • 异步刷盘：能够充分利用OS的PageCache的优势，只要消息写入PageCache即可将成功的ACK返回给Producer端。消息刷盘采用后台异步线程提交的方式进行，降低了读写延迟，提高了MQ的性能和吞吐量。

    4.2｜通信机制

    RocketMQ消息队列集群主要包括NameServer、Broker（Master/Slave）、Producer、Consumer4个角色，基本通信流程如下：

    1. Broker启动后需要完成一次将自己注册至NameServer的操作；随后每个30s定时向NameServer上报Topic路由信息；（30s上报，是否意味着Topic可能存在30s延迟？）
    1. 消息生产者Producer作为客户端发送消息时候，需要根据消息的Topic从本地缓存的TopicPublishInfoTable获取路由信息。如果没有则更新路由信息并从NameServer上重新拉取，同时Producer会默认每隔30s向NameServer拉取一次路由信息；
    1. 消息生产者Producer根据2中获取的路由信息选择一个队列（MessageQueue）进行消息发送；Broker接收到消息进行落盘处理；
    1. 消息消费者Consumer根据2中获取的路由信息，并在完成客户端的负载均衡后，选择其中的某一个或某几个消息队列来拉取消息并进行消费；

RocketMQ-Remoting模块是RocketMQ消息队列中负责网络通信的模块，它几乎被其他所有需要网络通信的模块所依赖。
RocketMQ的RPC通信采用Netty组建作为底层通信库，同样也遵循来Reactor多线程模型并且在这之上做了对应的拓展和优化。

4.3｜负载均衡

RocketMQ中的负载均衡都在Client端完成，具体来说的话，主要可以分为Producer端发送消息时候的负载均衡和Consumer端订阅消息的负载均衡。

4.4｜事务消息

RocketMQ在4.3.0版本中已经支持分布式事务消息，这里RocketMQ采用了2PC的思想来实现来提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如图所示：
image.png
事务消息分为两个流程：

  • 事务消息的发送及提交：
    • 发送消息
    • 服务器响应消息写入结果
    • 根据发送结果执行本地事务
    • 根据本地事务状态执行Commit或Rollback，Commit之后消息对消费者可见
  • 补偿流程
    • 对没有Commit/Rollback的事务消息，从服务端发起一次“回查”
    • Producer收到回查消息，检查消息对应的本地事务的状态
    • 根据本地事务的状态，重新Commit或者Rollback

      4.5｜消息查询

      按照MessageId查询消息

      RocketMQ中的MessageId的长度总共有16字节，其中包含了消息存储主机地址（IP地址和端口），消息Commit Log offset。Client端从MessageId中解析出Broker的地址和Commit Log的偏移地址后封装成一个RPC请求后通过Remoting发送。Broker接收到请求后根据Commit Log offset和size去CommitLog中找到真正的记录并解析成一个完整的消息返回。

      按照Message Key查询消息

      “按照Message Key查询消息”，主要是基于RocketMQ的IndexFile索引文件来实现的。RocketMQ的索引文件逻辑结构，类似JDK中HashMap的实现。索引文件的具体结构如下：
      image.png
      IndexFile索引文件为用户提供通过“按照Message Key查询消息”的消息索引查询服务，IndexFile文件的存储位置是：$HOME\store\index${fileName}，文件名fileName是以创建时的时间戳命名的，文件大小是固定的，等于40+500W4+2000W20= 420000040个字节大小。如果消息的properties中设置了UNIQ_KEY这个属性，就用 topic + “#” + UNIQ_KEY的value作为 key 来做写入操作。如果消息设置了KEYS属性（多个KEY以空格分隔），也会用 topic + “#” + KEY 来做索引。
      其中的索引数据包含了Key Hash/CommitLog Offset/Timestamp/NextIndex offset 这四个字段，一共20 Byte。NextIndex offset 即前面读出来的 slotValue，如果有 hash冲突，就可以用这个字段将所有冲突的索引用链表的方式串起来了。Timestamp记录的是消息storeTimestamp之间的差，并不是一个绝对的时间。整个Index File的结构如图，40 Byte 的Header用于保存一些总的统计信息，4500W的 Slot Table并不保存真正的索引数据，而是保存每个槽位对应的单向链表的头。202000W 是真正的索引数据，即一个 Index File 可以保存 2000W个索引。
      “按照Message Key查询消息”的方式，RocketMQ的具体做法是，主要通过Broker端的QueryMessageProcessor业务处理器来查询，读取消息的过程就是用topic和key找到IndexFile索引文件中的一条记录，根据其中的commitLog offset从CommitLog文件中读取消息的实体内容。