RocketMQ是一个消息中间件而Kafka最早是做日志处理的,它们的产生的出发点不一样，设计的时候解决问题的目标也是不一样的。

每种消息中间件设计的时候是解决某一种问题的。

对于程序来说我们使用消息中间件更关心的是消息的吞吐量、消息的可靠性、消息的性能、消息的可扩展性、消息的失败重试、事务消息等等。我们程序员更关心应用层面的东西。

而Kafka日志处理是允许消息丢失，也可以允许消息堆积的。

在每年的双十一当天,整个阿里巴巴集团通过RocketMQ流转的线上消息达到了万亿级,峰值TPS达到了5600万.

RocketMQ特点

经历过双十一考验，性能没啥太大的问题。
持久化（零拷贝/磁盘顺序存储/页缓存）
底层通讯是Netty

  1. 亿级消息的堆积能力，单个队列中的百万级消息的累积容量。
    2. 高可用性：Broker服务器支持多Master多Slave的同步双写以及Master多Slave的异步复制模式，其中同步双写可保证消息不丢失。
    3. 高可靠性：生产者将消息发送到Broker端有三种方式，同步、异步和单向，其中同步和异步都可以保证消息成功的成功发送。Broker在对于消息刷盘有两种策略：同步刷盘和异步刷盘，其中同步刷盘可以保证消息成功的存储到磁盘中。消费者的消费模式也有集群消费和广播消费两种，默认集群消费，如果集群模式中消费者挂了，一个组里的其他消费者会接替其消费。综上所述，是高可靠的。
    4. 支持分布式事务消息：这里是采用半消息确认和消息回查机制来保证分布式事务消息的
    5. 支持消息过滤：建议采用消费者业务端的tag过滤
    6. 支持顺序消息：消息在Broker中是采用队列的FIFO模式存储的，也就是发送是顺序的，只要保证消费的顺序性即可。
    7. 支持定时消息和延迟消息：Broker中由定时消息的机制，消息发送到Broker中，不会立即被Consumer消费，会等到一定的时间才被消费。延迟消息也是一样，延迟一定时间之后才会被Consumer消费。

你知道RocketMQ为什么速度快吗？

是因为使用了顺序存储、Page Cache和异步刷盘。

  1. 我们在写入commitlog的时候是顺序写入的，这样比随机写入的性能就会提高很多
  2. 写入commitlog的时候并不是直接写入磁盘，而是先写入操作系统的PageCache
  3. 最后由操作系统异步将缓存中的数据刷到磁盘

RocketMQ的核⼼组件

RocketMQ主要由NameServer、Broker、Producer以及Consumer四部分构成。
image.png

NameServer

NameServer（类似Zookeeper的一个服务寻址和协调）

1.主要负责对于源数据的管理，包括了对于Topic和路由信息的管理,，所以 name server 集群可以简单理解为无状态（实际情况下可能存在每个 nameserver 机器上的数据有短暂的不一致现象，但是通过定时更新，大部分情况下都是一致的）

2.NameServer是⼀个功能⻬全的服务器，其⻆⾊类似Dubbo中的Zookeeper，但NameServer与Zookeeper相⽐更轻量。主要是因为每个NameServer节点互相之间是独⽴的，没有任何信息交互。

为什么用NameServer?原因是因为Zookeeper的大量的写性能不是很高的.

NameServer服务充当路由消息的提供者。Broker Server会在启动时向所有的Name Server注册自己的服务信息，并且后续通过心跳请求的方式保证这个服务信息的实时性。broker 会定时的发送路由信息到 nameserver 中的每一个机器，来进行更新，节点之间无任何信息同步
生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。多个Namesrv实例组成集群，但相互独立，没有信息交换。

这种特性也就意味着NameServer中任意的节点挂了，只要有一台服务节点正常，整个路由服务就不会有影响。当然，这里不考虑节点的负载情况。

Producer

消息⽣产者，负责产⽣消息，⼀般由业务系统负责产⽣消息。
Producer由⽤户进⾏分布式部署，消息由Producer通过多种负载均衡模式发送到Broker集群，发送低延时，⽀持快速失败。

一个消息生产者会把业务应用系统里产生的消息发送到broker服务器。RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异步方式均需要Broker返回确认信息，单向发送不需要。

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

Broker

消息中转⻆⾊，负责存储消息，转发消息。

  • Broker是具体提供业务的服务器，单个Broker节点与所有的NameServer节点保持⻓连接及⼼跳，并会定时将Topic信息注册到NameServer，顺带⼀提底层的通信和连接都是基于Netty实现的。
  • Broker负责消息存储，以Topic为纬度⽀持轻量级的队列，单机可以⽀撑上万队列规模，⽀持消息推拉模型.
  • 官⽹上有数据显示：具有上亿级消息堆积能⼒，同时可严格保证消息的有序性.

broker 集群：一个集群有一个统一的名字，即 brokerClusterName，默认是 DefaultCluster。一个集群下有多个 master，每个 master 下有多个 slave。master 和 slave 算是一组，拥有相同的 brokerName, 不同的 brokerId，master 的 brokerId 是 0，而 slave 则是大于 0 的值。master 和 slave 之间可以进行同步复制或者是异步复制。

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

Broker Server是RocketMQ真正的业务核心，包含了多个重要的子模块：

  • Remoting Module：整个Broker的实体，负责处理来自clients端的请求。
  • Client Manager：负责管理客户端(Producer/Consumer)和维护Consumer的Topic订阅信息
  • Store Service：提供方便简单的API接口处理消息存储到物理硬盘和查询功能。
  • HA Service：高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。
  • Index Service：根据特定的Message key对投递到Broker的消息进行索引服务，以提供消息的快速查询。

而Broker Server要保证高可用需要搭建主从集群架构。RocketMQ中有两种Broker架构模式：

  • 普通集群：

这种集群模式下会给每个节点分配一个固定的角色，master负责响应客户端的请求，并存储消息。slave则只负责对master的消息进行同步保存，并响应部分客户端的读请求。消息同步方式分为同步同步和异步同步。

这种集群模式下各个节点的角色无法进行切换，也就是说，master节点挂了，这一组Broker就不可用了。

  • Dledger高可用集群：

Dledger是RocketMQ自4.5版本引入的实现高可用集群的一项技术。这个模式下的集群会随机选出一个节点作为master，而当master节点挂了后，会从slave中自动选出一个节点升级成为master。

Dledger技术做的事情：

Dledger的关键部分是在他的节点选举上。Dledger是使用Raft算法来进行节点选举的。这里简单介绍下Raft算法的选举过程:

首先：每个节点有三个状态，Leader，follower和candidate(候选人)。正常运行的情况下，集群中会有一个leader，其他都是follower，follower只响应Leader和Candidate的请求，而客户端的请求全部由Leader处理，即使有客户端请求到了一个follower，也会将请求转发到leader。

集群刚启动时，每个节点都是follower状态，之后集群内部会发送一个timeout信号，所有follower就转成candidate去拉取选票，获得大多数选票的节点选为leader，其他候选人转为follower。如果一个timeout信号发出时，没有选出leader，将会重新开始一次新的选举。而Leader节点会往其他节点发送心跳信号，确认他的leader状态。

— 然后会启动定时器，如果在指定时间内没有收到Leader的心跳，就会转为Candidate状态，然后向其他成员发起投票请求，如果收到半数以上成员的投票，则Candidate会晋升为Leader。然后leader也有可能会退化成follower。

然后，在Raft协议中，会将时间分为一些任意时间长度的时间片段，叫做term。term会使用一个全局唯一，连续递增的编号作为标识，也就是起到了一个逻辑时钟的作用。

image.png

在每一个term时间片里，都会进行新的选举，每一个Candidate都会努力争取成为leader。获得票数最多的节点就会被选举为Leader。被选为Leader的这个节点，在一个term时间片里就会保持leader状态。这样，就会保证在同一时间段内，集群中只会有一个Leader。在某些情况下，选票可能会被各个节点瓜分，形成不了多数派，那这个term可能直到结束都没有leader，直到下一个term再重新发起选举，这也就没有了Zookeeper中的脑裂问题。而在每次重新选举的过程中， leader也有可能会退化成为follower。也就是说，在这个集群中， leader节点是会不断变化的。

然后，每次选举的过程中，每个节点都会存储当前term编号，并在节点之间进行交流时，都会带上自己的term编号。如果一个节点发现他的编号比另外一个小，那么他就会将自己的编号更新为较大的那一个。而如果leader或者candidate发现自己的编号不是最新的，他就会自动转成follower。如果接收到的请求term编号小于自己的编号，term将会拒绝执行。

在选举过程中，Raft协议会通过心跳机制发起leader选举。节点都是从follower状态开始的，如果收到了来自leader或者candidate的心跳RPC请求，那他就会保持follower状态，避免争抢成为candidate。而leader会往其他节点发送心跳信号，来确认自己的地位。如果follower一段时间(两个timeout信号)内没有收到Leader的心跳信号，他就会认为leader挂了，发起新一轮选举。

选举开始后，每个follower会增加自己当前的term，并将自己转为candidate。然后向其他节点发起投票请求，请求时会带上自己的编号和term，也就是说都会默认投自己一票。之后candidate状态可能会发生以下三种变化：

  • 赢得选举，成为leader： 如果它在一个term内收到了大多数的选票，将会在接下的剩余term时间内称为leader，然后就可以通过发送心跳确立自己的地位。(每一个server在一个term内只能投一张选票，并且按照先到先得的原则投出)
  • 其他节点成为leader： 在等待投票时，可能会收到其他server发出心跳信号，说明其他leader已经产生了。这时通过比较自己的term编号和RPC过来的term编号，如果比对方大，说明leader的term过期了，就会拒绝该RPC,并继续保持候选人身份; 如果对方编号不比自己小,则承认对方的地位,转为follower。
  • 选票被瓜分,选举失败: 如果没有candidate获取大多数选票, 则没有leader产生, candidate们等待超时后发起另一轮选举. 为了防止下一次选票还被瓜分,必须采取一些额外的措施, raft采用随机election timeout(随机休眠时间)的机制防止选票被持续瓜分。通过将timeout随机设为一段区间上的某个值, 因此很大概率会有某个candidate率先超时然后赢得大部分选票。

所以以三个节点的集群为例，选举过程会是这样的：

  1. 集群启动时，三个节点都是follower，发起投票后，三个节点都会给自己投票。这样一轮投票下来，三个节点的term都是1，是一样的，这样是选举不出Leader的。
  2. 当一轮投票选举不出Leader后，三个节点会进入随机休眠，例如A休眠1秒，B休眠3秒，C休眠2秒。
  3. 一秒后，A节点醒来，会把自己的term加一票，投为2。然后2秒时，C节点醒来，发现A的term已经是2，比自己的1大，就会承认A是Leader，把自己的term也更新为2。实际上这个时候，A已经获得了集群中的多数票，2票，A就会被选举成Leader。这样，一般经过很短的几轮选举，就会选举出一个Leader来。
  4. 到3秒时，B节点会醒来，他也同样会承认A的term最大，他是Leader，自己的term也会更新为2。这样集群中的所有Candidate就都确定成了leader和follower.
  5. 然后在一个任期内，A会不断发心跳给另外两个节点。当A挂了后，另外的节点没有收到A的心跳，就会都转化成Candidate状态，重新发起选举。

Dledger还会采用Raft协议进行多副本的消息同步：

简单来说，数据同步会通过两个阶段，一个是uncommitted阶段，一个是commited阶段。

Leader Broker上的Dledger收到一条数据后，会标记为uncommitted状态，然后他通过自己的DledgerServer组件把这个uncommitted数据发给Follower Broker的DledgerServer组件。

接着Follower Broker的DledgerServer收到uncommitted消息之后，必须返回一个ack给Leader Broker的Dledger。然后如果Leader Broker收到超过半数的Follower Broker返回的ack之后，就会把消息标记为committed状态。

再接下来， Leader Broker上的DledgerServer就会发送committed消息给Follower Broker上的DledgerServer，让他们把消息也标记为committed状态。这样，就基于Raft协议完成了两阶段的数据同步。

最后，关于Dledger以及Raft协议的更底层的详细资料，后续会有一个分布式一致性协议的专题，将会结合其他分布式一致性算法做统一讲解，这里就不深入展开了。

Consumer

消息消费者，负责消费消息，⼀般是后台系统负责异步消费。一个消息消费者会从Broker服务器拉取消息、并将其提供给应用程序。从用户应用的角度而言提供了两种消费形式：拉取式消费、推动式消费。

  • Consumer也由⽤户部署，⽀持PUSH和PULL两种消费模式，⽀持集群消费和⼴播消息，提供实时的消息订阅机制。

  • 拉取式消费的应用通常主动调用Consumer的拉消息方法从Broker服务器拉消息、主动权由应用控制。一旦获取了批量消息，应用就会启动消费过程。
  • 推动式消费模式下Broker收到数据后会主动推送给消费端，该消费模式一般实时性较高。

消费者同样会把同一类Consumer组成一个集合，叫做消费者组，这类Consumer通常消费同一类消息且消费逻辑一致。消费者组使得在消息消费方面，实现负载均衡和容错的目标变得非常容易。要注意的是，消费者组的消费者实例必须订阅完全相同的Topic。RocketMQ 支持两种消息模式：集群消费（Clustering）和广播消费（Broadcasting）。

  • 集群消费模式下,相同Consumer Group的每个Consumer实例平均分摊消息。
  • 广播消费模式下，相同Consumer Group的每个Consumer实例都接收全量的消息。

RocketMQ实现原理

RocketMQ由NameServer注册中心集群、Producer生产者集群、Consumer消费者集群和若干Broker（RocketMQ进程）组成，它的架构原理是这样的：

  1. Broker在启动的时候去向所有的NameServer注册，并保持长连接，每30s发送一次心跳
  2. Producer在发送消息的时候从NameServer获取Broker服务器地址，根据负载均衡算法选择一台服务器来发送消息
  3. Conusmer消费消息的时候同样从NameServer获取Broker地址，然后主动拉取消息来消费

RocketMQ简介 - 图3
image.png

RocketMQ的消息领域模型

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

主要分为Message、Topic、Queue、Offset以及Group这⼏部分。

image.png

1）Topic
Topic表示消息的第⼀级类型，⽐如⼀个电商系统的消息可以分为：交易消息、物流消息等。⼀条消息必须有⼀个Topic。
最细粒度的订阅单位，⼀个Group可以订阅多个Topic的消息。

同一个Topic下的数据，会分片保存到不同的Broker上，而每一个分片单位，就叫做MessageQueue。MessageQueue是生产者发送消息与消费者消费消息的最小单位。

2）Tag
Tag表示消息的第⼆级类型，⽐如交易消息⼜可以分为：交易创建消息，交易完成消息等。RocketMQ提供2级消息分类，⽅便灵活控制。

3)Group
组，⼀个组可以订阅多个Topic。

4）Message Queue
消息的物理管理单位。⼀个Topic下可以有多个Queue，Queue的引⼊使得消息的存储可以分布式集群化，具有了⽔平扩展能⼒。

在RocketMQ 中，所有消息队列都是持久化，⻓度⽆限的数据结构，所谓⻓度⽆限是指队列中的每个存储单元都是定⻓，访问其中的存储单元使⽤Offset 来访问，offset 为 java long 类型，64 位，理论上在 100年内不会溢出，所以认为是⻓度⽆限。
也可以认为 Message Queue 是⼀个⻓度⽆限的数组，Offset 就是下标。

5) 消息

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

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

为什么RocketMQ不使用Zookeeper作为注册中心呢？

我认为有以下几个点是不使用zookeeper的原因：

  1. 根据CAP理论，同时最多只能满足两个点，而zookeeper满足的是CP，也就是说zookeeper并不能保证服务的可用性，zookeeper在进行选举的时候，整个选举的时间太长，期间整个集群都处于不可用的状态，而这对于一个注册中心来说肯定是不能接受的，作为服务发现来说就应该是为可用性而设计。
  2. 基于性能的考虑，NameServer本身的实现非常轻量，而且可以通过增加机器的方式水平扩展，增加集群的抗压能力，而zookeeper的写是不可扩展的，而zookeeper要解决这个问题只能通过划分领域，划分多个zookeeper集群来解决，首先操作起来太复杂，其次这样还是又违反了CAP中的A的设计，导致服务之间是不连通的。
  3. 持久化的机制来带的问题，ZooKeeper 的 ZAB 协议对每一个写请求，会在每个 ZooKeeper 节点上保持写一个事务日志，同时再加上定期的将内存数据镜像（Snapshot）到磁盘来保证数据的一致性和持久性，而对于一个简单的服务发现的场景来说，这其实没有太大的必要，这个实现方案太重了。而且本身存储的数据应该是高度定制化的。
  4. 消息发送应该弱依赖注册中心，而RocketMQ的设计理念也正是基于此，生产者在第一次发送消息的时候从NameServer获取到Broker地址后缓存到本地，如果NameServer整个集群不可用，短时间内对于生产者和消费者并不会产生太大影响。

Broker是怎么保存数据

RocketMQ主要的存储文件包括commitlog文件、consumequeue文件、indexfile文件。

Broker在收到消息之后，会把消息保存到commitlog的文件当中，而同时在分布式的存储当中，每个broker都会保存一部分topic的数据，同时，每个topic对应的messagequeue下都会生成consumequeue文件用于保存commitlog的物理位置偏移量offset，indexfile中会保存key和offset的对应关系。

image.png

CommitLog文件保存于${Rocket_Home}/store/commitlog目录中，从图中我们可以明显看出来文件名的偏移量，每个文件默认1G，写满后自动生成一个新的文件。

image.png

由于同一个topic的消息并不是连续的存储在commitlog中，消费者如果直接从commitlog获取消息效率非常低，所以通过consumequeue保存commitlog中消息的偏移量的物理地址，这样消费者在消费的时候先从consumequeue中根据偏移量定位到具体的commitlog物理文件，然后根据一定的规则（offset和文件大小取模）在commitlog中快速定位。

image.png

如果 Broker 宕了，NameServer 是怎么感知到的？

Broker 会定时（30s）向 NameServer 发送心跳
然后 NameServer 会定时（10s）运行一个任务，去检查一下各个 Broker 的最近一次心跳时间，如果某个 Broker 超过 120s 都没发送心跳了，那么就认为这个 Broker 已经挂掉了。

RocketMQ 对生产者有哪些容错机制

基于 Dledger 实现 RocketMQ 高可用自动切换
RocketMQ 4.5 之后支持了一种叫做 Dledger 机制，基于 Raft 协议实现的一个机制。
我们可以让一个 Master Broker 对应多个 Slave Broker， 一旦 Master Broker 宕机了，在多个 Slave 中通过 Dledger 技术将一个 Slave Broker 选为新的 Master Broker 对外提供服务。
在生产环境中可以是用 Dledger 机制实现自动故障切换，只要 10 秒或者几十秒的时间就可以完成。

Producer 发送消息到 Broker，是随机选择一个 Broker 还是有一定的规则？

一般是负载均衡做随机选择，但也可以走其他策略，比如根据某个字段来 hash

Master和Slave之间是怎么同步数据的呢？

而消息在master和slave之间的同步是根据raft协议来进行的：

  1. 在broker收到消息后，会被标记为uncommitted状态
  2. 然后会把消息发送给所有的slave
  3. slave在收到消息之后返回ack响应给master
  4. master在收到超过半数的ack之后，把消息标记为committed
  5. 发送committed消息给所有slave，slave也修改状态为committed