RocketMQ高级特性①

一、消息存储

1. 存储介质

• 关系型数据库

ActiveMQ（默认采用的KahaDB做消息存储）可选用JDBC的方式来做消息持久化，通过简单的xml配置信息即可实现JDBC消息存储。由于，普通关系型数据库（如Mysql）在单表数据量达到千万级别的情况下，其IO读写性能往往会出现瓶颈。在可靠性方面，该种方案非常依赖DB，如果一旦DB出现故障，则MQ的消息就无法落盘存储会导致线上故障。

• 文件

目前业界较为常用的几款产品（RocketMQ/Kafka/RabbitMQ）均采用的是消息刷盘至所部署虚拟机/物理机的文件系统来做持久化（刷盘一般可以分为异步刷盘和同步刷盘两种模式）。消息刷盘为消息存储提供了一种高效率、高可靠性和高性能的数据持久化方式。除非部署MQ机器本身或是本地磁盘损坏，否则一般是不会出现无法持久化的故障问题。

2. 性能对比

文件系统 > 关系型数据库DB

二、负载均衡

RocketMQ中的负载均衡都在Client端完成，具体来说的话，主要可以分为Producer端发送消息时候的负载均衡和Consumer端订阅消息的负载均衡。

Producer的负载均衡

如图所示，5 个队列可以部署在一台机器上，也可以分别部署在 5台不同的机器上，发送消息通过轮询队列的方式 发送，x每个队列接收平均的消息量。通过增加机器，可以水平扩展队列容量。 另外也可以自定义方式选择发往哪个队列。

Consumer的负载均衡

如图所示，如果有 5 个队列，2 个 consumer，那么第一个Consumer 消费 3 个队列，第二consumer 消费 2 个队列。 这样即可达到平均消费的目的，可以水平扩展 Consumer 来提高消费能力。但是 Consumer 数量要小于等于队列数 量，如果 Consumer超过队列数量，那么多余的Consumer 将不能消费消息 。

三、事务消息

Apache RocketMQ在4.3.0版中已经支持分布式事务消息，这里RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如下图所示。

事务消息发送步骤如下：

1. 生产者将半事务消息发送至消息队列RocketMQ服务端。
2. 消息队列RocketMQ服务端将消息持久化成功之后，向生产者返回Ack确认消息已经发送成功，此时消息为半事务消息。
3. 生产者开始执行本地事务逻辑。
4. 生产者根据本地事务执行结果向服务端提交二次确认结果（Commit或是Rollback），服务端收到确认结果后处理逻辑如下：

• 二次确认结果为Commit：服务端将半事务消息标记为可投递，并投递给消费者。
• 二次确认结果为Rollback：服务端不会将该消息投递给消费者，并按照如下逻辑进行回查处理。

5. 事务消息回查步骤如下：

• 在断网或者是生产者应用重启的特殊情况下，上述步骤4提交的二次确认最终未到达服务端，经过固定时间后，服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查。
• 生产者收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
• 生产者根据检查得到的本地事务的最终状态再次提交二次确认，服务端仍按照步骤4对半事务消息进行处理。

我正在参与2024腾讯技术创作特训营最新征文，快来和我瓜分大奖！