RocketMQ实战（三）：分布式事务

以你之姓@ 2022-06-13 03:24 216阅读 0赞

转自 http://www.jianshu.com/p/53324ea2df92

#### 本篇博客主要讨论的话题是：顺序消费、RMQ在分布式事务中的应用等。    ####

# 关于多Master多Slave的说明 #

由于在之前的博客中已经搭建了双Master，其实多Master多Slave大同小异，因此这里并不会一步步的演示搭建多Master多Slave，而是从思路上，分析下重点应该注意的配置项。

![1240][]

多Master多Slave

> 第一，这四台机器，对外是一个统一的整体，是一个rocketmq cluster，因此需要brokerClusterName保持统一
> 
> 第二，123机器是121的从，124机器是122的从，如何在配置中体现？ 主和从的brokerName需要保持一致，另外brokerId标示了谁是主，谁是从（brokerId=0的就是主，大于0的就是从）
> 
> 第三，注意namesrvAddr的地址是4台NameServer
> 
> 第四，配置项中brokerRole需要指明 ASYNC\_MASTER（异步复制Master） or SYNC\_MASTER（同步双写Master） or SLAVE（从）
> 
> 第五，和以前的多Master启动方式一致，先启动4台Namesrv，然后用指定配置文件的方式启动Master/Slave即可
> 
> 第六，多Master多Slave的好处在于，即便集群中某个broker挂了，也可以继续消费，保证了实时性的高可用，但是并不是说某个master挂了，slave就可以升级master，开源版本的rocketmq是不可以的。也就是说，在这种情况下，slave只能提供读的功能，将失去消息负载的能力。

#     #

# Queue in Topic #

> 对于RocketMQ而言，Topic只是一个逻辑上的概念，真正的消息存储其实是在Topic中的Queue中。想一想，为什么RocketMQ要这要设计呢？其实是为了消息的顺序消费，后文中将为大家介绍。

![1240 1][]

queue in topic

![1240 2][]

默认一个Topic中4个队列

![1240 3][]

配置文件中指定

# 初步认识RocketMQ的核心模块    #

![1240 4][]

rocketmq模块

> rocketmq-broker：接受生产者发来的消息并存储（通过调用rocketmq-store），消费者从这里取得消息。
> 
> rocketmq-client：提供发送、接受消息的客户端API。
> 
> rocketmq-namesrv：NameServer，类似于Zookeeper，这里保存着消息的TopicName，队列等运行时的元信息。（有点NameNode的味道）
> 
> rocketmq-common：通用的一些类，方法，数据结构等
> 
> rocketmq-remoting：基于Netty4的client/server + fastjson序列化 + 自定义二进制协议
> 
> rocketmq-store：消息、索引存储等
> 
> rocketmq-filtersrv：消息过滤器Server，需要注意的是，要实现这种过滤，需要上传代码到MQ！【一般而言，我们利用Tag足以满足大部分的过滤需求，如果更灵活更复杂的过滤需求，可以考虑filtersrv组件】
> 
> rocketmq-tools：命令行工具

# Order Message #

> RocketMQ提供了3种模式的Producer：
> 
> NormalProducer（普通）、OrderProducer（顺序）、TransactionProducer（事务）
> 
> 在前面的博客当中，涉及的都是NormalProducer，调用传统的send方法，消息是无序的。接下来，我们来看看顺序消费。模拟这样一个场景，如果一个用户完成一个订单需要3条消息，比如订单的创建、订单的支付、订单的发货，很显然，同一个用户的订单消息必须要顺序消费，但是不同用户之间的订单可以并行消费。

生产者端代码示例：

![1240 5][]

顺序消息模式

> 注意，一个Message除了Topic/Tag外，还有Key的概念。
> 
> 上图的send方法不同于以往，有一个MessageQueueSelector，将用于指定特定的消息发往特定的队列当中！

![1240 6][]

顺序消费

> 注意在以前普通消费消息时设置的回调是MessageListenerConcurrently，而顺序消费的回调设置是MessageListenerOrderly。

当我们启动2个Consumer进行消费时，可以观察到：

![1240 7][]

多个消费者消费的结果

> 可以观察得到，虽然从全局上来看，消息的消费不是有序的，但是每一个订单下的3条消息是顺序消费的！
> 
> 其实，如果需要保证消息的顺序消费，那么很简单，首先需要做到一组需要有序消费的消息发往同一个broker的同一个队列上！其次消费者端采用有序Listener即可。
> 
> 这里，RocketMQ底层是如何做到消息顺序消费的，看一看源码你就能大概了解到，至少来说，在多线程消费场景下，一个线程只去消费一个队列上的消息，那么自然就保证了消息消费的顺序性，同时也保证了多个线程之间的并发性。也就是说其实broker并不能完全保证消息的顺序消费，它仅仅能保证的消息的顺序发送而已！

关于多线程消费这块，RocketMQ早就替我们想好了，这样设置即可：

![1240 8][]

消费多线程设置

> 想一想，在ActiveMQ中，我们如果想实现并发消费的话，恐怕还得搞个线程池提交任务吧，RocketMQ让我们的工作变得简单！

# Transaction Message #

> 在说事务消息之前，我们先来说说分布式事务的那些事！
> 
> 什么是分布式事务，我的理解是一半事务。怎么说，比如有2个异构系统，A异构系统要做T1，B异构系统要做T2，要么都成功，要么都失败。
> 
> 要知道异构系统，很显然，不在一个数据库实例上，它们往往分布在不同物理节点上，本地事务已经失效。

![1240 9][]

2阶段提交

> 2阶段提交协议，Two-Phase Commit，是处理分布式事务的一种常见手段。2PC，存在2个重要角色：事务协调器（TC），事务执行者。
> 
> 2PC，可以看到节点之间的通信次数太多了，时间很长！时间变长了，从而导致，事务锁定的资源时间也变长了，造成资源等待时间变长！在高并发场景下，存在严重的性能问题！

下面，我们来看看MQ在高并发场景下，是如何解决分布式事务的。

> 考虑生活中的场景：
> 
> 我们去北京庆丰包子铺吃炒肝，先去营业员那里付款（Action1），拿到小票（Ticket），然后去取餐窗口排队拿炒肝（Action2）。思考2个问题：第一，为什么不在付款的同时，给顾客炒肝？如果这样的话，会增加处理时间，使得后面的顾客等待时间变长，相当于降低了接待顾客的能力（降低了系统的QPS）。第二，付了款，拿到的是Ticket，顾客为什么会接受？从心理上说，顾客相信Ticket会兑现炒肝。事实上也是如此，就算在最后炒肝没了，或者断电断水（系统出现异常），顾客依然可以通过Ticket进行退款操作，这样都不会有什么损失！（虽然这么说，但是实际上包子铺最大化了它的利益，如果炒肝真的没了，浪费了顾客的时间，不过顾客顶多发发牢骚，最后接受）
> 
> 生活已经告诉我们处理分布式事务，保证数据最终一致性的思路！这个Ticket（凭证）其实就是消息！

![1240 10][]

业务和消息生成耦合在一起

> 业务操作和消息的生成耦合在一起，保证了只要A银行的账户发生扣款，那么一定会生成一条转账消息。只要A银行系统的事务成功提交，我们可以通过实时消息服务，将转账消息通知B银行系统，如果B银行系统回复成功，那么A银行系统可以在table中设置这条转账消息的状态。
> 
> 这样耦合的方式，从架构上来看，就有点不太优雅，而且存在一些问题。比如说，消息的存储实质上是在A银行系统中的，如果A银行系统出了问题，将导致无法转账。如果解耦，将消息独立出来呢？

![1240 11][]

业务和消息解耦

> 如上图所示，消息数据独立存储，业务和消息解耦，实质上消息的发送有2次，一条是转账消息，另一条是确认消息。

到这里，我们先来看看基于RocketMQ的代码：

![1240 12][]

生产者示例代码

> 生产者这里用到是：TransactionMQProducer。
> 
> 这里涉及到2个角色：本地事务执行器（代码中的TransactionExecuterImpl）、服务器回查客户端Listener（代码中的TransactionCheckListener）。
> 
> 如果事务消息发送到MQ上后，会回调  本地事务执行器；但是此时事务消息是prepare状态，对消费者还不可见，需要  本地事务执行器  返回RMQ一个确认消息。

![1240 13][]

本地事务执行器

> 事务消息是否对消费者可见，完全由事务返回给RMQ的状态码决定（状态码的本质也是一条消息）。

![1240 14][]

回查Listener

![1240 15][]

运行结果

> 生产者发送了2条消息给RMQ，有一条本地事务执行成功，有一条本地事务执行失败。
> 
> 2条业务消息 + 2条确认消息  因此是4条；
> 
> 注意到消费者只消费了一条数据，就是只有告诉RMQ本地事务执行成功的那条消息才会被消费！因此是1条！
> 
> 但是，注意到本地事务执行失败的消息，RMQ并没有check listener？这是为什么呢？因为RMQ在3.0.8的时候还是支持check listener回查机制的，但是到了3.2.6的时候将事务回查机制“阉割”了！
> 
> 那么3.0.8的时候，RMQ是怎么做事务回查的呢？看一看源码，你会知道，其实事务消息开始是prepare状态，然后RMQ会将其持久化到MySQL当中，然后如果收到确认消息，就删除掉这条prepare消息，如果迟迟收不到确认消息，那么RMQ会定时的扫描prepare消息，发送给produce group进行回查确认！
> 
> 到这里，问题来了，要知道3.2.6版本，没有回查机制了，会存在问题么？
> 
> 当然会存在问题！假设，我们发送一条转账事务消息给RMQ，成功后回调本地事务，DB减操作成功，刚准备给RMQ一个确认消息，此时突然断电，或者网络抖动，使得这条确认消息没有发送出去。此时RMQ中的那条转账事务消息，始终处于prepare状态，消费者读取不到，但是却已经完成一方的账户资金变动！！！

既然，RMQ3.2.6版本不为我们进行回查，那么只能由我们自己完成了。具体怎么做呢，咱们下期再来分析~

see u , good night~

[1240]: /images/20220613/36f4919e2299461398e44ef1da5b8c82.png
[1240 1]: /images/20220613/479113cb7f38495882b1b82166d03958.png
[1240 2]: /images/20220613/db387cdf469440e4b0a9e8ec34f639d1.png
[1240 3]: /images/20220613/001820aa02fe438eb3eb668bc0cc8e83.png
[1240 4]: /images/20220613/b9049dcfc91943fb836aa52b8b1ed01b.png
[1240 5]: /images/20220613/b1594d0b9b8b4356b20b4712b1034754.png
[1240 6]: /images/20220613/dc2c590512bf4edbad4553aa44fe1031.png
[1240 7]: /images/20220613/a59c20e989cf4d3eb85ffb36d8d9f8d8.png
[1240 8]: /images/20220613/9c026ae98cc646fd8cee614bbf42bef3.png
[1240 9]: /images/20220613/f8d7cb6b93d34cea9ce10e4fb73cc0c2.png
[1240 10]: /images/20220613/e4f103eb864741dfb6efb5bbf3ac6626.png
[1240 11]: /images/20220613/db3042ad2afb46c4aa337f2768454d5d.png
[1240 12]: /images/20220613/b863fa1e2b194a76b99300b756c6f4bc.png
[1240 13]: /images/20220613/fc459e4ece3147709028f72e38445c81.png
[1240 14]: /images/20220613/70eed4c6f8994ca8a514c007f19b1058.png
[1240 15]: /images/20220613/fae9b63625674223885b9250bd1590a0.png