Rocket MQ 的前世与今生

RocketMQ是⼀款阿⾥巴巴开源的消息中间件,在2017年9⽉份成为Apache的顶级项⽬,是国内⾸个互联⽹中间件在 Apache 上的顶级项⽬。

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(1)

RocketMQ经历了4个大阶段(如上图):

阶段1:Metaq(Metamorphosis) 1.x 由开源社区 killme2008 维护,最后⼀次更新是在2017年1⽉份。

阶段2:Metaq 2.x 于 2012 年 10 ⽉份上线,在淘宝内部被⼴泛使⽤。

阶段3:RocketMQ 3.x 基于阿⾥内部开源共建原则, RocketMQ 项⽬只维护核⼼功能,且去除了所有其他运⾏时依赖, 核⼼功能最简化。每个 BU 的个性化需求都在 RocketMQ 项⽬之上进⾏深度定制。 RocketMQ 向其他 BU 提供的仅仅是Jar 包,例如要定制⼀个 Broker,那么只需要依赖 rocketmq-broker 这个 jar 包即 可,可通过 API 迕⾏交互,如果定制 client,则依赖 rocketmq-Client 这个 jar 包,对其提供的 api 进⾏ 再封装。

阶段4:进⼊Apache 2016年11⽉28⽇,阿⾥巴巴向 Apache 软件基⾦会捐赠消息中间件 RocketMQ,成为 Apache 孵 化项⽬。美国时间 2017 年 9 ⽉ 25 ⽇,Apache 软件基⾦会(ASF)宣布 Apache®RocketMQ™ 已孵化 成为 Apache 顶级项⽬(TLP ),是国内⾸个互联⽹中间件在 Apache 上的顶级项⽬。

各种MQ产品的比较

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(2)

此处比较并没有列出ActivateMQ,因为ActivateMQ目前使用占额已经越老越小,所以此次 不再拿出来进行比较。

功能对比:

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(3)

总结:

使用场景

1.异步解耦

场景:用户注册后,需要发送注册邮件和短信通知,以告知用户注册成功;传统的做法有以下两种。

串行方式:

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(4)

并行方式:

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(5)

MQ解耦:

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(6)

总结:从三种实现方式可以看出,使用MQ进行异步解耦能让服务的功能变得更单一、更高效,明确功能的主次。

2.削峰填谷

场景:在秒杀或团队抢购活动中,由于用户请求量较大,导致流量暴增,秒杀的应用在处理如此大量的访问流量后,下游的通知系统无法承载海量的调用量,甚至会导致系统崩溃等问题而发生漏通知的情况。

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(7)

总结:在互联网开发过程中突发大流量情况非常多,也有很多成熟的解决方案,使用MQ来进行削峰填谷也是业界比较成熟的一个方案。

3.数据同步与check

场景:订单数据存储为分片数据库,不利于聚合查询和批量操作,利用binlog将数据同步到MQ中,业务消费MQ对数据加工处理,可以将数据同步到其它存储系统,也可以做数据Check等

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(8)

总结:采用订阅Binlog方式,将数据打进MQ进行临时存储,其它服务可以通过订阅MQ进行处理自己的业务,比如数据同步、数据一致性检查待。

4.其它场景

当然MQ还有其它一些使用场景,在这里就不在一一介绍了,具体如下:

这些使用场景可以说是MQ的优点,但是同样也会带来一些问题:

使用MQ后的缺点:

1)系统更复杂,多了一个MQ组件

2)消息传递路径更长,延时会增加

3)消息可靠性和重复性互为矛盾,消息不丢不重难以同时保证

4)上游无法知道下游的执行结果,这一点是很致命的

架构原理

基本概念(来自官网)

1.消息模型(Message Model)

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

2 消息生产者(Producer)

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

3 消息消费者(Consumer)

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

4 主题(Topic)

表示一类消息的集合,每个主题包含若干条消息,每条消息只能属于一个主题,是RocketMQ进行消息订阅的基本单位。

5 代理服务器(Broker Server)

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

6 名字服务(Name Server)

名称服务充当路由消息的提供者。生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。多个Namesrv实例组成集群,但相互独立,没有信息交换。

7 拉取式消费(Pull Consumer)

Consumer消费的一种类型,应用通常主动调用Consumer的拉消息方法从Broker服务器拉消息、主动权由应用控制。一旦获取了批量消息,应用就会启动消费过程。

8 推动式消费(Push Consumer)

Consumer消费的一种类型,该模式下Broker收到数据后会主动推送给消费端,该消费模式一般实时性较高。

9 生产者组(Producer Group)

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

10 消费者组(Consumer Group)

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

11 集群消费(Clustering)

集群消费模式下,相同Consumer Group的每个Consumer实例平均分摊消息。

12 广播消费(Broadcasting)

广播消费模式下,相同Consumer Group的每个Consumer实例都接收全量的消息。

13 普通顺序消息(Normal Ordered Message)

普通顺序消费模式下,消费者通过同一个消息队列( Topic 分区,称作 Message Queue) 收到的消息是有顺序的,不同消息队列收到的消息则可能是无顺序的。

14 严格顺序消息(Strictly Ordered Message)

严格顺序消息模式下,消费者收到的所有消息均是有顺序的。

15 消息(Message)

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

16 标签(Tag)

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

技术架构

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(9)

RocketMQ架构上主要分为四部分,如上图所示:

BrokerServer 核心模块

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(10)

部署架构

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(11)

RocketMQ 网络部署特点

结合部署架构图,描述集群工作流程:

存储架构

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(12)

消息存储架构图中主要有下面三个跟消息存储相关的文件构成。

(1) commitLog:消息主体以及元数据的存储主体,存储Producer端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G, 文件名长度为20位,左边补零,剩余为起始偏移量,比如00000000000000000000代表了第一个文件,起始偏移量为0,文件大小为1G=1073741824;当第一个文件写满了,第二个文件为00000000001073741824,起始偏移量为1073741824,以此类推。消息主要是顺序写入日志文件,当文件满了,写入下一个文件;

(2) ConsumeQueue:消息消费队列,引入的目的主要是提高消息消费的性能,由于RocketMQ是基于主题topic的订阅模式,消息消费是针对主题进行的,如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中,ConsumeQueue(逻辑消费队列)作为消费消息的索引,保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset,消息大小size和消息Tag的hashCode值。consumequeue文件可以看成是基于topic的commitlog索引文件,故consumequeue文件夹的组织方式如下:topic/queue/file三层组织结构,具体存储路径为:$HOME/store/consumequeue/{topic}/{queueId}/{FileName}。同样consumequeue文件采取定长设计,每一个条目共20个字节,分别为8字节的commitlog物理偏移量、4字节的消息长度、8字节tag hashcode,单个文件由30W个条目组成,可以像数组一样随机访问每一个条目,每个ConsumeQueue文件大小约5.72M;

(3) IndexFile:IndexFile(索引文件)提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是:HOME\store\indexHOME\store\index{fileName},文件名fileName是以创建时的时间戳命名的,固定的单个IndexFile文件大小约为400M,一个IndexFile可以保存 2000W个索引,IndexFile的底层存储设计为在文件系统中实现HashMap结构,故rocketmq的索引文件其底层实现为hash索引。

在上面的RocketMQ的消息存储整体架构图中可以看出,RocketMQ采用的是混合型的存储结构,即为Broker单个实例下所有的队列共用一个日志数据文件(即为CommitLog)来存储。RocketMQ的混合型存储结构(多个Topic的消息实体内容都存储于一个CommitLog中)针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构,Producer发送消息至Broker端,然后Broker端使用同步或者异步的方式对消息刷盘持久化,保存至CommitLog中。只要消息被刷盘持久化至磁盘文件CommitLog中,那么Producer发送的消息就不会丢失。正因为如此,Consumer也就肯定有机会去消费这条消息。当无法拉取到消息后,可以等下一次消息拉取,同时服务端也支持长轮询模式,如果一个消息拉取请求未拉取到消息,Broker允许等待30s的时间,只要这段时间内有新消息到达,将直接返回给消费端。这里,RocketMQ的具体做法是,使用Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue(逻辑消费队列)和IndexFile(索引文件)数据。

消息刷盘

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(13)

同步刷盘:只有在消息真正持久化至磁盘后RocketMQ的Broker端才会真正返回给Producer端一个成功的ACK响应。同步刷盘对MQ消息可靠性来说是一种不错的保障,但是性能上会有较大影响,一般适用于金融业务应用该模式较多。

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(14)

异步刷盘:能够充分利用OS的PageCache的优势,只要消息写入PageCache即可将成功的ACK返回给Producer端。消息刷盘采用后台异步线程提交的方式进行,降低了读写延迟,提高了MQ的性能和吞吐量

其它设计原理

消息样例

普通消息-同步发送

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(15)

此种方式应用场景非常广泛,例如重要通知邮件、报名短信通知、营销短信系统等。

普通消息-异步发送

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(16)

异步发送一般用于链路耗时较长,对响应时间较为敏感的业务场景,例如,您视频上传后通知启动转码服务,转码完成后通知推送转码结果等。

普通消息-单向发送

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(17)

适用于某些耗时非常短,但对可靠性要求并不高的场景,例如日志收集。

普通消息-三种发放方式对比

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(18)

顺序消息

在MQ的模型中,顺序需要由3个阶段去保障:

  1. 消息发送时保持顺序
  2. 消息被存储时保持和发送的顺序⼀致
  3. 消息被消费时保持和存储的顺序⼀致

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(19)

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(20)

上图是RocketMQ顺序消息原理的介绍,将不同订单的消息路由到不同的分区中。文档只是给出了Producer顺序的处理,Consumer消费时通过一个分区只能有一个线程消费的方式来保证消息顺序。

顺序消息-Producer

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(21)

顺序消息-Consumer

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(22)

总结:从上面顺序消息的发送和接收代码可以看出,在发送时是采用取模方式进行选择固定队列 ,在接收时是采用单线程的方式去消费队列,从而保证消息的顺序性。(注:单核单线程情况)

延迟消息

延迟消息处理流程:

  1. 修改消息Topic名称和队列信息
  2. 转发消息到延迟主题SCHEDULE_TOPIC_XXXX的CosumeQueue中
  3. 延迟服务消费SCHEDULE_TOPIC_XXXX消息
  4. 将信息重新存储到CommitLog中
  5. 将消息投递到⽬标Topic中
  6. 消费者消费⽬标topic中的数据

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(23)

ConsumerQueue格式说明

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(24)

Commit Log Offset:记录在CommitLog中的位置。

Size:记录消息的大小

Message Tag HashCode:记录消息Tag的哈希值,用于消息过滤。特别的,对于延迟消息,这个字段记录的是消息的投递时间戳。这也是为什么java中hashCode方法返回一个int型,只占用4个字节,而这里Message Tag HashCode字段却设计成8个字节的原因

广播消息

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(25)

批量消息

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(26)

批量发送可以提⾼发送性能,但有⼀定的限制:

事务消息

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(27)

事务消息发送:

1. 发送⽅将半事务消息发送⾄消息队列 RocketMQ 版服务端。

2. 消息队列 RocketMQ 版服务端将消息持久化成功之后,向发送⽅返回 Ack 确认消息已经发送成功,此时消息为半事务消息。

3. 发送⽅开始执⾏本地事务逻辑。

4. 发送⽅根据本地事务执⾏结果向服务端提交⼆次确认(Commit 或是 Rollback),服务端收到 Commit 状态则将半事务消息标记为可投递,订阅⽅最终将收到该消息;服务端收到 Rollback 状态则删除半事务消息,订阅⽅将不会接受该消息。

事务消息回查:

1.在断⽹或者是应⽤重启的特殊情况下,上述步骤 4 提交的⼆次确认最终未到达服务端,经过固定 时间后服务端将对该消息发起消息回查。

2. 发送⽅收到消息回查后,需要检查对应消息的本地事务执⾏的最终结果。

3. 发送⽅根据检查得到的本地事务的最终状态再次提交⼆次确认,服务端仍按照步骤 4 对半事务消 息进⾏操作。

事务状态

事务消息共有三种状态,提交状态、回滚状态、中间状态:

事务消息-Producer

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(28)

事务消息-Listener

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(29)

事务消息原理

rocketmq中间件如何推送(互联网公司都喜欢的消息队列)(30)

HALF消息:

OP消息:

回查: 对⽐HALF消息和OP消息进⾏回查

结语

Rocket MQ是目前MQ产品中使用非常广泛的,它既可以保证数据的传输可靠性,又有很可观的性能表现,这也是它越来越受欢迎的原因。

,