MQ系列之ROCKETMQ

RocketMQ介绍

RocketMQ是阿里巴巴开源的一个消息中间件，在阿里内部历经了双十一等很多高并发场景的考验，能够处理亿万级别的消息。2016年开源后捐赠给Apache，现在是Apache的一个顶级项目。

早期阿里使用ActiveMQ，但是，当消息开始逐渐增多后，ActiveMQ的IO性能很快达到了瓶颈。于是，阿里开始关注Kafka。但是Kafka是针对日志收集场景设计的，他的高级功能并不是很贴合阿里的业务场景。尤其当他的Topic过多时，由于Partition文件也会过多，这就会加大文件索引的耗时，会严重影响IO性能。于是阿里才决定自研中间件，最早叫做MetaQ，后来改名成为RocketMQ。最早他所希望解决的最大问题就是多Topic下的IO性能压力。但是产品在阿里内部的不断改进，RocketMQ开始体现出一些不一样的优势。

主流MQ对比

	优点	缺点	适用场景
Kafka	吞吐量非常大，性能非常好，技术生态完整	功能比较单一	分布式日志收集，大数据采集
RabbitMQ	消息可靠性高，功能全面	吞吐量较低。消息积压会影响性能。 erlang语言比较小众	企业内部系统调用
RocketMQ	高吞吐、高性能、高可用，高级功能非常全	技术生态相对没有那么完整	几乎全场景。尤其适合金融场景

其中RocketMQ，孵化自阿里巴巴。历经阿里多年双十一的严格考验，RocketMQ可以说是从全世界最严苛的高并发场景中摸爬滚打出来的过硬产品，也是少数几个在金融场景比较适用的MQ产品。

从横向对比来看，RocketMQ与Kafka和RabbitMQ相比。RocketMQ的消息吞吐量虽然和Kafka相比还是稍有差距，但是却比RabbitMQ高很多。

在阿里内部，RocketMQ集群每天处理的请求数超过5万亿次，支持的核心应用超过3000个。而RocketMQ最大的优势就是他天生就为金融互联网而生。

他的消息可靠性相比Kafka也有了很大的提升，而消息吞吐量相比RabbitMQ也有很大的提升。

另外，RocketMQ的高级功能也越来越全面，广播消费、延迟队列、死信队列等等高级功能一应俱全，甚至某些业务功能比如事务消息，已经呈现出领先潮流的趋势。

运行架构

img.png

RocketMQ的架构主要分为三大部分，分别是NameServer、Broker和Client。

NameServer

NameServer是RocketMQ的名称服务，它主要负责集群的路由信息管理。NameServer是无状态的，可以部署多个实例来实现高可用。

Broker

Broker是RocketMQ的核心组件，它主要负责消息的存储和转发。Broker可以分为主从两种角色，主Broker负责处理客户端的读写请求，从Broker负责同步主Broker的数据，实现高可用。

Client

Client是RocketMQ的客户端组件，它主要负责与Broker进行通信，发送和接收消息。Client可以分为生产者和消费者两种角色，生产者负责发送消息，消费者负责接收消息。

消息模型

image.png

RocketMQ的消息模型主要包括主题（Topic）、队列（Queue）和消息（Message）。

生产者和消费者都可以指定一个Topic发送消息或者拉取消息。而Topic是一个逻辑概念。

Topic中的消息会分布在后面多个MessageQueue当中。这些MessageQueue会分布到一个或者多个broker中。

MessageQueue是物理概念，是消息存储的基本单元。每个MessageQueue其实就是一个FIFO队列数据结构，生产者发送的消息会被追加到MessageQueue的末尾，消费者则从MessageQueue的头部开始消费消息。

MessageQueue类似kafka的Partition概念，一个Topic可以有多个MessageQueue，这样就可以实现消息的并行处理，提高系统的吞吐量。

同一时间，只能有一个消费者实例消费同一个MessageQueue中的消息。

消息确认机制

RocketMQ要支持互联网金融场景，那么消息安全是必须优先保障的。而消息安全有两方面的要求，一方面是生产者要能确保将消息发送到Broker上。另一方面是消费者要能确保从Broker上争取获取到消息。

1、生产端确认消息正常发送到RocketMQ

消息生产者发送消息分为三种方式

同步发送，生产者发送消息后会阻塞等待Broker的响应，直到收到Broker的确认消息或者超时
异步发送，生产者发送消息后不会阻塞等待Broker的响应，而是通过回调函数来处理Broker的确认消息
单向发送，生产者发送消息后不会等待Broker的响应，也不会进行任何确认

采用消息确认加多次重试的机制保证消息能够可靠发送到Broker上。但并不保证消息一定能被消费者消费到，因为消费者可能会因为各种原因消费失败。

2、消费端确认消息正常消费

我们之前分析生产者的可靠性问题，核心的解决思路就是通过确认Broker端的状态来保证生产者发送消息的可靠性。对于RocketMQ的消费者来说，保证消息处理可靠性的思路也是类似的。只不过这次换成了Broker等待消费者返回消息处理状态。

consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

这个返回值是一个枚举值，有两个选项 CONSUME_SUCCESS和RECONSUME_LATER。如果消费者返回CONSUME_SUCCESS，那么消息自然就处理结束了。

但是如果消费者没有处理成功，返回的是RECONSUME_LATER，Broker就会过一段时间再发起消息重试。

如果消费者一直处理失败，那么消息就会一直被重试，直到达到最大重试次数，默认是16次，就会进入死信队列。

广播消息

广播模式和集群模式是RocketMQ的消费者端处理消息最基本的两种模式。

集群模式下，一个消息，只会被一个消费者组中的多个消费者实例共同处理一次。广播模式下，一个消息，则会推送给所有消费者实例处理，不再关心消费者组。

默认模式(也就是集群模式)下，Broker端会给每个ConsumerGroup维护一个统一的Offset，这个Offset可以保证一个消息，在同一个ConsumerGroup内只会被消费一次。

而广播模式的实现方式，是将Offset转移到消费者端自行保管，这样Broker端只管向所有消费者推送消息，而不用负责维护消费进度。

顺序消息机制

Rocketmq由于存在多个MessageQueue的设计，所以天然是不保证消息顺序的。因为同一个Topic下的消息会被分布到多个MessageQueue中，而每个MessageQueue是独立的FIFO队列，消费者是并行消费多个MessageQueue的消息，所以无法保证消息的全局顺序。

要解决这个问题思路也很简单，那就是将需要顺序处理的消息全部发送到同一个MessageQueue中。这样消费者在消费这个MessageQueue的消息时，就能保证消息的顺序性。

过滤消息

同一个Topic下有多种不同的消息，消费者只希望关注某一类消息。

例如，某系统中给仓储系统分配一个Topic，在Topic下，会传递过来入库、出库等不同的消息，仓储系统的不同业务消费者就需要过滤出自己感兴趣的消息，进行不同的业务操作。

img_2.png

包含两种过滤方式，标签过滤和SQL过滤。

事务消息

事务消息是RocketMQ非常有特色的一个高级功能。他的基础诉求是通过RocketMQ的事务机制，来保证上下游的数据一致性。

以电商为例，用户支付订单这一核心操作的同时会涉及到下游物流发货、积分变更、购物车状态清空等多个子系统的变更。这种场景，非常适合使用RocketMQ的解耦功能来进行串联。

img_3.png

考虑到事务的安全性，即要保证相关联的这几个业务一定是同时成功或者同时失败的。如果要将四个服务一起作为一个分布式事务来控制，可以做到，但是会非常麻烦。

而使用RocketMQ在中间串联了之后，事情可以得到一定程度的简化。由于RocketMQ与消费者端有失败重试机制，所以，只要消息成功发送到RocketMQ了，那么可以认为Branch2.1，Branch2.2，Branch2.3这几个分支步骤，是可以保证最终的数据一致性的。

这样，一个复杂的分布式事务问题，就变成了MinBranch1和Branch2两个步骤的分布式事务问题。

然后，在此基础上，RocketMQ提出了事务消息机制，采用两阶段提交的思路，保证Main Branch1和Branch2之间的事务一致性。

img_4.png

具体的实现思路是这样的：

img_5.png

流程如下：

生产者将消息发送至Apache RocketMQ服务端。
Apache RocketMQ服务端将消息持久化成功之后，向生产者返回Ack确认消息已经发送成功，此时消息被标记为"暂不能投递"，这种状态下的消息即为半事务消息。
生产者开始执行本地事务逻辑。
生产者根据本地事务执行结果向服务端提交二次确认结果（Commit或是Rollback），服务端收到确认结果后处理逻辑如下：
二次确认结果为Commit：服务端将半事务消息标记为可投递，并投递给消费者。
二次确认结果为Rollback：服务端将回滚事务，不会将半事务消息投递给消费者。
在断网或者是生产者应用重启的特殊情况下，若服务端未收到发送者提交的二次确认结果，或服务端收到的二次确认结果为Unknown未知状态，经过固定时间后，服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查。
生产者收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
生产者根据检查到的本地事务的最终状态再次提交二次确认，服务端仍按照步骤4对半事务消息进行处理。

MQ系列之ROCKETMQ