存储层

一、消息存储

为了保证RocketMQ的高可靠性要求，MQ会对消息进行持久化。 

1. 生产者发送消息
2. MQ收到消息后，对消息进行持久化，在储存中新增一条记录
3. 返回ACK给生产者
4. MQ推送消息给对应的消费者，然后等待消费者返回ACK
5. 如果消费者返回了ACK，则MQ执行消息删除的操作，若超时未返回，MQ会认为消费失败重新推送消息，重复执行4，5，6步骤
6. MQ删除消息

二、存储介质

关系型数据库

例如Apache的ActiveMQ就可以通过JDBC来进行消息的持久化，设置一些XML的配置信息即可。但是，普通的关系型数据库，如MySQL，在数据量达到千万级别后会造成IO的性能瓶颈。在可靠性方面，这种方式非常依赖于数据库的性能，一旦数据库宕机，就会导致消息无法存储的线上故障。

文件系统

目前业界较为常用的几款产品（RocketMQ/Kafka/RabbitMQ）均采用的是消息刷盘至所部署虚拟机/物理机的文件系统来做持久化（刷盘一般可以分为异步刷盘和同步刷盘两种模式）。消息刷盘为消息存储提供了一种高效率、高可靠性和高性能的数据持久化方式。除非部署MQ机器本身或是本地磁盘挂了，否则一般是不会出现无法持久化的故障问题。

所以在性能上：文件系统＞关系型数据库

三、消息的存储与发送的性能保障

消息存储

我们知道，消息多是通过网络发送过来的，RocketMQ如何保证消息存储的速度匹配上网络传输速度呢。其实以目前大多数SATA3接口的SSD顺序写的速度能达到600MB/s（更不说M.2接口了），但随机写差不多只有100KB/s，相差了6000倍。所以RocketMQ在存储消息时采用顺序写的方式来保证效率。

消息发送（指MQ）

一台MQ服务器把本地消息发送至客户端分两步

1. 从本地读取消息；
2. 将读取的内容通过网络发送出去。

因linux系统分为【用户态】和【内核态】，文件操作、网络操作需要涉及这两种形态的切换，免不了进行数据复制。

看似只有两步，实际在系统内部进行了四次数据的复制，如果数据非常大的话，4次复制会导致读的速度非常慢。

1. 从磁盘复制数据到内核态内存；
2. 从内核态内存复制到用户态内存（也就是应用程序独占的内存）；
3. 然后从用户态内存复制到网络驱动的内核态内存；
4. 最后是从网络驱动的内核态内存复 制到网卡中进行传输。

那么RocketMQ是如何提高读的效率的呢，它是利用了Java中的MappedByteBuffer来实现一个叫做“零拷贝”的技术，通过省去数据向用户态内存的复制，来提高读取速度。

四、消息存储结构

#存储路径
storePathRootDir=/opt/rocketmq/store-m
#commitLog 存储路径
storePathCommitLog=/opt/rocketmq/store-m/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/opt/rocketmq/store-m/consumequeue
#消息索引存储路径
storePathIndex=/opt/rocketmq/store-m/index
#checkpoint 文件存储路径
storeCheckpoint=/opt/rocketmq/store-m/checkpoint
#abort 文件存储路径
abortFile=/opt/rocketmq/store-m/abort

对应系统中 

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

CommitLog：存储消息的元数据，发送的消息的所有信息（Topic，QueueId，Message）都会存到这里面

我们可以看到，查看CommitLog，里面就有一个大小为1G，用于存储消息的文件（当前只有一个，消息多了会自动创建新的，大小也是1G）。

这是因为前面提到的，RocketMQ使用MappedByteBuffer这种内存映射的方式有几个限制，其中之一是一次只能映射1.5~2G 的文件至用户态的虚拟内存，所以每一个文件的大小是1G（小于2G即可）。 

ConsumerQueue：存储的是CommitLog里每个消息的索引，为的是提升消息读取的效率

ConsumerQueue里面存了消息最小的偏移量，已经消费的偏移量，最大的偏移量，且丢失后能通过CommitLog还原。 可以看到每一个Topic都有四个消息队列（0，1，2，3），每一个消息队列会对应一个ConsumerQueue。其和CommitLog里的文件同名，大小不大。也就意味着Linux能够将其加载到内存中读取，进一步加快读消息的速度。

IndexFile：也是索引文件，与ConsumerQueue通过偏移量查找不同，IndexFile提供了通过时间区间或key来查的方式

五、刷盘机制

同步刷盘

在返回写成功状态时，消息已经被写入磁盘。具体流程是，消息写入内存的PAGECACHE后，立刻通知刷盘线程刷盘， 然后等待刷盘完成，刷盘线程执行完成后唤醒等待的线程，返回消息写成功的状态。

异步刷盘

在返回写成功状态时，消息可能只是被写入了内存的PAGECACHE，写操作的返回快，吞吐量大；当内存里的消息量积累到一定程度时，统一触发写磁盘动作，快速写入。

#配置:刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=SYNC_FLUSH

六、高可用性机制

RocketMQ 分布式集群是通过Master 和Slave 的配合达到高可用性的

Master 和Slave 的区别：在Broker 的配置文件中，参数brokerId的值为0 表明这个Broker 是Master ，大于0 表明这个Broker 是Slave ，同时broker Role 参数也会说明这个Broker 是Master 还是Slave

Master 角色的Broker 支持读和写， Slave 角色的Broker 仅支持读，也就是Producer 只能和Master 角色的Broker 连接写入消息； Consumer 可以连接Master 角色的Broker ，也可以连接Slave角色的Broker来读取消息。

消息消费高可用

在Consumer 的配置文件中，并不需要设置是从Master 读还是从Slave读，当Master 不可用或者繁忙的时候， Consumer 会被自动切换到从Slave 读。有了自动切换Consumer 这种机制，当一个Master 角色的机器出现故障后，Consumer 仍然可以从Slave 读取消息，不影响Consumer 程序。这就达到了消费端的高可用性。

消息发送高可用

在创建Topic 的时候，把Topic的多个Message Queue 创建在多个Broker 组上（相同Broker 名称，不同broker Id的机器组成一个Broker组），这样当一个Broker 组的Master 不可用后，其他组的Master 仍然可用， Producer 仍然可以发送消息。RocketMQ 目前还不支持把Slave 自动转成Master ，如果机器资源不足，需要把Slave 转成Master ，则要手动停止Slave 角色的Broker ，更改配置文件，用新的配置文件启动Broker 。

七、消息主从复制

如果一个Broker 组有Master 和Slave, 消息需要从Master 复制到Slave上，有同步和异步两种复制方式。

同步复制

同步复制方式是等Master 和Slave 均写成功后才反馈给客户端写成功状态；

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

异步复制

异步复制方式是只要Master 写成功即可反馈给客户端写成功状态。

在异步复制方式下，系统拥有较低的延迟和较高的吞吐量，但是如果Master 出了故障，有些数据因为没有被写人Slave ，有可能会丢失

配置

同步复制和异步复制是通过Broker 配置文件里的brokerRole 参数进行设置的，这个参数可以被设置成ASYNC_MASTER 、SYNC_MASTER 、SLAVE 三个值中的一个。

总结

实际应用中要结合业务场景，合理设置刷盘方式和主从复制方式，尤其是SYNC_FLUSH 方式，由于频繁地触发磁盘写动作， 会明显降低性能。通常情况下，应该把Master 和Slave 配置成ASYNC_FLUSH 的刷盘方式，主从之间配置成SYNC_MASTER 的复制方式，这样即使有一台机器出故障， 仍然能保证数据不丢，是个不错的选择。