RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件前身是MetaQ是阿里参考Kafka特点研发的一个队列模型的消息中间件后开源给apache基金会成为了apache的顶级开源项目具有高性能、高可靠、高实时、分布式特点。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。分为ProducerGroupConsumerGroup代表某一类的生产者和消费者一般来说同一个服务可以作为Group同一个Group一般来说发送和消费的消息都是一样的。Producer Group生产者组代表某一类的生产者比如我们有多个秒杀系统作为生产者这多个合在一起就是一个 Producer Group 生产者组它们一般生产相同的消息。Consumer Group消费者组代表某一类的消费者比如我们有多个短信系统作为消费者这多个合在一起就是一个 Consumer Group 消费者组它们一般消费相同的消息。Queue队列在Kafka中叫Partition。每个Queue内部是有序的在RocketMQ中分为读和写两种队列一般来说读写队列数量一致如果不一致就会出现很多问题。Message Queue消息队列主题被划分为一个或多个子主题即消息队列。一个 Topic 下可以设置多个消息队列发送消息时执行该消息的 Topic RocketMQ 会轮询该 Topic 下的所有队列将消息发出去。消息的物理管理单位。一个Topic下可以有多个QueueQueue的引入使得消息的存储可以分布式集群化具有了水平扩展能力。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大小而且消息堆积后性能下降不会太大因为内存中数据多少对于对外提供的访问能力影响有限。消息堆积到持久化存储系统中例如DBKV存储文件记录形式。当消息不能在内存Cache命中时要不可避免的访问磁盘会产生大量读IO读IO的吞吐量直接决定了消息堆积后的访问能力。评估消息堆积能力主要有以下四点消息能堆积多少条多少字节即消息的堆积容量。消息堆积后发消息的吞吐量大小是否会受堆积影响消息堆积后正常消费的Consumer是否会受影响消息堆积后访问堆积在磁盘的消息时吞吐量有多大简单来说RocketMQ支持大量消息堆积消息会存在内存超出内存的消息会持久化到磁盘中。4.4.5 定时消息定时消息是指消息发到Broker后不能立刻被Consumer消费要到特定的时间点或者等待特定的时间后才能被消费。如果要支持任意的时间精度在Broker层面必须要做消息排序如果再涉及到持久化那么消息排序要不可避免的产生巨大性能开销。RocketMQ支持定时消息但是不支持任意时间精度支持特定的level例如定时5s10s1m等。简单来说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 主从模式下主节点返回消息给客户端的时候是否需要同步从节点。同步复制也叫 “同步双写”也就是说只有消息同步双写到主从结点上时才返回写入成功。异步复制消息写入主节点之后就直接返回写入成功。异步复制会不会也像异步刷盘那样影响消息的可靠性呢答案是不会的因为两者就是不同的概念对于消息可靠性是通过不同的刷盘策略保证的而像异步同步复制策略仅仅是影响到了可用性。为什么呢其主要原因是 RocketMQ 是不支持自动主从切换的当主节点挂掉之后生产者就不能再给这个主节点生产消息了。比如这个时候采用异步复制的方式在主节点还未发送完需要同步的消息的时候主节点挂掉了这个时候从节点就少了一部分消息。但是此时生产者无法再给主节点生产消息了消费者可以自动切换到从节点进行消费仅仅是消费所以在主节点挂掉的时间只会产生主从结点短暂的消息不一致的情况降低了可用性而当主节点重启之后从节点那部分未来得及复制的消息还会继续复制。扩展知识1在单主从架构中如果一个主节点挂掉了那么也就意味着整个系统不能再生产了。那么这个可用性的问题能否解决呢一个主从不行那就多个主从的呗别忘了在我们最初的架构图中每个 Topic 是分布在不同 Broker 中的。但是这种复制方式同样也会带来一个问题那就是无法保证严格顺序。在上文中我们提到了如何保证的消息顺序性是通过将一个语义的消息发送在同一个队列中使用 Topic 下的队列来保证顺序性的。如果此时我们主节点 A 负责的是订单 A 的一系列语义消息然后它挂了这样其他节点是无法代替主节点A的如果我们任意节点都可以存入任何消息那就没有顺序性可言了。这点我并不认同我理解主从的对列信息应该是一样的我从主节点读到哪里如果主节点挂掉应该是可以到从结点去读取的如果不能这样搞个主从就完全没有意义了。因为主从的信息是一样的对队列的顺序是有影响的我不可能把不同的信息搞两个队列分别放到主从机器。扩展知识2在 RocketMQ 中采用了 Dledger 解决主从数据同步问题。他要求在写入消息的时候要求至少消息复制到半数以上的节点之后才给客⼾端返回写⼊成功并且它是⽀持通过选举来动态切换主节点的。这里我就不展开说明了读者可以自己去了解。也不是说 Dledger 是个完美的方案至少在 Dledger 选举过程中是无法提供服务的而且他必须要使用三个节点或以上如果多数节点同时挂掉他也是无法保证可用性的而且要求消息复制板书以上节点的效率和直接异步复制还是有一定的差距的。这个机制感觉就像大众化的版本基本思路都一样为了保证数据可用性我还是推荐同步复制当大多数节点复制成功就认为复制完毕和ETCD的Raft协议的日志同步原理一样。4.4.8 容错机制在实际使用RocketMQ的时候我们并不能保证每次发送的消息都刚好能被消费者一次性正常消费成功可能会存在需要多次消费才能成功或者一直消费失败的情况那作为发送者该做如何处理呢RocketMQ提供了ack机制以保证消息能够被正常消费。发送者为了保证消息肯定消费成功只有使用方明确表示消费成功RocketMQ才会认为消息消费成功。中途断电抛出异常等都不会认为成功——即都会重新投递。当然如果消费者不告知发送者我这边消费信息异常那么发送者是不会知道的所以消费者在设置监听的时候需要给个回调。为了保证消息是肯定被至少消费成功一次RocketMQ会把这批消息重发回Brokertopic不是原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写入和消费消息。