消息中间件
rocketmq

rocketmq-消息的存储和发送

由于分布式消息队列对于可靠性的要求比较高，所以需要保证生产者将消息发送到broker之后，保证消 息是不出现丢失的，因此消息队列就少不了对于可靠性存储的要求。

MQ消息存储选择

从主流的几种MQ消息队列采用的存储方式来看，主要会有三种

  1. 分布式KV存储，比如ActiveMQ中采用的levelDB、Redis， 这种存储方式对于消息读写能力要求不 高的情况下可以使用
  2. 文件系统存储，常见的比如kafka、RocketMQ、RabbitMQ都是采用消息刷盘到所部署的机器上的 文件系统来做持久化，这种方案适合对于有高吞吐量要求的消息中间件，因为消息刷盘是一种高效 率，高可靠、高性能的持久化方式，除非磁盘出现故障，否则一般是不会出现无法持久化的问题
  3. 关系型数据库，比如ActiveMQ可以采用mysql作为消息存储，关系型数据库在单表数据量达到千 万级的情况下IO性能会出现瓶颈，所以ActiveMQ并不适合于高吞吐量的消息队列场景。

总的来说，对于存储效率，文件系统要优于分布式KV存储，分布式KV存储要优于关系型数据库

RocketMQ就是采用文件系统的方式来存储消息，消息的存储是由ConsumeQueue和CommitLog配合 完成的。CommitLog是消息真正的物理存储文件。ConsumeQueue是消息的逻辑队列，有点类似于数据库的索引文件，里面存储的是指向CommitLog文件中消息存储的地址。

每个Topic下的每个Message Queue都会对应一个ConsumeQueue文件，文件的地址是: {store_home}/consumequeue/{topicNmae}/{queueId}/​{filename}，默认路径: /root/store

在rocketMQ的文件存储目录下，可以看到这样一个结构的的而文件。

我们只需要关心Commitlog、Consumequeue、Index。

CommitLog

CommitLog是用来存放消息的物理文件，每个broker上的commitLog被当前机器上的所有 consumerQueue共享，不做任何的区分。

CommitLog中的文件默认大小为1G，可以动态配置；当一个文件写满以后，会生成一个新的 commitlog文件。所有的Topic数据是顺序写入在CommitLog文件中的。

文件名的长度为20位，左边补0，剩余未起始偏移量，比如
00000000000000000000 表示第一个文件， 文件大小为1024X1024X1024，当第一个文件写满之后，生

成第二个文件
000000000001073741824 表示第二个文件，起始偏移量为1073741824

ConsumeQueue

consumeQueue表示消息消费的逻辑队列，这里面包含MessageQueue在commitlog中的其实物理位 置偏移量offset，消息实体内容的大小和Message Tag的hash值。对于实际物理存储来说， consumeQueue对应每个topic和queueid下的文件，每个consumeQueue类型的文件也是有大小，每 个文件默认大小约为600W个字节，如果文件满了后会也会生成一个新的文件

IndexFile

索引文件，如果一个消息包含Key值的话，会使用IndexFile存储消息索引。Index索引文件提供了对 CommitLog进行数据检索，提供了一种通过key或者时间区间来查找CommitLog中的消息的方法。在物 理存储中，文件名是以创建的时间戳命名，固定的单个IndexFile大小大概为400M，一个IndexFile可以 保存2000W个索引

abort

broker在启动的时候会创建一个空的名为abort的文件，并在shutdown时将其删除，用于标识进程是否正常退出，如果不正常退出，会在启动时做故障恢复

消息存储的整体结构

RocketMQ的消息存储采用的是混合型的存储结构，也就是Broker单个实例下的所有队列公用一个日志 数据文件CommitLog。这个是和Kafka又一个不同之处。

为什么不采用kafka的设计，针对不同的partition存储一个独立的物理文件呢?这是因为在kafka的设计 中，一旦kafka中Topic的Partition数量过多，队列文件会过多，那么会给磁盘的IO读写造成比较大的压 力，也就造成了性能瓶颈。所以RocketMQ进行了优化，消息主题统一存储在CommitLog中。

当然，这种设计并不是银弹，它也有它的优缺点：

优点在于:由于消息主题都是通过CommitLog来进行读写，ConsumerQueue中只存储很少的数据， 所以队列更加轻量化。对于磁盘的访问是串行化从而避免了磁盘的竞争

缺点在于:消息写入磁盘虽然是基于顺序写，但是读的过程确是随机的。读取一条消息会先读取 ConsumeQueue，再读CommitLog，会降低消息读的效率。

消息发送到消息接收的整体流程

  1. Producer将消息发送到Broker后，Broker会采用同步或者异步的方式把消息写入到CommitLog。 RocketMQ所有的消息都会存放在CommitLog中，为了保证消息存储不发生混乱，对CommitLog 写之前会加锁，同时也可以使得消息能够被顺序写入到CommitLog，只要消息被持久化到磁盘文 件CommitLog，那么就可以保证Producer发送的消息不会丢失。

  1. commitLog持久化后，会把里面的消息Dispatch到对应的Consume Queue上，Consume Queue 相当于kafka中的partition，是一个逻辑队列，queue中的消息存储了这个实际的消息在CommiLog中的起始offset， log大小和MessageTag的hashCode。

  1. 当消费者进行消息消费时，会先读取consumerQueue , 逻辑消费队列ConsumeQueue保存了指 定Topic下的队列消息在CommitLog中的起始物理偏移量Offset，消息大小、和消息Tag的 HashCode值

  1. 直接从consumequeue中读取消息是没有数据的，真正的消息主体在commitlog中，所以还需要 从commitlog中读取消息

什么时候清理物理消息文件?

消息存储在CommitLog之后，的确是会被清理的，但是这个清理只会在以下任一条件成立才会批量删除消息文件(CommitLog):

  1. 消息文件过期(默认72小时)，且到达清理时点(默认是凌晨4点)，删除过期文件。
  2. 消息文件过期(默认72小时)，且磁盘空间达到了水位线(默认75%)，删除过期文件。
  3. 磁盘已经达到必须释放的上限(85%水位线)的时候，则开始批量清理文件(无论是否过期)，直到空间充足。

注:若磁盘空间达到危险水位线(默认90%)，出于保护自身的目的，broker会拒绝写入服务。