分布式事务学习笔记 2020-02-25

分布式事务学习项目：流量充值中心
git地址：https://github.com/barrywangmeng/data-refill-center

以流量充值业务为模型，来模拟分布式事务的实现。

从最开始的单库到后面拆分成多库，从2pc到3pc，从springboot到springcloud，一步步模拟分布式事务中遇到的各种问题。

分布式事务参见的集中方案

• XA分布式事务：一般用于单系统多库的场景
• TCC方案：try-confirm-cancel方案, 微服务链式调用场景
• 可靠消息最终一致性方案：针对于耗时的请求操作，使用可靠性消息异步解耦
• 最大努力通知方案：适用于耗时且非核心业务，例如下单成功发送短信、发送通知等操作

技术栈&知识点

• SpringBoot
• SpringCloud
• XA规范
• 分布式事务2PC、3PC原理
• JTA、Atomikos框架
• TCC事务
• 可靠消息MQ
  ……

版本

• master：springboot版本，单体应用多数据库版本，使用Atomikos框架支持XA规范分布式事务
• cloud：springcloud版本，使用tcc分布式事务

cloud版本如图，需要自己导入一个个项目：

image.png

流量充值中心的整体架构设计

流量充值中心的整体架构设计.jpg

流量充值中心运转流程

02_流量充值中心运转流程.jpg

SpringBoot实现流量中心拆库操作

数据库拆分如下：

image.png

SpringBoot多数据源配置：(详情见application.yml配置文件)

 1activity:
 2  datasource:
 3    type: com.alibaba.druid.pool.DruidDataSource
 4    url: jdbc:mysql://127.0.0.1:3306/data-refill-center-activity?useUnicode=true&characterEncoding=utf-8
 5    username: root
 6    password: 123456
 7    driverClassName: com.mysql.jdbc.Driver
 8    initialSize: 10
 9    minIdle: 50
10    maxActive: 500
11    maxWait: 60000 
12    timeBetweenEvictionRunsMillis: 60000  
13    minEvictableIdleTimeMillis: 300000
14    validationQuery: SELECT 1 FROM DUAL
15    testWhileIdle: true
16    testOnBorrow: false  
17    testOnReturn: false  
18    poolPreparedStatements: true 
19    maxPoolPreparedStatementPerConnectionSize: 20
20    filters: stat,wall,log4j
21    connectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000

Spring声明式事务

Spring通过@Transaction 注解来实现声明式事务。
原理如下图：

image.png

分布式事务初探

分布式事务是指挥操作多个数据库的事务，这里可以先来说下XA规范:

XA是由X/Open组织提出的分布式事务的规范。
XA规范主要定义了（全局）事务管理器（Transaction Manager）和（局部）资源管理器（Resource Manager）之间的接口。
XA接口是双向的系统接口，在事务管理器（Transaction Manager）以及一个或多个资源管理器（Resource Manager）之间形成通信桥梁。
XA引入的事务管理器充当上文所述全局事务中的“协调者”角色。事务管理器控制着全局事务，管理事务生命周期，并协调资源。
资源管理器负责控制和管理实际资源（如数据库或JMS队列）。

上面概念有些抽象，我们只需要知道X/Open的组织定义了分布式事务的模型，这里面有几个角色：

1. AP（Application，应用程序）：我们自己的系统
2. TM（Transaction Manager，事务管理器）：一个在系统里嵌入的一个专门管理横跨多个数据库的事务的一个组件
3. RM（Resource Manager，资源管理器）：数据库（比如MySQL）
4. CRM（Communication Resource Manager，通信资源管理器）:可以是消息中间件（但是也可以不用这个东西）

XA就是定义好的那个TM与RM之间的接口规范，就是管理分布式事务的那个组件跟各个数据库之间通信的一个接口

2PC分布式事务

Two-Phase-Commitment-Protocol，两阶段提交协议

X/Open组织定义的一套分布式事务的模型，我们可以按照这个模型去实现分布式事务。

如果我们明白了XA原理也也就等于明白了2PC原理了。2PC是基于XA规范搞的一套分布式事务的理论，也可以叫做一套规范，或者是协议。

原理如下：

（1）准备阶段
TM先发送个prepare消息给各个数据库，让各个库先把分布式事务里要执行的各种操作，先准备执行，其实此时各个库会差不多先执行好，不提交事务

prepare消息发送后，各个库先在本地开个事务，然后执行好SQL，这里不提交事务，但是会有对应的日志记录

然后各个数据库都返回一个响应消息给事务管理器，如果成功了就发送一个成功的消息，如果失败了就发送一个失败的消息

（2）提交阶段

第一种情况，要是TM发现某个数据库返回失败，此时TM直接判定这个分布式事务失败，然后TM通知所有数据库，全部回滚。反之，则通知所有数据库提交事务。

原理图如下：（结合流量充值项目代码）

04_2PC协议.jpg

2PC缺陷

1. 同步阻塞：在阶段一里执行prepare操作会占用资源，一直到整个分布式事务完成，才会释放资源，这个过程中，如果有其他人要访问这个资源，就会被阻塞住
2. 单点故障：TM是个单点，一旦挂掉就完蛋了
3. 事务状态丢失：即使把TM做成一个双机热备的，一个TM挂了自动选举其他的TM出来，但是如果TM挂掉的同时，接收到commit消息的某个库也挂了，此时即使重新选举了其他的TM
   ，压根儿不知道这个分布式事务当前的状态，因为不知道哪个库接收过commit消息，哪个接收过commit消息的库也挂了
4. 脑裂问题：在阶段二中，如果发生了脑裂问题，那么就会导致某些数据库没有接收到commit消息，有些库收到了commit消息，结果有些库没有收到

具体如图：

2PC存在的问题.jpg

3PC 分布式事务

Three-Phase-Commitment 三阶段提交协议

主要是针对2PC的一些问题做了改进，改变成下面的过程：

1、 CanCommit阶段
这个阶段就是TM发送一个CanCommit消息给各个数据库，各个数据库返回个结果，这里并不直接执行SQL语句，只是告诉TM我的网络是通畅的。

2、PreCommit阶段
如果各个数据库对CanCommit返回都是成功，那么就进入PreCommit阶段，这个时候各个数据库会执行SQL语句，但是并不提交事务

3、DoCommit阶段
如果PreCommit都返回成功，那么各个数据库进入DoCommit阶段，TM给各个数据库发送提交事务指令。反之则发送abort消息给各个数据库。

跟2PC相比，主要是有两点改进：

1. 引入CanCommit阶段
2. DoCommit阶段引入超时机制，如果PreCommit阶段返回都是成功，如果超时时间到了，数据库还没有收到DoCommit或者abort指令则会自己提交事务

3PC存在的问题：
1、脑裂问题：如果TM发送abort消息，但是某个数据库由于网络原因没有接收到，那么这个数据库过了一定时间后就会自己提交事务

执行原理图：

3PC执行原理.jpg

Mysql 对分布式事务的支持

Mysql是支持XA分布式事务的，而且支持2PC的协议。具体代码可参见com.barrywang.data.refill.center.MySQLXATest.java

原理图如下：

Mysql XA分布式事务实现.jpg

JTA事务原理

Java Transaction API 一套分布式事务的编程API

如果是跨多个库的事务，是通过JTA API来支持的，通过JTA API可以协调和管理横跨多个数据库的分布式事务，一般来说会结合JNDI。

JTA&Atomikos

Atomikos：客户端的TM第三方库，基于Mysql的XA API来实现分布式事务

基于TCC分布式事务

TCC：Try-Cancle-Commit 三个阶段

TCC方案细节.png

springcloud、springboot或者dubbo项目都可以使用开源框架bytetcc(https://github.com/liuyangming/ByteTCC) 来实现分布式事务
推荐一篇文章： bytetcc原理(https://www.cnblogs.com/jajian/p/10014145.html)

可靠消息最终一致性方案

实现原理如图：

可靠消息最终一致性方案.jpg

可靠消息最终一致性方案涉及到4个组件：

1. 上游服务：发送MQ消息通知下游服务执行某个操作
2. 可靠消息服务：协调上下游服务的消息传递，确保数据一致性，可以认为这个所谓的可靠消息服务是我们自己开发的，也是一个spring cloud的服务，只不过这个服务是通用的，是所有服务所有系统都基于这个可靠消息服务来实现可靠消息最终一致性的方案。“可靠消息”四个字，这一切都是基于可靠消息服务来做的，方案设计，消息如何保持可靠性
3. MQ消息中间件：这个一般是RocketMQ、RabbitMQ或者Kafka
4. 下游服务：就是那个要被调用的服务

所谓的分布式事务，上游服务他要执行一个本地的数据库操作，下游服务也要执行一个本地的数据库操作，现在尽量就是希望是说上游服务和下游服务的数据库操作要么同时完成，要么同时不完成。

具体的执行流程如下所示：

（1）上游服务发送一个待确认消息给可靠消息服务

（2）可靠消息服务将这个待确认的消息保存到自己本地数据库里，保存起来，但是不发给MQ，这个时候消息的状态是“待确认”

（3）上游服务操作本地数据库

（4）上游服务根据自己操作本地数据库的结果，来通知可靠消息服务，可以确认发送消息了，或者是删除消息。
操作完本地数据库之后，会有两个结果，第一个结果是操作失败了，第二个结果是操作成功了，如果本地数据库操作失败了，本地操作会回滚，回滚之后，上游服务就要通知可靠消息服务删除消息；如果本地数据库操作成功了，那么此时本地事务就提交了，接着就可以通知可靠消息服务发送消息

（5）可靠消息服务将这个消息的状态修改为“已发送”，并且将消息发送到MQ中间件里去
这个环节是必须包裹在一个事务里的，如果发送MQ失败报错，那么可靠消息服务更新本地数据库里的消息状态为“已发送”的操作也必须回滚，反之如果本地数据库里的消息状态为“已发送”，那么必须成功投递消息到MQ里去

@Transactional
public void confirmMessage(Long messageId) {
messageDAO.updateStatus(messageId, MessageStatus.SENT);
rabbitmqProducer.send(message);
}

如果更新数据库里的消息状态报错了，那么消息根本不会投递到MQ里去；如果更新数据库里的消息状态成功了，但是事务还没提交，然后将消息投递到MQ里去报错了，此时事务管理器会感知到这个异常，然后会直接回滚掉整个事务，更新数据库里消息状态的操作也会回滚掉的

就可以保证说，更新数据库里的消息状态和投递消息到MQ，要么一起成功，要么一起失败，这里这第5个步骤，必须是一起成功或者是一起失败

MQ，rabbitmq，都有事务消息的一个实现，你可以先去投递一个prepare的消息，接着如果说数据库操作成功过了，那么就commit那个消息发送给rabbitmq，然后如果数据库操作失败了，就通知mq去rollback一条消息

但是MQ的事务消息最好别轻易用，因为那个性能实在是太低了，吞吐量太差

所以说我这里给大家介绍的是上面的那种方案

（6）下游服务从MQ里监听到一条消息

（7）下游服务根据消息，在自己本地操作数据库

（8）下游服务对本地数据库操作完成之后，对MQ进行ack操作，确认这个消息处理成功

现在的MQ中间件，无论是rabbitmq、rocketmq、kafka，都是支持手动ack。如果你是使用的默认自动ack的模式，那么就会导致消息的丢失；现在一般都会用手动ack，当本地操作执行成功之后，再对MQ执行手动的ack确认

只有当我手动ack确认之后，mq才会删除消息

如果我还没ack，本地数据库比如操作失败报错了，此时MQ一直没收到ack消息，会怎么样呢？此时MQ会保证重新投递一次消息，可以给其他的消费者实例去消费

（9）下游服务对MQ进行ack之后，再给可靠消息服务发送个请求，通知该服务说，ok，我这里处理完毕了，可靠消息服务收到通知之后，将消息的状态修改为“已完成”