Spring Cloud Alibaba 微服务框架 RocketMQ：分布式消息队列

HTTP/RPC通信：优点是通信实时，缺点是服务之间的耦合性高

消息队列通信：优点是降低了服务之间的耦合性，提高了系统的处理能力，缺点是通信非实时

RocketMQ是一个低延迟、高可靠、可伸缩、易于使用的分布式消息中间件（也称消息队列）

RocketMQ具有高吞吐、低延迟、海量消息堆积等优点，同时提供顺序消息、事务消息、定时消息、消息重试与追踪等功能

非常适合在电商、金融等领域广泛使用

诸如秒杀、抢红包、企业开门红等大型活动皆会带来较高的流量脉冲，很可能因没做相应的保护而导致系统超负荷甚至崩溃

或因限制太过导致请求大量失败而影响用户体验，RocketMQ可提供削峰填谷的服务来解决这些问题

交易系统作为淘宝/天猫主站最核心的系统，每笔交易订单数据的产生会引起几百个下游业务系统的关注

包括物流、购物车、积分、流计算分析等，整体业务系统庞大而且复杂，RocketMQ可实现异步通信和应用解耦，确保主站业务的连续性

证券交易过程中的时间优先原则，交易系统中的订单创建、支付、退款等流程，航班中的旅客登机消息处理等

与先进先出（First In First Out，缩写FIFO）原理类似，RocketMQ提供的顺序消息即保证消息的FIFO

交易系统、红包等场景需要确保数据的最终一致性，大量引入RocketMQ的分布式事务

既可以实现系统之间的解耦，又可以保证最终的数据一致性

数据在“流动”中产生价值，传统数据分析大都基于批量计算模型，无法做到实时的数据分析

利用RocketMQ与流式计算引擎相结合，可以很方便地实现对业务数据进行实时分析

双11大促商品需要实时感知价格的变化，大量并发访问会导致会场页面响应时间长

集中式缓存因为带宽瓶颈限制商品变更的访问流量，通过RocketMQ构建分布式缓存，可实时通知商品数据的变化

在官网下载RocketMQ的最新版本4.6.0

解压源码并编译打包

启动集群管理NameServer，默认端口是9876

启动消息服务器Broker，指定NameServer的IP地址和端口，支持指定多个NameServer

name-server指定RocketMQ的NameServer地址

将指定名称为output的Binding消息发送到TopicTest

＠EnableBinding（{Source.class}）表示绑定配置文件中名称为output的消息通道Binding

Source类中定义的消息通道名称为output

发送HTTP请求http://localhost:8081/send？msg=tcever将消息发送到RocketMQ中

自定义消息通道的名称

name-server指定RocketMQ的NameServer地址

将指定名称为input的Binding消息接收TopicTest消息

＠EnableBinding（{Sink.class}）表示绑定配置文件中名称为input的消息通道Binding

Sink类中定义的消息通道的名称为input

＠StreamListener表示定义一个消息监听器，接收RocketMQ中的消息

自定义消息通道的名称

消费失败重试

定时消息

事务消息

顺序消息有明显缺陷

图片

Producer：消息发布的角色，主要负责把消息发送到Broker，支持分布式集群方式部署。

Consumer：消息消费者的角色，主要负责从Broker订阅消息消费，支持分布式集群方式部署。

Broker：消息存储的角色，主要负责消息的存储、投递和查询，以及服务高可用保证，支持分布式集群方式部署。

NameServer：服务管理的角色，主要负责管理Broker集群的路由信息，支持分布式集群方式部署。

每条消息必须属于一个Topic

每条消息拥有唯一的MessageID，且可以携带具有业务标识的Key

每个主题都包含若干条消息

每条消息都只能属于一个主题

Topic是RocketMQ进行消息订阅的基本单位

默认情况下一个Topic会对应多个Queue

Topic是逻辑概念，Queue是物理存储

在Consumer消费Topic消息时底层实际则拉取Queue的消息

来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签

标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统

消费者可以根据Tag实现对不同子主题的不同消费的处理逻辑，实现更好的扩展性

这类Producer发送同一类消息且发送逻辑一致

如果发送的是事务消息且原始生产者在发送之后崩溃，则Broker服务器会联系同一生产者组的其他生产者实例以提交或回溯消费

这类Consumer通常消费同一类消息且消费逻辑一致

消费者组使得在消息消费方面，实现负载均衡和容错的目标变得非常容易

要注意的是，消费者组的消费者实例必须订阅完全相同的Topic

是一个非常简单的Topic路由注册中心

其角色类似于Dubbo中依赖的ZooKeeper，支持Broker的动态注册与发现

核心功能

ZooKeeper具备选举功能，选举机制的原理就是少数服从多数

ZooKeeper的选举机制必须由ZooKeeper集群中的多个实例共同完成

ZooKeeper集群中的多个实例必须相互通信，如果实例数很多，网络通信就会变得非常复杂且低效

NameServer的设计目标是让网络通信变简单，从而使性能得到极大的提升

为了避免单点故障，NameServer也必须以集群的方式部署，但集群中各实例间相互不进行网络通信

NameServer是无状态的，可以任意部署多个实例

Broker向每一台NameServer注册自己的路由信息，因此每一个NameServer实例都保存一份完整的路由信息

NameServer中的Broker、Topic等状态信息不会持久存储，都是由各个角色定时上报并存储到内存中的(NameServer支持配置参数的持久化，一般用不到)

NameServer与每台Broker机器保持长连接，间隔30s从路由注册表中将故障机器移除

NameServer为了降低实现的复杂度，并不会立刻通知客户端的Producer和Consumer

NameServer是CAP中的AP架构

单个Broker跟所有NameServer保持心跳请求，心跳间隔为30秒，心跳请求中包括当前Broker所有的Topic信息。NameServer会反查Broker的心跳信息， 如果某个Broker在2分钟之内都没有心跳，则认为该Broker下线，调整Topic跟Broker的对应关系。但此时NameServer不会主动通知Producer、Consumer有Broker宕机

Consumer跟Broker是长连接，会每隔30秒发心跳信息到Broker。Broker端每10秒检查一次当前存活的Consumer，若发现某个Consumer 2分钟内没有心跳， 就断开与该Consumer的连接，并且向该消费组的其他实例发送通知，触发该消费者集群的负载均衡(rebalance)

Producer每30秒从NameServer获取Topic跟Broker的映射关系，更新到本地内存中。再跟Topic涉及的所有Broker建立长连接，每隔30秒发一次心跳。 在Broker端也会每10秒扫描一次当前注册的Producer，如果发现某个Producer超过2分钟都没有发心跳，则断开连接

问题1：当某个NameServer因宕机或网络问题下线了，Broker如何同步路由信息？

问题2：NameServer如果检测到Broker宕机，并没有立即通知Producer和Consumer， Producer将消息发送到故障的Broker怎么办？Consumer从Broker订阅消息失败怎么办？

问题3：由于NameServer集群中的实例相互不通信，在某个时间点不同NameServer实例保存的路由注册信息可能不一致

解决方案：由客户端解决，Producer和Consumer操作失败重试（Consumer消费也只是出现多一点延迟而已）

在Kafka中，Topic是逻辑概念，分区（Partition）是物理概念。 1个Topic可以设置多个分区，每个分区可以设置多个副本（Replication），即有1个Master分区、多个Slave分区

拓扑图

拓扑描述

在RocketMQ中，Topic也是逻辑概念，队列（Queue）是物理概念（对应Kafka中的分区）。 1个Topic可以设置多个队列，每个队列也可以有多个副本，即有1个Master队列、多个Slave队列。

拓扑图

拓扑描述

在Kafka中

在RocketMQ中

在applicaton.properties文件中设置同步发送

MessageBuilder设置Header信息头，表示这是一条顺序消息，将消息固定地发送到第0个消息队列

在applicaton.properties文件中设置顺序消费

消费实现代码不用改

局部有序

全局有序

同步

异步

单向

原理描述

核心源码

顺序消息

事务消息

延迟消息

日常开发过程中最常用

最常用的场景

尽量保证消息高性能收发即可

发送选择队列有2种机制

核心源码

过程描述

弊端

集群消费（Clustering:默认）：在集群消费模式下每条消息只会被ConsumerGroup的一个Consumer消费

广播消费（Broadcasting）：在广播消费模式下每条消息会被ConsumerGroup的每个Consumer消费

消费核心代码

消费流程描述

顺序消费，也称为有序消费

并发消费（默认），也称为乱序消费

同一个消息队列提供给Consumer中的多个消费线程拉取消费

Consumer中会维护一个消费线程池，多个消费线程可以并发去同一个消息队列中拉取消息进行消费

如果某个消费线程在监听器中进行业务处理时抛出异常，当前消费线程拉取的消息会进行重试，不影响其他消费线程和消息队列的消费进度

消费成功的线程正常提交消费进度

并发消费相比顺序消费没有资源争抢上锁的过程，消费消息的速度比顺序消费要快很多

RocketMQ不保证消息不被重复消费

如果业务对消费重复非常敏感，必须要在业务层面进行幂等性处理

具体实现可以通过分布式锁来完成

at-most-once（最多一次）：广泛使用

at-least-once（最少一次）：广泛使用

exactly-only-once（精确仅一次）

配置文件

核心代码

代码分析

核心代码

代码分析

配置文件

核心代码

消费事务消息与消费普通消息的代码是一样的，无须做任何修改

RocketMQ采用了2PC的方案来提交事务消息

第一阶段Producer向Broker发送预处理消息（也称半消息），此时消息还未被投递出去，Consumer不能消费

第二阶段Producer向Broker发送提交或回滚消息

发送预处理消息成功后，开始执行本地事务

如果本地事务执行成功，发送提交请求提交事务消息，消息会投递给Consumer

如果本地事务执行失败，发送回滚请求回滚事务消息，消息不会投递给Consumer

如果本地事务状态未知，网络故障或Producer宕机，Broker未收到二次确认的消息。 由Broker端发送请求给Producer进行消息回查，确认提交或回滚。 如果消息状态一直未被确认，需要人工介入处理

RocketMQ的消息的存储是由ConsumeQueue和CommitLog配合来完成的

ConsumeQueue中只存储很少的数据，消息主体都是通过CommitLog来进行读写

如果某个消息只在CommitLog中有数据，而ConsumeQueue中没有，则消费者无法消费

RocketMQ的事务消息实现就利用了这一点

数据存储设计

动态伸缩的能力

消息实时投递

这主要得益于其数据存储方式的设计

数据存储的核心由两部分组成

数据存储流程

寻道：磁头移动定位到指定磁道，时间很长，是指找到数据在哪个地方。

旋转延迟：等待指定扇区旋转至磁头下，机械硬盘和每分钟多少转有关系，时间很短。

数据传输：数据通过系统总线从磁盘传送到内存，时间很短。

磁盘读写最慢的动作是寻道

缩短寻道时间就能有效提升磁盘的读写速度，最优的方式就是不用寻道

随机写会导致磁头不停地更换磁道，时间都花在寻道上了

顺序写几乎不用换磁道，或者寻道时间很短

CommitLog文件是负责存储消息数据的文件

所有Topic的消息都会先存在{ROCKETMQ_HOME}/store/commitlog文件夹下的文件中

消息数据写入CommitLog文件是加锁串行追加写入

为了保证消息发送的高吞吐量，使用单个文件存储所有Topic的消息

当消息到达CommitLog文件后，会通过线程异步几乎实时地将消息转发给消费队列文件

每个CommitLog文件的默认大小是1GB，写满1GB再写新的文件，大量数据I/O都在顺序写同一个CommitLog文件

文件名按照该文件起始的总的字节偏移量offset命名，文件名固定长度为20位，不足20位前面补0

文件名这样设计的目的

消费Broker中存储消息的实际工作就是读取文件，但消息数据文件中所有Topic的消息数据是混合在一起的

消费消息时是区分Topic消费的，这就导致如果消费时也读取CommitLog文件会使得消费消息的性能差、吞吐量低

为了解决消息数据文件顺序写难以读取的问题，RocketMQ中设计了消费队列ConsumeQueue

ConsumeQueue负责存储消费队列文件

在消息写入CommitLog文件时，会异步转发到ConsumeQueue文件，然后提供给Consumer消费

ConsumeQueue文件中并不存储具体的消息数据，只存CommitLog的偏移量offset、消息大小size、消息Tag Hashcode

ConsumeQueue文件中存储的内容

每个Topic在某个Broker下对应多个队列Queue，默认是4个消费队列Queue

每一条记录的大小是20B，默认一个文件存储30万个记录

文件名同样也按照字节偏移量offset命名，文件名固定长度为20位，不足20位前面补0

在集群模式下，Broker会记录客户端对每个消费队列的消费偏移量，定位到ConsumeQueue里相应的记录

并通过CommitLog的Offset定位到CommitLog文件里的该条消息

消费定位流程图

消费Broker中存储消息的实际工作就是读取文件，消息队列文件是一种数据结构上的设计

磁盘顺序读写和消息队列文件的设计，为了高性能读数据，除此之外，还使用了操作系统中的PageCache机制

RocketMQ读取消息依赖操作系统的PageCache

检查要读的数据是否在上次预读的Cache中

如果没有命中Cache，操作系统从磁盘中读取对应的数据页，并将该数据页之后的连续几页一起读入Cache中， 再将应用需要的数据返回给应用，这种方式称为跳跃读取

如果命中Cache，上次缓存的数据有效，操作系统认为在顺序读盘，则继续扩大缓存的数据范围， 将之前缓存的数据页之后的几页数据再读取到Cache中

由于ConsumeQueue和CommitLog文件都是顺序写入的，客户端消费也是按顺序消费，所以能极大的提高PageCache命中率

按速度高低排列为：CPU>RAM>DISK

CPU与RAM之间、RAM与DISK之间的速度和容量差异是指数级的

为了在速度和容量上折中

利用了Java NIO中的FileChannel模型

可以直接将物理文件映射到缓冲区的PageCache

少了一次数据拷贝过程，提高了读写速度

图片

许多web应用都会向用户提供大量的静态内容，这意味着有很多data从硬盘读出之后，会原封不动的通过socket传输给用户

这种操作看起来可能不会怎么消耗CPU，但是实际上它是低效的

data每次穿过user态和kernel边界时，都会被copy，这会消耗cpu，并且占用RAM的带宽

零拷贝指的是用户态与内核态间copy数据的次数为零

零拷贝避免了用户态和内核态间的copy、减少了两次用户态内核态间的切换

零拷贝可以提高数据传输效率，但对于需要在用户传输过程中对数据进行加工的场景（如加密）就不适合使用零拷贝

为什么不是直接把data传到user buffer中，看起来比较低效(多了一次copy)，原因如一下：

kernel buffer是用来提高性能的

在进行读操作的时候，kernel buffer起到了预读cache的作用

当写请求的data size比kernel buffer的size小的时候，这能够显著的提升性能（批处理逻辑）

在进行写操作时，kernel buffer的存在可以使得写请求完全异步

缺点：当请求的data size远大于kernel buffer size的时候，这个方法本身变成了性能的瓶颈

Java NIO中的FileChannel拥有transferTo和transferFrom两个方法

该接口常被用于高效的网络/文件的数据传输和大文件拷贝

在操作系统支持的情况下，通过该方法传输数据并不需要将源数据从内核态拷贝到用户态，再从用户态拷贝到目标通道的内核态

同时也减少了两次用户态和内核态间的上下文切换，也即使用了“零拷贝”

一个Consumer实例可以同时消费多个消息队列中的消息

消息队列数在创建Topic时可以指定，也可以在运行中修改

如果一个Topic的消息量特别大，但Broker集群水位压力还是很低， 就可以对该Topic的消息队列进行扩容，Topic的消息队列数跟消费速度成正比

如果一个Topic的消息量特别小，但该Topic的消息队列数很多，则可以对该Topic消息队列缩容

如果一个Topic的消息量特别大，但Broker集群水位很高，此时就需要对Broker机器扩容

如果Broker集群水位很低，则可以适当减少Broker服务器来节约成本

扩容方式

Push推模式

Pull拉模式

Push和Pull这两种获取消息方式的缺点都很明显，单一的方式难以应对复杂的消费场景

所以RocketMQ中提供了一种推/拉结合的长轮询机制来平衡推/拉模式各自的缺点

长轮询本质上是对普通pull模式的优化

kafka也是采用pull模式

对pull模式做了优化

消息发送的高可用

消息存储的高可用

消息消费的高可用

集群管理的高可用

重试机制仅支持同步发送方式，不支持异步和单向发送方式

根据发送失败的异常类型处理策略略有不同

图片

在开启的情况下，消息发送失败的时候会将失败的Broker暂时排除在队列选择列表外

规避时间是衰减的，如果Broker一直不可用，会被NameServer检测到并在Producer更新路由信息时进行剔除

先按轮询算法选择一个消息队列

从故障列表判断该消息队列是否可用

故障机器FaultItem在消息发送失败时进入故障列表faultItemTable中

FaultItem存储了Broker名称、响应时长、故障规避开始时间，最重要的是故障规避开始时间，会用来判断Queue是否可用

故障周期时长为0就代表了没有故障

当响应时长超过100毫秒时代表Broker可能机器出现问题，也会进入故障列表

网络异常、Broker异常、客户端异常则设定故障周期为10分钟

同步刷盘

异步刷盘

RocketMQ为了提高消息消费的高可用性

避免Broker发生单点故障引起存储在Broker上的消息无法及时消费

同时避免单个机器上硬盘坏损出现消息数据丢失

同步复制：SYNC_MASTER

异步复制：ASYNC_MASTER

通过配置文件${ROCKETMQ_HOME}/conf/broker.conf里的brokerRole参数进行设置复制方式

在实际应用中需要结合业务场景，合理设置刷盘方式和主从复制方式

不建议使用SYNC_FLUSH同步它刷盘方式，因为它会频繁地触发写磁盘操作，性能下降很明显

高性能是RocketMQ的一个明显特点，因此放弃性能是不合适的选择

通常可以把Master和Slave设置成ASYNC_FLUSH异步刷盘、SYNC_MASTER同步复制，这样即使有一台服务器出故障，仍然可以保证数据不丢失

RocketMQ的Consumer在拉取消息时，Broker会判断Master服务器的消息堆积量以决定Consumer是否从Slave服务器拉取消息消费

默认一开始从Master服务器上拉取消息

如果Master服务器的消息堆积超过了物理内存的40%，则会在返回给Consumer的消息结果里告知Consumer， 下次需要从其他Slave服务器上拉取消息

实际业务场景中无法避免消费消息失败的情况

消费失败可能是因为业务处理中调用远程服务网络问题失败，不代表消息一定不能被消费，通过重试可以解决

进入机制

特点

进入机制

特点

通常故障恢复需要一定的时间，如果不间断地重试，重试又失败的情况会占用并浪费资源

所以RocketMQ的消费重试机制采用时间衰减的方式，使用了自身定时消费的能力

首次在10秒后重试消费，如果消费成功则不再重试，如果消费失败则继续重试消费

第二次在30秒后重试消费，依此类推，每次重试的间隔时间都会加长

直到超出最大重试次数（默认为16次），则进入死信队列不再重试

重试消费过程中的间隔时间使用了定时消息

重试的消息数据并非直接写入重试队列，而是先写入定时消息队列，再通过定时消息的功能转发到重试队列

也称延迟消息，等待指定的延迟时间后再进行消费（RocketMQ支持）

除了支持消费重试机制，延迟消息也适用于一些处理异步任务的场景

RocketMQ不支持任意时间精确的延迟消息

仅支持1s、5s、10s、30s、1min、2min、3min、4min、5min、6min、7min、8min、9min、10min、20min、30min、1h、2h

在实际业务场景中，业务应用在消费消息的过程中偶尔会出现一些异常情况

例如程序发布导致的重启，或网络突然出现问题

此时正在进行业务处理的消息可能消费完了，也可能业务逻辑执行到一半没有消费完

通过消息的ACK机制识别有没有消费完

在消息投递过程中，不是消息从Broker发送到Consumer就算消费成功了

需要Consumer明确给Broker返回消费成功状态才算（提交ACK给broker成功）

如果从Broker发送到Consumer后，已经完成了业务处理，但在给Broker返回消费成功状态之前

Consumer发生宕机或断电、断网等情况，Broker未收到反馈，则不会保存消费进度（提交ACK给broker失败）

Consumer重启之后，消息会重新投递，此时也会出现重复消费的场景

因此消息幂等性需要业务自行保证

Broker集群部署是消息存储高可用的基本保障

最直接的表现是Broker出现单机故障或重启时，不会影响RocketMQ整体的服务能力

单Master模式仅部署一台Broker机器，属于非集群模式

优点：部署方式简单

缺点

一个集群全部都是Master机器，没有Slave机器，属于不配置主从复制的场景

优点

缺点

每个Master配置一个Slave，有多对Master-Slave，主从复制采用异步复制方式，主备有短暂消息延迟（毫秒级）

优点

缺点

每个Master配置一个Slave，有多对Master-Slave，主从复制采用同步复制方式，即只有主备都写成功，才向应用返回成功

优点

缺点

线上推荐使用异步刷盘+同步复制的多Master多Slave模式

图片

应用程序通过Spring Cloud Stream注入的输入通道inputs和输出通道outputs与消息中间件Middleware通信

消息通道通过特定的中间件绑定器Binder实现连接到外部代理

Spring Cloud Stream的实现基于发布/订阅机制

Spring Messaging（Spring Framework）

Spring Integration（Spring Framework）

Binders（Spring Cloud Stream）

Bindings（Spring Cloud Stream）

Spring Cloud Stream提供了简单易用的消息编程模型，内部基于发布/订阅模型实现

Spring Cloud Stream的Binder提供标准协议，不同的消息中间件都可以按照标准协议接入

Binder提供bindConsumer和bindProducer接口协议，分别用于构造生产者和消费者

图片

MessageChannel（output）： 消息通道，用于发送消息，Spring Cloud Stream的标准接口。

MessageChannel（input）： 消息通道，用于订阅消息，Spring Cloud Stream的标准接口。

Binder bindProducer： 目标绑定器，将发送通道发过来的消息发送到RocketMQ消息服务器， 由Spring Cloud Alibaba团队按照Spring Cloud Stream的标准协议实现。

Binder bindConsumer： 目标绑定器，将接收到RocketMQ消息服务器的消息推送给订阅通道， 由Spring Cloud Alibaba团队按照Spring Cloud Stream的标准协议实现。

图片

图片