消息队列系列4 - 原理初探之RocketMQ
RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件，前身是MetaQ，是阿里参考Kafka特点研发的一个队列模型的消息中间件，后开源给apache基金会成为了apache的顶级开源项目，具有高性能、高可靠、高实时、分布式特点。
先提前祝大家五一快乐！！！![]()
![]()
![]()
因为个人学习重点调整，消息队列目前只先发这4篇文章，后面会选择一种消息队列，重点介绍环境搭建和使用姿势上，也就是消息队列的第5篇文章，预计3个月之内会补上去，期待后续...
4.1 基本概念
先对常用的词汇有个基本认识，相关词汇后面会再详细介绍：
NameServer：一个功能齐全的服务器，其角色类似Dubbo中的Zookeeper。 Producer：消息生产者，负责产生消息，一般由业务系统负责产生消息。 Consumer：消息消费者，负责消费消息，一般是后台系统负责异步消费。 Broker：消息中转角色，负责存储消息，转发消息。 Message：消息，一条消息必须有一个主题（Topic），主题可以看做是你的信件要邮寄的地址。（一条消息也可以拥有一个可选的标签（Tag）和额处的键值对，它们可以用于设置一个业务 Key 并在 Broker 上查找此消息以便在开发期间查找问题。） Topic：主题，可以看做消息的归类，它是消息的第一级类型。（比如一个电商系统可以分为：交易消息、物流消息等，一条消息必须有一个 Topic 。Topic 与生产者和消费者的关系非常松散，一个 Topic 可以有0个、1个、多个生产者向其发送消息，一个生产者也可以同时向不同的 Topic 发送消息。一个 Topic 也可以被 0个、1个、多个消费者订阅。） Tag：子主题，它是消息的第二级类型，用于为用户提供额外的灵活性。（使用标签，同一业务模块不同目的的消息就可以用相同 Topic 而不同的 Tag 来标识。比如交易消息又可以分为：交易创建消息、交易完成消息等，一条消息可以没有 Tag 。标签有助于保持您的代码干净和连贯，并且还可以为 RocketMQ 提供的查询系统提供帮助。） Group：分组，一个组可以订阅多个Topic。（分为ProducerGroup，ConsumerGroup，代表某一类的生产者和消费者，一般来说同一个服务可以作为Group，同一个Group一般来说发送和消费的消息都是一样的。） Producer Group：生产者组，代表某一类的生产者，比如我们有多个秒杀系统作为生产者，这多个合在一起就是一个 Producer Group 生产者组，它们一般生产相同的消息。 Consumer Group：消费者组，代表某一类的消费者，比如我们有多个短信系统作为消费者，这多个合在一起就是一个 Consumer Group 消费者组，它们一般消费相同的消息。 Queue：队列，在Kafka中叫Partition。（每个Queue内部是有序的，在RocketMQ中分为读和写两种队列，一般来说读写队列数量一致，如果不一致就会出现很多问题。） Message Queue：消息队列，主题被划分为一个或多个子主题，即消息队列。（一个 Topic 下可以设置多个消息队列，发送消息时执行该消息的 Topic ，RocketMQ 会轮询该 Topic 下的所有队列将消息发出去。消息的物理管理单位。一个Topic下可以有多个Queue，Queue的引入使得消息的存储可以分布式集群化，具有了水平扩展能力。）
4.2 消息模型
RockerMQ 中的消息模型就是按照主题模型所实现的，在主题模型中，消息的生产者称为发布者（Publisher），消息的消费者称为订阅者（Subscriber），存放消息的容器称为主题（Topic）。RocketMQ 中的主题模型到底是如何实现的呢？
我们可以看到在整个图中有 Producer Group、Topic、Consumer Group 三个角色，你可以看到图中生产者组中的生产者会向主题发送消息，而主题中存在多个队列，生产者每次生产消息之后是指定主题中的某个队列发送消息的。
每个主题中都有多个队列（这里还不涉及到 Broker），集群消费模式下，一个消费者集群多台机器共同消费一个 topic 的多个队列，一个队列只会被一个消费者消费。如果某个消费者挂掉，分组内其它消费者会接替挂掉的消费者继续消费。就像上图中 Consumer1 和 Consumer2 分别对应着两个队列，而 Consuer3 是没有队列对应的，所以一般来讲要控制消费者组中的消费者个数和主题中队列个数相同。这个简直和kafak一毛一样啊！
当然也可以消费者个数小于队列个数，只不过不太建议。如下图：
每个消费组在每个队列上维护一个消费位置，为什么呢？因为我们刚刚画的仅仅是一个消费者组，我们知道在发布订阅模式中一般会涉及到多个消费者组，而每个消费者组在每个队列中的消费位置都是不同的。如果此时有多个消费者组，那么消息被一个消费者组消费完之后是不会删除的（因为其它消费者组也需要呀），它仅仅是为每个消费者组维护一个消费位移（offset），每次消费者组消费完会返回一个成功的响应，然后队列再把维护的消费位移加一，这样就不会出现刚刚消费过的消息再一次被消费了。
可能你还有一个问题，为什么一个主题中需要维护多个队列？答案是提高并发能力。的确，每个主题中只存在一个队列也是可行的。你想一下，如果每个主题中只存在一个队列，这个队列中也维护着每个消费者组的消费位置，这样也可以做到发布订阅模式。如下图：
但是，这样我生产者是不是只能向一个队列发送消息？又因为需要维护消费位置所以一个队列只能对应一个消费者组中的消费者，这样是不是其他的 Consumer 就没有用武之地了？从这两个角度来讲，并发度一下子就小了很多。
所以总结来说，RocketMQ 通过使用在一个 Topic 中配置多个队列，并且每个队列维护每个消费者组的消费位置，实现了主题模式/发布订阅模式。
4.3 系统架构
讲完了消息模型，我们理解起 RocketMQ 的技术架构起来就容易多了。RocketMQ 技术架构中有四大角色 NameServer、Broker、Producer、Consumer。这4大角色，已经在基本概念中简单解释过，对于相关词汇，这里再重点解释一下。
Broker：主要负责消息的存储、投递和查询以及服务高可用保证。说白了就是消息队列服务器嘛，生产者生产消息到 Broker，消费者从 Broker 拉取消息并消费。这里，我还得普及一下关于 Broker、Topic 和队列的关系。上面我讲解了 Topic 和队列的关系——一个 Topic 中存在多个队列，那么这个 Topic 和队列存放在哪呢？一个 Topic 分布在多个 Broker 上，一个 Broker 可以配置多个 Topic，它们是多对多的关系。如果某个 Topic 消息量很大，应该给它多配置几个队列，并且尽量多分布在不同 Broker 上，以减轻某个 Broker 的压力。Topic 消息量都比较均匀的情况下，如果某个 broker 上的队列越多，则该 broker 压力越大。 NameServer：不知道你们有没有接触过 ZooKeeper 和 Spring Cloud 中的 Eureka，它其实也是一个注册中心，主要提供两个功能：Broker 管理和路由信息管理。说白了就是 Broker 会将自己的信息注册到 NameServer 中，此时 NameServer 就存放了很多 Broker 的信息（Broker的路由表），消费者和生产者就从 NameServer 中获取路由表然后照着路由表的信息和对应的 Broker 进行通信（生产者和消费者定期会向 NameServer 去查询相关的 Broker 的信息）。 Producer：消息发布的角色，支持分布式集群方式部署。 Consumer：消息消费的角色，支持分布式集群方式部署。支持以 push 推，pull 拉两种模式对消息进行消费，同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制。
听完了上面的解释你可能会觉得，这玩意好简单。不就是这样的么？
嗯？你可能会发现一个问题，这老家伙 NameServer 干啥用的，这不多余吗？直接 Producer、Consumer 和 Broker 直接进行生产消息，消费消息不就好了么？但是，我们上文提到过 Broker 是需要保证高可用的，如果整个系统仅仅靠着一个 Broker 来维持的话，那么这个 Broker 的压力会不会很大？所以我们需要使用多个 Broker 来保证负载均衡。如果说，我们的消费者和生产者直接和多个 Broker 相连，那么当 Broker 修改的时候必定会牵连着每个生产者和消费者，这样就会产生耦合问题，而 NameServer 注册中心就是用来解决这个问题的。
当然，RocketMQ 中的技术架构肯定不止前面那么简单，因为上面图中的四个角色都是需要做集群的。我给出一张官网的架构图，大家尝试理解一下。
其实和我们最开始画的那张乞丐版的架构图也没什么区别，主要是一些细节上的差别，听我细细道来。
第一、我们的 Broker 做了集群并且还进行了主从部署，由于消息分布在各个 Broker 上，一旦某个 Broker 宕机，则该 Broker 上的消息读写都会受到影响。所以 RocketMQ 提供了 master/slave 的结构，salve 定时从 master 同步数据（同步刷盘或者异步刷盘），如果 master 宕机，则 slave 提供消费服务，但是不能写入消息（后面我还会提到）。 第二、为了保证 HA，我们的 NameServer 也做了集群部署，但是请注意它是去中心化的。也就意味着它没有主节点，你可以很明显地看出 NameServer 的所有节点是没有进行 Info Replicate 的，在 RocketMQ 中是通过单个 Broker 和所有 NameServer 保持长连接，并且在每隔 30 秒 Broker 会向所有 Nameserver 发送心跳，心跳包含了自身的 Topic 配置信息，这个步骤就对应这上面的 Routing Info。 第三、在生产者需要向 Broker 发送消息的时候，需要先从 NameServer 获取关于 Broker 的路由信息，然后通过轮询的方法去向每个队列中生产数据以达到负载均衡的效果。 第四、消费者通过 NameServer 获取所有 Broker 的路由信息后，向 Broker 发送 Pull 请求来获取消息数据。Consumer 可以以两种模式启动—— 广播（Broadcast）和集群（Cluster）。广播模式下，一条消息会发送给同一个消费组中的所有消费者，集群模式下消息只会发送给一个消费者。
4.4 高级特性&常见问题
4.4.1 顺序消费
在上面的技术架构介绍中，我们已经知道了 RocketMQ 在主题上是无序的、它只有在队列层面才是保证有序的。这又扯到两个概念——普通顺序和严格顺序。所谓普通顺序是指消费者通过同一个消费队列收到的消息是有顺序的，不同消息队列收到的消息则可能是无顺序的。普通顺序消息在 Broker 重启情况下不会保证消息顺序性（短暂时间）。
所谓严格顺序是指消费者收到的所有消息均是有顺序的。严格顺序消息即使在异常情况下也会保证消息的顺序性。但是，严格顺序看起来虽好，实现它可会付出巨大的代价。如果你使用严格顺序模式，Broker 集群中只要有一台机器不可用，则整个集群都不可用。你还用啥？现在主要场景也就在 binlog 同步。一般而言，我们的 MQ 都是能容忍短暂的乱序，所以推荐使用普通顺序模式。（这个严格顺序，感觉没太懂，后面再查一下相关资料。。。）
那么，我们现在使用了普通顺序模式，我们从上面学习知道了在 Producer 生产消息的时候会进行轮询（取决你的负载均衡策略）来向同一主题的不同消息队列发送消息。那么如果此时我有几个消息分别是同一个订单的创建、支付、发货，在轮询的策略下这三个消息会被发送到不同队列，因为在不同的队列此时就无法使用 RocketMQ 带来的队列有序特性来保证消息有序性了。
那么，怎么解决呢？其实很简单，我们需要处理的仅仅是将同一语义下的消息放入同一个队列（比如这里是同一个订单），那我们就可以使用 Hash 取模法来保证同一个订单在同一个队列中就行了。
4.4.2 重复消费
就两个字——幂等。在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。比如说，这个时候我们有一个订单的处理积分的系统，每当来一个消息的时候它就负责为创建这个订单的用户的积分加上相应的数值。可是有一次，消息队列发送给订单系统 FrancisQ 的订单信息，其要求是给 FrancisQ 的积分加上 500。但是积分系统在收到 FrancisQ 的订单信息处理完成之后返回给消息队列处理成功的信息的时候出现了网络波动（当然还有很多种情况，比如 Broker 意外重启等等），这条回应没有发送成功。
那么，消息队列没收到积分系统的回应会不会尝试重发这个消息？问题就来了，我再发这个消息，万一它又给 FrancisQ 的账户加上 500 积分怎么办呢？所以我们需要给我们的消费者实现幂等，也就是对同一个消息的处理结果，执行多少次都不变。
那么如何给业务实现幂等呢？这个还是需要结合具体的业务的。你可以使用写入 Redis 来保证，因为 Redis 的 key 和 value 就是天然支持幂等的。当然还有使用数据库插入法，基于数据库的唯一键来保证重复数据不会被插入多条。不过最主要的还是需要根据特定场景使用特定的解决方案，你要知道你的消息消费是否是完全不可重复消费还是可以忍受重复消费的，然后再选择强校验和弱校验的方式。毕竟在 CS 领域还是很少有技术银弹的说法。
简单了来说，幂等的校验，还是需要业务方来支持，因为你解决不了网络抖动问题哈~~
4.4.3 分布式事务
如何解释分布式事务呢？事务大家都知道吧？要么都执行要么都不执行。在同一个系统中我们可以轻松地实现事务，但是在分布式架构中，我们有很多服务是部署在不同系统之间的，而不同服务之间又需要进行调用。比如此时我下订单然后增加积分，如果保证不了分布式事务的话，就会出现A系统下了订单，但是B系统增加积分失败或者A系统没有下订单，B系统却增加了积分。前者对用户不友好，后者对运营商不利，这是我们都不愿意见到的。那么，如何去解决这个问题呢？
如今比较常见的分布式事务实现有 2PC、TCC 和事务消息（half 半消息机制）。每一种实现都有其特定的使用场景，但是也有各自的问题，都不是完美的解决方案。在 RocketMQ 中使用的是事务消息加上事务反查机制来解决分布式事务问题的。
下面是上图的执行流程：
A服务先发送个Half Message给Brock端，消息中携带 B服务 即将要+100元的信息。 当A服务知道Half Message发送成功后，那么开始第3步执行本地事务。 执行本地事务(会有三种情况1、执行成功。2、执行失败。3、网络等原因导致没有响应) 如果本地事务成功，那么Product像Brock服务器发送Commit,这样B服务就可以消费该message。 如果本地事务失败，那么Product像Brock服务器发送Rollback,那么就会直接删除上面这条半消息。 如果因为网络等原因迟迟没有返回失败还是成功，那么会执行RocketMQ的回调接口,来进行事务的回查。
4.4.4 消息堆积
消息中间件的主要功能是异步解耦，还有个重要功能是挡住前端的数据洪峰，保证后端系统的稳定性，这就要求消息中间件具有一定的消息堆积能力，消息堆积分以下两种情况：
消息堆积在内存Buffer，一旦超过内存Buffer，可以根据一定的丢弃策略来丢弃消息，如CORBA Notification规范中描述。适合能容忍丢弃消息的业务，这种情况消息的堆积能力主要在于内存Buffer大小，而且消息堆积后，性能下降不会太大，因为内存中数据多少对于对外提供的访问能力影响有限。 消息堆积到持久化存储系统中，例如DB，KV存储，文件记录形式。当消息不能在内存Cache命中时，要不可避免的访问磁盘，会产生大量读IO，读IO的吞吐量直接决定了消息堆积后的访问能力。
评估消息堆积能力主要有以下四点：
消息能堆积多少条，多少字节？即消息的堆积容量。 消息堆积后，发消息的吞吐量大小，是否会受堆积影响？ 消息堆积后，正常消费的Consumer是否会受影响？ 消息堆积后，访问堆积在磁盘的消息时，吞吐量有多大？
简单来说，RocketMQ支持大量消息堆积，消息会存在内存，超出内存的消息会持久化到磁盘中。
4.4.5 定时消息
定时消息是指消息发到Broker后，不能立刻被Consumer消费，要到特定的时间点或者等待特定的时间后才能被消费。如果要支持任意的时间精度，在Broker层面，必须要做消息排序，如果再涉及到持久化，那么消息排序要不可避免的产生巨大性能开销。RocketMQ支持定时消息，但是不支持任意时间精度，支持特定的level，例如定时5s，10s，1m等。
简单来说，RocketMQ支持定时消息，但是只支持固定时间，不支持任意精度时间。
4.4.6 回溯消费
4.4.7.1 同步刷盘和异步刷盘
上面我讲了那么多的 RocketMQ 的架构和设计原理，你有没有好奇，在 Topic 中的队列是以什么样的形式存在的？队列中的消息又是如何进行存储持久化的呢？我在上文中提到的同步刷盘和异步刷盘又是什么呢？它们会给持久化带来什么样的影响呢？下面我将给你们一一解释。
如上图所示，在同步刷盘中需要等待一个刷盘成功的 ACK，同步刷盘对 MQ 消息可靠性来说是一种不错的保障，但是性能上会有较大影响，一般地适用于金融等特定业务场景。而异步刷盘往往是开启一个线程去异步地执行刷盘操作。消息刷盘采用后台异步线程提交的方式进行，降低了读写延迟，提高了 MQ 的性能和吞吐量，一般适用于如发验证码等对于消息保证要求不太高的业务场景。一般地，异步刷盘只有在 Broker 意外宕机的时候会丢失部分数据，你可以设置 Broker 的参数 FlushDiskType 来调整你的刷盘策略（ASYNC_FLUSH 或者 SYNC_FLUSH）。
简单来说，同步刷盘是刷盘后请求再返回，异步刷盘是直接返回请求，再去慢慢刷盘，可能会导致数据丢失。
4.4.7.2 同步复制和异步复制
上面的同步刷盘和异步刷盘是在单个结点层面的，而同步复制和异步复制主要是指的 Borker 主从模式下，主节点返回消息给客户端的时候是否需要同步从节点。
同步复制：也叫 “同步双写”，也就是说，只有消息同步双写到主从结点上时才返回写入成功。 异步复制：消息写入主节点之后就直接返回写入成功。
扩展知识1：在单主从架构中，如果一个主节点挂掉了，那么也就意味着整个系统不能再生产了。那么这个可用性的问题能否解决呢？一个主从不行那就多个主从的呗，别忘了在我们最初的架构图中，每个 Topic 是分布在不同 Broker 中的。但是这种复制方式同样也会带来一个问题，那就是无法保证严格顺序。在上文中我们提到了如何保证的消息顺序性是通过将一个语义的消息发送在同一个队列中，使用 Topic 下的队列来保证顺序性的。如果此时我们主节点 A 负责的是订单 A 的一系列语义消息，然后它挂了，这样其他节点是无法代替主节点A的，如果我们任意节点都可以存入任何消息，那就没有顺序性可言了。（这点我并不认同，我理解主从的对列信息应该是一样的，我从主节点读到哪里，如果主节点挂掉，应该是可以到从结点去读取的，如果不能这样，搞个主从就完全没有意义了。因为主从的信息是一样的，对队列的顺序是有影响的，我不可能把不同的信息，搞两个队列，分别放到主从机器。）
扩展知识2：在 RocketMQ 中采用了 Dledger 解决主从数据同步问题。他要求在写入消息的时候，要求至少消息复制到半数以上的节点之后，才给客⼾端返回写⼊成功，并且它是⽀持通过选举来动态切换主节点的。这里我就不展开说明了，读者可以自己去了解。也不是说 Dledger 是个完美的方案，至少在 Dledger 选举过程中是无法提供服务的，而且他必须要使用三个节点或以上，如果多数节点同时挂掉他也是无法保证可用性的，而且要求消息复制板书以上节点的效率和直接异步复制还是有一定的差距的。
这个机制，感觉就像大众化的版本，基本思路都一样，为了保证数据可用性，我还是推荐同步复制，当大多数节点复制成功，就认为复制完毕，和ETCD的Raft协议的日志同步原理一样。
4.4.8 容错机制
在实际使用RocketMQ的时候我们并不能保证每次发送的消息都刚好能被消费者一次性正常消费成功，可能会存在需要多次消费才能成功或者一直消费失败的情况，那作为发送者该做如何处理呢？
RocketMQ提供了ack机制，以保证消息能够被正常消费。发送者为了保证消息肯定消费成功，只有使用方明确表示消费成功，RocketMQ才会认为消息消费成功。中途断电，抛出异常等都不会认为成功——即都会重新投递。当然，如果消费者不告知发送者我这边消费信息异常，那么发送者是不会知道的，所以消费者在设置监听的时候需要给个回调。
为了保证消息是肯定被至少消费成功一次，RocketMQ会把这批消息重发回Broker（topic不是原topic而是这个消费租的RETRY topic），在延迟的某个时间点（默认是10秒，业务可设置）后，再次投递到这个ConsumerGroup。而如果一直这样重复消费都持续失败到一定次数（默认16次），就会投递到DLQ死信队列。应用可以监控死信队列来做人工干预。
简单来说，通过ACK保证消息一定能正常消费，对于异常消息，会重新放回Broker，但是这样就会打乱消息的顺序，所以容错机制和消息严格顺序，鱼和熊掌不可兼得。
4.5 特性分析
这里才是内容的重点，不仅需要知道RocketMQ的特性，还需要知道支持这些特性的原因：
消息路由（不支持）：RocketMQ在处理消息之前是不允许消费者过滤一个主题中的消息。一个订阅的消费者在没有异常情况下会接受一个队列中的所有消息； 消息有序（部分支持）：需要将同一类的消息hash到同一个队列Queue中，才能支持消息的顺序，如果同一类消息散落到不同的Queue中，就不能支持消息的顺序，如果设定消息一定要正常消费，那么就不能保证消息顺序。 消息时序（可以支持）：可以发送定时消息，但是只能制定系统定义好的时间，不支持自定义时间； 容错处理（支持）：通过ACK机制，保证消息一定能正常消费，这个和RabbitMQ很像； 伸缩（支持）：整体架构其实和kafaka很像，可以扩容broker和内部队列数，或者增加消费组中的消费组数量，提高消费能力。 持久化（支持）：消息可以持久化到磁盘中，所以支持消息的回溯，和kafaka很像。 消息回溯（支持）：因为消息支持持久化，就支持回溯，可以理解是附带的功能。 高吞吐（非常好）：借鉴kafaka的设计，不会出现rabbitMQ的单Master抗压力问题，可以从多个borker写入和消费消息。