RocketMQ高性能核心原理与源码架构剖析

读、写队列

采用读写分离的方式，RocketMQ在创建Topic的时候会单独设置读队列和写队列，写队列负责写入以及同步数据到读队列，读队列会记录消费者的offset，负责消息拉取，通过MessageQueue的路由策略进行队列读、写指向
- 如果写队列大于读队列，就会有一部分写队列无法同步到读队列，这样会造成数据丢失(消息存入了但是读取不到)
- 如果写队列小于读队列，就会有一部分读队列无消息写入，造成消费者空转，耗费性能
综述
- 写队列 > 读队列，会造成消息丢失
- 写队列 < 读队列，会造成消费者空转，耗费性能
- 如果当Topic的MessageQueue进行缩减时，可以适当调整读写队列的数量
  - 先缩减写队列，待空出来的读队列上的消息消费完之后再去缩减读队列

消息持久化

RocketMQ采用顺序写磁盘来保证消息存储的速度以及mmap零拷贝技术来保证文件传输速度，文件存储结构采用稀疏索引方式(类似Kafka)
存储文件主要分为3部分
- CommitLog
  - 存储消息的所有实体
    - 消息都会被写入到CommitLog文件中，CommitLog由多个文件组成，没过文件固定1G，以首条消息的offset作为文件名
      - 对于生产者发送过来的消息都会依次存储到CommitLog文件中，相对于Kafka还需要寻找分区文件才能写入，减少了查找目标文件的时间，所以Kafka不适合过多Topic的场景
- ComsumerQueue
  - 存储消息在CommitLog的索引
    - 一个MessageQueue对应一个文件，记录当前MessageQueue被消费者组消费到哪个CommitLog
      - 消费者可以通过ComsumerQueue文件中的消息索引定位需要的消息记录
- IndexFile
  - 为了消息查询提供一种通过key或时间区间来查询消息的方法，这样不影响发送与消费消息的主流程
    - IndexFile文件主要是辅助消息检索，消费者可以通过ComsumerQueue文件中的消息索引定位需要的消息记录，但是如果要按照消息ID或者消息Key来检索文件，可以通过IndexFile文件来检索
      - IndexFile文件名为一串时间戳可以辅助检索时间区间
      - IndexFile结构与hash表类似，固定数量的槽位，每个槽位对应一条索引链(链表)，槽位的值对应最新的索引号
何时存储消息
- MQ收到一条消息后，会向生产者进行反馈后再进行消息存储，当MQ推送消息给消费者时，会等待消费者反馈后再进行消息标记为已消费，如果消息者一直不反馈就会不断重发，当达到一定次数就会跳过该消息，并且MQ会定期清理一些过期的消息
  - 不管是生产端还是消费端都需要进行ACK之后才能进行持久化

过期文件删除

消息既然需要持久化也需要对应的过期删除机制
如何判断过期文件
- RocketMQ中CommitLog文件和ComsumerQueue文件都是以偏移量命名，对于非当前写的文件如果超过了一定的保留时间会被认定为过期文件，随时都可以删除，所以对于RocketMQ的消息堆积也是有一定时间的，从而也会由于消息未消费导致消息丢失
何时删除过期文件
- RocketMQ中默认凌晨4点执行定时任务进行文件扫描，触发过期文件删除操作，如果磁盘空间不充足也会触发，所以官方建议Broker的磁盘空间不能少于4G

数据刷盘机制

同步刷盘
- 边写边存盘
  - 消息写入操作系统的页缓存后通知刷盘线程进行刷盘，刷盘成功之后唤醒等待的线程以及消息写成功的状态，保证了数据一定刷盘成功，吞吐量较小
异步刷盘
- 先写后续再刷盘
  - 消息可能只是写入操作系统的页缓存，就返回写入成功，但是会等待积累到一定程度去进行刷盘，保证了响应速度，但是容易丢数据

消息主从复制

Broker如果以主从集群架构进行部署，一个master会有多个slave，master会将数据同步到slave
同步复制
- master和slave的数据都写入成功之后才进行反馈，如果master故障，slave仍有数据备份，方便数据恢复，但是可能因为数据写入延迟降低了吞吐量
异步复制
- 保证master写入成功就进行反馈，再通过异步同步数据到slave，如果master出现故障会导致slave无法同步数据导致数据丢失

负载均衡

生产端负载均衡

生产端在发送消息时，默认会轮询目标Topic下的所有MessageQueue，并采用递增取模的方式往不同的MessageQueue上法消息，来达到让消息平摊到不同的MessageQueue上，而由于MessageQueue分布在不同的Broker上，所以消息也会存在于不同的Broker上
同时生产端在发消息时可以指定一个选择器(MessageQueueSelector)来保证消息的局部有序

消费端负载均衡

消费端是以MessageQueue为单位进行负载均衡的，分为集群模式和广播模式
- 集群模式
  - 在集群模式下，一条被订阅的消息在每个消费组只能被一个消费者消费，RocketMQ采用主动拉取的方式来消费消息，在拉取前需要指定MessageQueue，当消费端节点数量有变化时，都会触发一次负载均衡，会将MessageQueue数量以指定的分配算法摊到每个消费端节点
    - 同机房分配(AllocateMachineRoomNearby)
      - 将同机房的Consumer和Broker优先分配在一起，可以定制化规则
    - MessageQueue平均分配(AllocateMessageQueueAveragely)
      - 将所有MessageQueue平均分给每一个消费者
    - 不分配(AllocateMessageQueueByConfig
      - 直接指定一个MessageQueue列表，类似于广播模式，直接指定所有队列
    - 逻辑机房分配(AllocateMessageQueueByMachineRoom)
      - 按逻辑机房的概念进行分配，可以定制化规则
    - hash分配(AllocateMessageQueueConsistentHash)
      - 一致性哈希策略只需要指定一个虚拟节点数，是用的一个哈希环的算法，虚拟节点是为了让Hash数据在换上分布更为均匀
- 广播模式
  - 在广播模式下，一条被订阅的消息会发送给所有消费者，广播模式实现的关键是将消费者消费的偏移量保存在消费端而不是Broker进行维护

消息重试

首先对于广播模式下不存在消息重试机制，对于普通的消息，当消费失败之后可以进行消息重试
如何让消息进行重试
- 可以通过三种配置
  - 返回Action.ReconsumeLater(推荐)
  - 返回null
  - 抛出异常
- 如果希望消费失败之后不充实，可以直接返回Action.CommitMessage
重试消息如何处理
- 每个消费者都会维护一个重试队列，重试的消息会被安排进去(“%RETRY%”+ConsumeGroup)，默认允许重试16次，当达到阈值会被安排进入到死信队列(%DLQ%+ConsumGroup)
- 死信队列
  - 当一条消息消费失败自动重试一定次数之后，RocketMQ不会立刻丢弃而是安排到死信队列中
  - 特征
    - 一个死信队列对应一个ConsumGroup，而不是对应某个消费者实例
    - 如果一个ConsumeGroup没有产生死信，RocketMQ就不会为其创建相应的死信队列
    - 一个死信队列包含了这个ConsumeGroup里的所有死信消息，而不区分该消息属于哪个Topic
    - 死信队列中的消息不会再被消费者正常消费
    - 死信队列的有效期跟正常消息相同

消息幂等

在MQ系统中，对于消息幂等有三种实现语义
- at most once 最多⼀次：每条消息最多只会被消费⼀次
  - 可以⽤异步发送、sendOneWay等⽅式就可以保证
- at least once ⾄少⼀次：每条消息⾄少会被消费⼀次
  - 可以⽤同步发送、事务消息等很多⽅式能够保证
- exactly once 刚刚好⼀次：每条消息都只会确定的消费⼀次
  - RocketMQ只能保证at least once，保证不了exactly once
    - 云上版本支持
消息幂等的必要性
- 出现重复的情况
  - 发送时消息重复
    - 当⼀条消息已被成功发送到服务端并完成持久化，此时出现了⽹络闪断或者客户端宕机，导致服务端对客户端
      应答失败，如果此时⽣产者意识到消息发送失败并尝试再次发送消息，消费者后续会收到两条内容相同并且消息ID也相同的消息
  - 投递时消息重复
    - 消息消费的场景下，消息已投递到消费者并完成业务处理，当客户端给服务端反馈应答的时候⽹络闪断，为
      了保证消息⾄少被消费⼀次，Broker端将在⽹络恢复后再次尝试投递之前已被处理过的消息，消费者后续会收到两条内容相同并且消息ID也相同的消息
  - 负载均衡时消息重复(不限于⽹络抖动、Broker 重启以及订阅⽅应⽤重启)
    - 当 Broke端或客户端重启、扩容或缩容时，会触发Rebalance，此时消费者可能会收到重复消息
- 处理方式
  - 在RocketMQ中，是⽆法保证每个消息只被投递⼀次的，所以要在业务上⾃⾏来保证消息消费的幂等性，RocketMQ的每条消息都有⼀个唯⼀的MessageId，这个参数在多次投递的过程中是不会改变的，所以业务上可以⽤这个MessageId来作为判断幂等的关键依据，但是最好使用分布式ID来避免出现冲突

源码关注重点

NameServer的启动过程
- 在RocketMQ集群中，实际进行消息存储、推送等核⼼功能点的是Broker，⽽NameServer的作⽤，其实和微服务中的注册中⼼⾮常类似，他只是提供了Broker端的服务注册与发现功能
Broker服务启动过程
- Broker是整个RocketMQ的业务核⼼《所有消息存储、转发这些重要的业务都是Broker进⾏处理
Netty服务注册框架
- Netty的所有远程通信功能都由remoting模块实现
- 在RocketMQ中，涉及到的远程服务⾮常多，在RocketMQ中，NameServer主要是RPC的服务端RemotingServer，Broker对于客户端来说，是RPC的服务端RemotingServer，⽽对于NameServer来说，⼜是RPC的客户端
- RocketMQ基于Netty保持客户端与服务端的⻓连接Channel，只要Channel是稳定的，那么即可以从客户端发请求到服务端，同样服务端也可以发请求到客户端
Broker⼼跳注册管理
- Broker会在启动时向所有NameServer注册⾃⼰的服务信息，并且会定时往NameServer发送⼼跳信，⽽NameServer会维护Broker的路由列表，并对路由表进⾏实时更新
Producer发送消息过程
- Producer有两种
  - 普通发送者(DefaultMQProducer)
    - 只负责发送消息，发送完消息，就可以停⽌了
  - 事务消息发送者(TransactionMQProducer)
    - ⽀持事务消息机制，需要在事务消息过程中提供事务状态确认的服务，这就要求事务消息发送者虽然是⼀个客户端，但是也要完成整个事务消息的确认机制后才能退出
Consumer拉取消息过程
- 消费者也是有两种，推模式消费者和拉模式消费者
- 消费者组之间有集群模式和⼴播模式两种消费模式
- 消费者端的负载均衡
- 消费者端消息的有序性
  - MessageListenerConcurrently
  - MessageListenerOrderly
延迟消息机制
- 延迟消息的核⼼使⽤⽅法就是在Message中设定⼀个MessageDelayLevel参数，对应18个延迟级别。然后Broker中会创建⼀个默认的Schedule_Topic主题，这个主题下有18个队列，对应18个延迟级别，消息发过来之后，会先把消息存⼊Schedule_Topic主题中对应的队列，然后等延迟时间到了，再转发到⽬标队列，推送给消费者进⾏消费
⻓轮询机制
- RocketMQ对消息消费者提供了Push推模式和Pull拉模式两种消费模式，但是这两种消费模式的本质其实都是Pull拉模式，Push模式可以认为是⼀种定时的Pull机制
- 当使⽤Push模式时，RocketMQ实现了⼀种⻓轮询机制(long polling)
  - 当Broker接收到Consumer的Pull请求时，判断如果没有对应的消息，不⽤直接给Consumer响应，⽽是就将这个Pull请求给缓存起来
  - 当Producer发送消息过来时，增加⼀个步骤去检查是否有对应的已缓存的Pull请求，如果有，就及时将请求从缓存中拉取出来，并将消息通知给Consumer