MQ_03RocketMQ高性能的核心原理思维导图模板

读队列与写队列

读写分离的思想。RocketMQ在最MessageQueue的路由策略时，就可以通过指向不同的队列来实现读写分离

写队列

会创建对应的存储文件，负责消息写入

读队列

会记录Consumer的Offset，负责消息读取

场景分析

写队列>读队列

会有一部分写队列无法写入到读队列中，这一部分的消息就无法被读取，就会造成消息丢失

消息存入了，但是读不出来

写队列<读队列

一部分读队列里是没有消息写入的。如果有一个消费者被分配的是这些没有消息的读队列， 那这些消费者就无法消费消息，造成消费者空转，极大的浪费性能

写队列=读队列

通常在运行时，都需要设置读队列=写队列

队列缩减·

原来四个队列，现在要缩减成两个队列

先缩减读写队列，那么被缩减的MessageQueue上没有被消费的消息，就会丢失

先缩减写队列，待空出来的读队列上的消息都被消费完了之后，再来缩减读队列，这样就可以比较平稳的实现队列缩减了

消息持久化

RocketMQ消息直接采用磁盘文件保存消息，默认路径在${user_home}/store目录。这些存储目录可以在broker.conf中自行指定

存储文件主要部分

CommitLog

存储消息的元数据

所有消息都会顺序存入到CommitLog文件当中

CommitLog由多个文件组成，每个文件固定大小1G。以第一条消息的偏移量为文件名

文件结构

件大小是固定的，但是其中存储的每个消息单元长度是不固定

具体格式

ConsumerQueue

存储消息在CommitLog的索引

一个MessageQueue一个文件，记录当前MessageQueue被哪些消费者组消费到了哪一条CommitLog

文件结构

固定由30万个固定大小20byte的数据块组成

msgPhyOffset(8byte，消息在文件中的起始位置)

msgSize(4byte，消息在文件中占用的长度)

msgTagCode(8byte，消息的tag的Hash值)

常量CQ_STORE_UNIT_SIZE=20，这个常量就表示一个数据块的大小

IndexFile

为了消息查询提供了一种通过key或时间区间来查询消息的方法，这种通过IndexFile来查找消息的方法不影响发送与消费消息的主流程

文件结构

indexHeader(固定40byte)

slot(固定500W个，每个固定20byte)

index(最多500W*4个，每个固定20byte)

辅助的存储文件

checkpoint

数据存盘检查点

主要记录commitlog文件、ConsumeQueue文件以及IndexFile文件最后一次刷盘的时间戳

config/*.json

将RocketMQ的一些关键配置信息进行存盘保存

例如Topic配置、消费者组配置、消费者组消息偏移量Offset

abort

RocketMQ用来判断程序是否正常关闭的一个标识文件

正常情况下，会在启动时创建，而关闭服务时删除

如果遇到一些服务器宕机，或者kill -9这样一些非正常关闭服务的情况，这个abort文件就不会删除， 因此RocketMQ就可以判断上一次服务是非正常关闭的，后续就会做一些数据恢复的操作

要点

1.所有生产者发过来的消息，都会无差别的依次存储到Commitlog文件当中

好处是可以减少查找目标文件的时间，让消息以最快的速度落盘

对比kafka

Kafka需要寻找消息所属的Partition文件，再完成写入

当Topic比较多时，这样的Partition寻址就会浪费比较多的时间，所以Kafka不太适合多Topic的场景

RocketMQ的这种快速落盘的方式在多Topic场景下，优势就比较明显

2.每次存CommitLog文件时，都会去检查当前CommitLog文件空间是否足够， 如果不够的话，就重新创建一个CommitLog文件。文件名为当前消息的偏移量

3.ConsumeQueue文件主要是加速消费者的消息索引

每个文件夹对应RocketMQ中的一个MessageQueue，文件夹下的文件记录了每个MessageQueue中的消息在CommitLog文件当中的偏移量

消费者通过ComsumeQueue文件，就可以快速找到CommitLog文件中感兴趣的消息记录

消费者在ConsumeQueue文件当中的消费进度，会保存在config/consumerOffset.json文件当中

4.IndexFile文件主要是辅助消息检索

消费者进行消息消费时，通过ConsumeQueue文件就足够完成消息检索了

如果要按照MeessageId或者MessageKey来检索文件IndexFile文件就是用来辅助这类消息检索的

过期文件删除

消息既然要持久化，就必须有对应的删除机制

判断过期文件

CommitLog文件和ConsumeQueue文件都是以偏移量命名

对于非当前写的文件，如果超过了一定的保留时间，那么这些文件都会被认为是过期文件，随时可以删除

保留时间就是在broker.conf中配置的fileReservedTime属性

RocketMQ判断文件是否过期的唯一标准就是非当前写文件的保留时间，而并不关心文件当中的消息是否被消费过

所以，RocketMQ的消息堆积也是有时间限度的

删除过期文件

RocketMQ内部有一个定时任务，对文件进行扫描，并且触发文件删除的操作

用户可以指定文件删除操作的执行时间。在broker.conf中deleteWhen属性指定。默认是凌晨四点

RocketMQ还会检查服务器的磁盘空间是否足够，如果磁盘空间的使用率达到一定的阈值，也会触发过期文件删除

broker的磁盘空间不要少于4G(官方建议)

高效文件写

零拷贝技术加速文件读写

所谓的零拷贝技术，其实并不是不拷贝，而是要尽量减少CPU拷贝

CPU拷贝

应用程序需要与网卡、磁盘等硬件进行数据交互时，就需要在用户态和内核态之间来回的复制数据

这些操作，原本都是需要由CPU来进行任务的分配、调度等管理步骤

当发生大规模的数据读写操作时，CPU的占用率会非常高

DMA拷贝

由DMA来负责这些频繁的IO操作

DMA是一套独立的指令集，不会占用CPU的计算资源

mmap文件映射机制

文件的读写操

应用程序对磁盘文件的读与写，都需要经过内核态与用户态之间的状态切换，每次状态切换的过程中，就需要有大量的数据复制

在这个过程中，总共需要进行四次数据拷贝

磁盘与内核态之间的数据拷贝，在操作系统层面已经由CPU拷贝优化成了DMA拷贝

内核态与用户态之间的拷贝依然是CPU拷贝

零拷贝技术优化的重点，就是内核态与用户态之间的这两次拷贝

mmap文件映射的方式，就是在用户态不再保存文件的内容，而只保存文件的映射

文件的内存起始地址

文件大小

真实的数据，也不需要在用户态留存，可以直接通过操作映射，在内核态完成数据复制

sendFile机制

早期的sendfile实现机制其实还是依靠CPU进行页缓存与socket缓存区之间的数据拷贝

在拷贝过程中，并不直接拷贝文件的内容，而是只拷贝一个带有文件位置和长度等信息的文件描述符FD，这样就大大减少了需要传递的数据

最后，sendfile机制在内核态直接完成了数据的复制，不需要用户态的参与，所以这种机制的传输效率是非常稳定的

sendfile机制非常适合大数据的复制转移

顺序写加速文件写入磁盘

通常应用程序往磁盘写文件时，由于磁盘空间不是连续的，会有很多碎片

我们去写一个文件时，也就无法把一个文件写在一块连续的磁盘空间中，而需要在磁盘多个扇区之间进行大量的随机写

这个过程中有大量的寻址操作，会严重影响写数据的性能

顺序写机制是在磁盘中提前申请一块连续的磁盘空间，每次写数据时，就可以避免这些寻址操作，直接在之前写入的地址后面接着写就行

DefaultAppendMessageCallback的doAppend方法。在这个方法中，会以追加的方式将消息先写入到一个堆外内存byteBuffer中，然后再通过fileChannel写入到磁盘

刷盘机制保证消息不丢失

当应用程序写入一个文件时，文件内容并不会直接写入到硬件当中，而是会先写入到操作系统中的一个缓存PageCache中

PageCache

PageCache缓存以4K大小为单位，缓存文件的具体内容

PageCache依然是内存状态，所以一断电就会丢失

需要将内存状态的数据写入到磁盘当中，这样数据才能真正完成持久化，断电也不会丢失。这个过程就称为刷盘

操作系统只会在某些特定的时刻将PageCache写入到磁盘

对于有数据修改的PageCache，会标记为Dirty(脏页)状态。当Dirty Page的比例达到一定的阈值时，就会触发一次刷盘操作

操作系统也提供了一个系统调用，应用程序可以自行调用这个系统调用，完成PageCache的强制刷盘

同步刷盘

在返回写成功状态时，消息已经被写入磁盘

消息写入内存的PAGECACHE后，立刻通知刷盘线程刷盘， 然后等待刷盘完成，刷盘线程执行完成后唤醒等待的线程，返回消息写 成功的状态

同步刷盘机制会更频繁的调用fsync，所以吞吐量相比异步刷盘会降低，但是数据的安全性会得到提高

异步刷盘

在返回写成功状态时，消息可能只是被写入了内存的PAGECACHE，写操作的返回快，吞吐量大

当内存里的消息量积累到一定程度时，统一触发写磁盘动作，快速写入

配置方式

刷盘方式是通过Broker配置文件里的flushDiskType 参数设置的，这个参数被配置成SYNC_FLUSH、ASYNC_FLUSH中的 一个

消息主从复制

Broker以一个集群的方式部署，会有一个master节点和多个slave节点，消息需要从Master复制到Slave上

同步复制

同步复制是等Master和Slave都写入消息成功后才反馈给客户端写入成功的状态

在同步复制下，如果Master节点故障，Slave上有全部的数据备份，这样容易恢复数据。但是同步复制会增大数据写入的延迟，降低系统的吞吐量

异步复制

异步复制是只要master写入消息成功，就反馈给客户端写入成功的状态。然后再异步的将消息复制给Slave节点

在异步复制下，系统拥有较低的延迟和较高的吞吐量。但是如果master节点故障，而有些数据没有完成复制，就会造成数据丢失

配置方式

消息复制方式是通过Broker配置文件里的brokerRole参数进行设置的，这个参数可以被设置成ASYNC_MASTER、 SYNC_MASTER、SLAVE三个值中的一个

负载均衡

Producer负载均衡

Producer发送消息时，默认会轮询目标Topic下的所有MessageQueue，并采用递增取模的方式往不同的MessageQueue上发送消息， 以达到让消息平均落在不同的queue上的目的

Consumer负载均衡

Consumer也是以MessageQueue为单位来进行负载均衡。分为集群模式和广播模式

集群模式

集群消费模式下，每条消息只需要投递到订阅这个topic的Consumer Group下的一个实例即可

每当实例的数量有变更，都会触发一次所有实例的负载均衡，这时候会按照queue的数量和实例的数量平均分配queue给每个实例

分配算法分别对应AllocateMessageQueueStrategy下的六种实现类

AllocateMachineRoomNearby

将同机房的Consumer和Broker优先分配在一起

可以通过一个machineRoomResolver对象来定制Consumer和Broker的机房解析规则

还需要引入另外一个分配策略来对同机房的Broker和Consumer进行分配

AllocateMessageQueueAveragely

平均分配。将所有MessageQueue平均分给每一个消费者

AllocateMessageQueueAveragelyByCircle

轮询分配。轮流的给一个消费者分配一个MessageQueue

AllocateMessageQueueByConfig

不分配，直接指定一个messageQueue列表。类似于广播模式，直接指定所有队列

AllocateMessageQueueByMachineRoom

按逻辑机房的概念进行分配。又是对BrokerName和ConsumerIdc有定制化的配置

AllocateMessageQueueConsistentHash

一致性哈希策略只需要指定一个虚拟节点数，是用的一个哈希环的算法，虚拟节点是为了让Hash数据在换上分布更为均匀

广播模式

每一条消息都会投递给订阅了Topic的所有消费者实例，所以也就没有消息分配这一说

Consumer分配Queue时，所有Consumer都分到所有的Queue

关键是将消费者的消费偏移量不再保存到broker当中，而是保存到客户端当中，由客户端自行维护自己的消费偏移量