RocketMQ开发指导之一——RocketMQ简介

本系列文章主要介绍 RocketMQ 的相关知识&＃xff0c;并通过示例代码介绍 RocketMQ 的使用方法。

本文为系列文章的第一篇&＃xff0c;主要介绍 RocketMQ 的概要知识。

说明&＃xff1a;本文的部分内容参考了 https://www.jianshu.com/p/3afd610a8f7d 文章的相关内容。

1 概述

首先&＃xff0c;给出 GitHub 上 RocketMQ 项目的描述&＃xff0c;内容如下&＃xff1a;

Apache RocketMQ is a distributed messaging and streaming platform with low latency, high performance and reliability, trillion-level capacity and flexible scalability.

It offers a variety of features:

• Pub/Sub messaging model
• Scheduled message delivery
• Message retroactivity by time or offset
• Log hub for streaming
• Big data integration
• Reliable FIFO and strict ordered messaging in the same queue
• Efficient pull&push consumption model
• Million-level message accumulation capacity in a single queue
• Multiple messaging protocols like JMS and OpenMessaging
• Flexible distributed scale-out deployment architecture
• Lightning-fast batch message exchange system
• Various message filter mechanics such as SQL and Tag
• Docker images for isolated testing and cloud isolated clusters
• Feature-rich administrative dashboard for configuration, metrics and monitoring

RocketMQ 作为一款分布式的消息中间件&＃xff0c;经历了 Metaq1.x、Metaq2.x 的发展和淘宝双十一的洗礼&＃xff0c;证明其在功能和性能上远超 ActiveMQ。GitHub 上关于 RocketMQ 诞生的原因中&＃xff0c;也说明了随着交易量的大幅度增长&＃xff0c;ActiveMQ 到达了性能瓶颈&＃xff0c;而其他流行的消息解决方案&＃xff08;如 Kafka&＃xff09;都不能满足其需求的情况下&＃xff0c;才诞生了 RocketMQ。

RocketMQ 的优点如下&＃xff1a;

• RocketMQ 原生就是支持分布式的&＃xff0c;而 ActiveMQ 原生为单点性&＃xff1b;
• RocketMQ 可以保证严格的消息顺序&＃xff0c;而 ActiveMQ 无法保证&＃xff1b;
• RocketMQ 提供亿级消息的堆积能力&＃xff0c;这不是重点&＃xff0c;重点是堆积了亿级的消息后&＃xff0c;依然保持写入低延迟&＃xff1b;
• 丰富的消息拉取模式&＃xff08;Push or Pull&＃xff09;。Push 模式好理解&＃xff0c;比如在消费者端设置 Listener 回调&＃xff1b;而 Pull 模式&＃xff0c;控制权在于消费者&＃xff0c;即消费者需要主动地调用拉消息方法从 Broker 获取消息&＃xff0c;这里面就存在一个消费位置记录的问题&＃xff08;如果不记录&＃xff0c;会导致消息重复消费&＃xff09;&＃xff1b;
• 在 Metaq1.x/2.x 的版本中&＃xff0c;分布式协调采用的是 Zookeeper&＃xff0c;而 RocketMQ 自己实现了一个 NameServer&＃xff0c;这使得 RocketMQ 的分布式架构更加轻量级&＃xff0c;性能更好&＃xff1b;
• 消息失败重试机制、高效的订阅者水平扩展能力、强大的API、事务机制。

2 Producer/Consumer Group

ActiveMQ 中并没有 Group 这个概念&＃xff0c;而在 RocketMQ 中存在 Group 的机制&＃xff0c;理解该机制对于深入理解 RocketMQ 非常重要。

RocketMQ 通过 Group 机制&＃xff0c;天然地支持了消息负载均衡。例如&＃xff0c;某个 Topic 有 9 条消息&＃xff0c;其中一个 Consumer Group 有 3 个实例&＃xff08;3 个进程/3 台机器&＃xff09;&＃xff0c;那么每个实例将均摊 3 条消息&＃xff0c;由此实现了负载均衡。&＃xff08;注意&＃xff1a;RocketMQ 只有一种模式&＃xff0c;即发布订阅模式&＃xff09;

3 集群模式

RocketMQ 有多种 Broker 集群部署模式&＃xff0c;常见的包括&＃xff1a;单 Master 模式、多 Master 模式、多 Master 多 Slave 模式&＃xff08;异步复制&＃xff09;、多 Master 多 Slave 模式&＃xff08;同步双写&＃xff09;等。这里需要强调一下&＃xff1a;RocketMQ 的 Slave 只能被消费者读取&＃xff0c;不可以被生产者写入&＃xff0c;类似于 MySQL 的主从机制。下面分别介绍这几种 Broker 集群部署模式。

3.1 单Master模式

很显然&＃xff0c;单 Master 模式部署风险较大&＃xff0c;一旦这个 Broker 重启或宕机&＃xff0c;会导致整个服务不可用&＃xff0c;通常线上环境都不会使用此模式。

3.2 多Master模式

集群中全是 Master&＃xff0c;没有 Slave&＃xff0c;例如 2 个 Master 或 3 个 Master。此时&＃xff0c;如果某一个 Broker 重启或宕机&＃xff0c;对应用是无影响的。此模式的缺点在于&＃xff0c;当某个 Master 宕机时&＃xff0c;该 Master 上未被消费的消息在 Master 恢复之前是不可以订阅的&＃xff0c;消息的实时性会受到影响。

3.3 多Master多Slave模式&＃xff08;异步复制&＃xff09;

此模式下&＃xff0c;有多个 Master&＃xff0c;每个 Master 会配置一个或多个 Slave&＃xff0c;由此实现了系统的高可用性。同时&＃xff0c;Master 与 Slave 之间的消息同步&＃xff0c;采用异步复制的方式&＃xff0c;主备之间会短暂的消息延迟&＃xff0c;这种延迟是 MS 级别的。如果 Master 宕机&＃xff0c;消费者可以从 Slave 上进行消息消费&＃xff0c;不影响消息实时性&＃xff0c;但是由于 Master 的宕机&＃xff0c;会导致丢失掉极少量&＃xff08;尚未同步到 Slave 上&＃xff09;的消息。

3.4 多Master多Slave模式&＃xff08;同步双写&＃xff09;

此模式下&＃xff0c;有多个 Master&＃xff0c;每个 Master 会配置一个或多个 Slave&＃xff0c;由此实现了系统的高可用性。同时&＃xff0c;Master 与 Slave 之间的消息同步&＃xff0c;采用同步双写的方式&＃xff0c;也就是在 Master 和 Slave 都写成功的前提下&＃xff0c;才会向应用&＃xff08;生产者&＃xff09;返回成功。显然&＃xff0c;此种模式下&＃xff0c;无论是数据还是服务都不是单点的&＃xff0c;所以服务与数据的可用性都非常高。此模式的缺点在于&＃xff0c;性能会比异步复制稍低。

多 Master 多 Slave 模式的部署架构图&＃xff0c;如下所示&＃xff1a;

4 RocketMQ vs ActiveMQ vs Kafka

下面给出一张 RocketMQ、ActiveMQ 和 Kafka 的技术和特性的对比表&＃xff0c;表内容如下&＃xff1a;

5 Pull&Push模式

首先介绍一下 Push 和 Pull 两种消费模式&＃xff0c;内容如下&＃xff1a;

• Push 模式&＃xff1a;由消息中间件&＃xff08;MQ 消息服务器代理&＃xff09;主动地将消息推送给消费者。采用 Push 方式的情况下&＃xff0c;broker 可以尽可能实时地将消息发送给消费者进行消费&＃xff0c;但是&＃xff0c;在消费者的处理消息的能力较弱时&＃xff08;比如消费者端的业务系统处理一条消息的流程比较复杂、其中的调用链路比较多导致消费时间比较久&＃xff0c;概括起来就是“慢消费问题”&＃xff09;&＃xff0c;broker 不断地向消费者 Push 消息&＃xff0c;会导致消费者端的缓冲区溢出&＃xff0c;从而产生异常&＃xff1b;
• Pull 模式&＃xff1a;由消费者主动向消息中间件&＃xff08;MQ 消息服务器代理&＃xff09;拉取消息。采用 Pull 方式时&＃xff0c;重点是如何设置 Pull 消息的频率。例如&＃xff0c;生产者可能在 1 分钟内连续生产了 1000 条消息&＃xff0c;然后 2 小时内没有新消息产生&＃xff0c;在这种情况下&＃xff0c;如果每次 Pull 的时间间隔比较久&＃xff0c;就会增加消息的延迟&＃xff0c;即消息到达消费者的时间会加长&＃xff0c;MQ 中消息的堆积量变大&＃xff1b;反之&＃xff0c;如果每次 Pull 的时间间隔较短&＃xff0c;但是在一段时间内 MQ 中并没有任何消息可以消费&＃xff0c;那么又会产生很多无效的 Pull 请求的 RPC 开销&＃xff0c;影响 MQ 整体的网络性能&＃xff08;即“消息延迟与忙等待”&＃xff09;。

介绍完一般的 Push 与 Pull 消费方式后&＃xff0c;再来看一下 RocketMQ 的这两种消费方式&＃xff0c;内容如下&＃xff1a;

• RocketMQ 的 Pull 方式下&＃xff0c;Consumer 主动获取 MessageQueue 的 Set&＃xff08;集合&＃xff09;&＃xff0c;遍历该集合中的每一个队列&＃xff0c;发送 Pull 的请求&＃xff08;参数中带有队列中的消息偏移量&＃xff09;&＃xff0c;同时需要 Consumer 端自己保存消息消费的 offset&＃xff08;偏移量&＃xff09;至本地变量中。由此可见&＃xff0c;在 Pull 模式下&＃xff0c;需要业务应用自身去完成比较多的事情&＃xff0c;所以在实际应用中&＃xff0c;Pull 方式用的较少&＃xff1b;
• RocketMQ 的 Push 方式下&＃xff0c;Consumer 注册了一个监听器&＃xff0c;当 Consumer 收到消息时&＃xff0c;会主动调用这个监听器完成消费&＃xff0c;并进行相关的业务逻辑处理。由此可见&＃xff0c;在 Push 方式下&＃xff0c;业务应用代码只需要完成消息消费的代码逻辑即可&＃xff0c;无需参与 MQ 本身的一些任务处理。

说明&＃xff1a;RocketMQ 的 Push 方式本质上也属于 Pull 方式&＃xff0c;因为当 Consumer 从 broker 成功获取到消息后&＃xff0c;Consumer 需要调用监听器&＃xff0c;主动去 broker 轮询拉取消息完成消费。这种 Push 方式既解决了普通的 Push 方式的“慢消费问题”&＃xff0c;同时相对于纯 Pull 模式来说&＃xff0c;在代码实现上又简单了许多。

正如上面的说明所述&＃xff0c;RocketMQ 的消费方式&＃xff08;Pull 方式 和 Push 方式&＃xff09;本质上都是基于 Pull 方式的&＃xff0c;即都是采用 consumer 轮询从 broker 拉取消息。而在轮询过程中&＃xff0c;加入了一种长轮询机制&＃xff08;对普通轮询的一种优化&＃xff09;&＃xff0c;来平衡 Push/Pull 模型的各自缺点。长轮询机制的基本设计思路是&＃xff1a;消费者如果第一次尝试 pull 消息失败&＃xff08;如 Broker 端没有可以消费的消息&＃xff09;&＃xff0c;Broker 并不立即给消费者客户端返回 Response 响应消息&＃xff0c;而是先 hold 并挂起该请求&＃xff08;将请求保存至 pullRequestTable 本地缓存变量中&＃xff09;&＃xff0c;然后 Broker 端的后台独立线程 PullRequestHoldService 会从 pullRequestTable 本地缓存变量中不断地去取&＃xff0c;具体的做法是查询待拉取消息的偏移量是否小于消费队列最大偏移量&＃xff0c;如果条件成立则说明有新消息达到 Broker 端&＃xff08;这里&＃xff0c;在 RocketMQ 的 Broker 端会有一个后台独立线程 ReputMessageService 不停地构建 ConsumeQueue/IndexFile 数据&＃xff0c;同时取出 hold 住的请求并进行二次处理&＃xff09;&＃xff0c;则通过重新调用一次业务处理器 PullMessageProcessor 的处理请求方法 processRequest()&＃xff0c;来重新尝试拉取消息&＃xff08;此处&＃xff0c;每隔 5s 重试一次&＃xff0c;默认长轮询整体的时间设置为 30s&＃xff09;。

RocketMQ 使用的这种长轮询机制&＃xff08;Pull 方式和 Push 方式都具有&＃xff09;&＃xff0c;解决了一般的 Push 方式的“慢消费问题”&＃xff0c;同时&＃xff0c;解决了一般的 Pull 方式的“消息延迟与忙等待问题”&＃xff0c;并且&＃xff0c;使用 RocketMQ 的 Push 机制&＃xff0c;还可以减小消费者端的代码逻辑复杂度&＃xff0c;所以 RocketMQ 的 Push 方式同时具有三个优点。

综上所述&＃xff0c;在实际应用中&＃xff0c;我们一般会采用 RocketMQ 的 Push 方式进行消息消费。

关于 RocketMQ 具体的部署方法&＃xff0c;请参考本系列文章的第二篇。