分布式事务

学习：https://blog.csdn.net/hancoder/article/details/120213532
demo: https://gitee.com/Linging241/distributed-transaction.git

1、分布式事务的产生场景

1. 典型的场景就是微服务架构 微服务之间通过远程调用完成事务操作。 比如：订单微服务和库存微服务，下单的同时订单微服务请求库存微服务减库存。 简言之：跨JVM进程产生分布式事务。
   

2. 单体系统访问多个数据库实例 当单体系统需要访问多个数据库（实例）时就会产生分布式事务。 比如：用户信息和订单信息分别在两个MySQL实例存储，用户管理系统删除用户信息，需要分别删除用户信息及用户的订单信息，由于数据分布在不同的数据实例，需要通过不同的数据库链接去操作数据，此时产生分布式事务。 简言之：跨数据库实例产生分布式事务。

   

3. 多服务访问同一个数据库实例 比如：订单微服务和库存微服务即使访问同一个数据库也会产生分布式事务，原因就是跨JVM进程，两个微服务持有了不同的数据库链接进行数据库操作，此时产生分布式事务。

2、分布式事务解决方案

2.1 2PC（两阶段提交）

2.1.1 什么是2PC

2PC即两阶段提交协议，是将整个事务流程分为两个阶段，准备阶段（Prepare phase）、提交阶段（commit phase），2是指两个阶段，P是指准备阶段，C是指提交阶段。

注意这只是协议或者说是理论指导，只阐述了大方向，具体落地还是有会有差异的。

数据库支持的2pc【2二阶段提交】，又叫做XA Transactions

2PC（Two-phase commit protocol），中文叫二阶段提交。 二阶段提交是一种强一致性设计，2PC 引入一个事务协调者的角色来协调管理各参与者（也可称之为各本地资源）的提交和回滚，二阶段分别指的是准备（投票）和提交两个阶段。

在计算机中部分关系数据库如Oracle、MySQL支持两阶段提交协议，如下图：

1. 准备阶段（Prepare phase）：事务管理器给每个参与者发送Prepare消息，每个数据库参与者在本地执行事务，并写本地的Undo/Redo日志，此时事务没有提交。（Undo日志是记录修改前的数据，用于数据库回滚，Redo日志是记录修改后的数据，用于提交事务后写入数据文件）

2. 提交阶段（commit phase）：如果事务管理器收到了参与者的执行失败或者超时消息时，直接给每个参与者发送回滚(Rollback)消息；否则，发送提交(Commit)消息；参与者根据事务管理器的指令执行提交或者回滚操作，并释放事务处理过程中使用的锁资源。注意:必须在最后阶段释放锁资源。

如果任一资源管理器在第一阶段返回准备失败，那么事务管理器会要求所有资源管理器在第二阶段执行回滚操作。通过事务管理器的两阶段协调，最终所有资源管理器要么全部提交，要么全部回滚，最终状态都是一致的。

2PC优点：

1. 原理简单，实现方便

2PC缺点：

1. 同步阻塞问题。执行过程中，所有参与节点都是事务阻塞型的。当参与者占有公共资源时，其他第三方节点访问公共资源不得不处于阻塞状态。(1pc准备阶段，只执行sql，而不提交，并且占用数据库连接资源)
2. 单点故障。由于协调者的重要性，一旦协调者发生故障。参与者会一直阻塞下去。尤其在第二阶段，协调者发生故障，那么所有的参与者还都处于锁定事务资源的状态中，而无法继续完成事务操作。（如果是协调者挂掉，可以重新选举一个协调者，但是无法解决因为协调者宕机导致的参与者处于阻塞状态的问题）
3. 数据不一致。在二阶段提交的阶段二中，当协调者向参与者发送commit请求之后，发生了局部网络异常或者在发送commit请求过程中协调者发生了故障，这回导致只有一部分参与者接受到了commit请求。而在这部分参与者接到commit请求之后就会执行commit操作。但是其他部分未接到commit请求的机器则无法执行事务提交。于是整个分布式系统便出现了数据部一致性的现象。
4. 泰国保守：如果在协调者指示参与者进行事务提交询问的过程中，参与者出现故障而导致协调者始终无法获取到所有参与者的响应的消息的话，这时协调者只能依靠其自身的超时机制来判断是否中断事务，这样的策略比较保守，换句话说，二阶段提交协议没有设计相应的容错机制，当任意一个参与者节点宕机，那么协调者超时没收到响应，就会导致整个事务回滚失败。

2.2.2 2PC协议实现XA方案

1、理论基础：

2PC的传统方案是在数据库层面实现的，如Oracle、MySQL都支持2PC协议，为了统一标准减少行业内不必要的对接成本，需要制定标准化的处理模型及接口标准，国际开放标准组织Open Group定义了分布式事务处理模型 DTP（Distributed Transaction Processing Reference Model）。

为了让大家更明确XA方案的内容程，下面新用户注册送积分为例来说明：

DTP模型定义TM和RM之间通讯的接口规范叫XA，简单理解为数据库提供的2PC接口协议，基于数据库的XA协议来实现2PC又称为XA方案。

DTP模型定义如下角色：

以上三个角色之间的交互方式如下：

1） TM向AP提供 应用程序编程接口，AP通过TM提交及回滚事务。

2） TM交易中间件通过XA接口来通知RM数据库事务的开始、结束以及提交、回滚等。

总结：

整个2PC的事务流程涉及到三个角色AP、RM、TM。

AP指的是使用2PC分布式事务的应用程序；

RM指的是资源管理器，它控制着分支事务；

TM指的是事务管理器，它控制着整个全局事务。

（1）在准备阶段RM执行实际的业务操作，但不提交事务，资源锁定；

（2）在提交阶段TM会接受RM在准备阶段的执行回复，只要有任一个RM执行失败，TM会通知所有RM执行回滚操作，否则，TM将会通知所有RM提交该事务。提交阶段结束资源锁释放。

XA方案的问题：

1. 需要本地数据库支持XA协议。
2. 资源锁需要等到两个阶段结束才释放，性能较差。

Java的XA方案实现-JTA

---

2.2.3 2PC协议实现Seata-AT方案

1、理论基础：

Seata是由阿里中间件团队发起的开源项目 Fescar，后更名为Seata，它是一个是开源的分布式事务框架。

传统2PC的问题在Seata中得到了解决，它通过对本地关系数据库的分支事务的协调来驱动完成全局事务，是工作在应用层的中间件。主要优点是性能较好，且不长时间占用连接资源，它以高效并且对业务0侵入的方式解决微服务场景下面临的分布式事务问题，它目前提供AT模式(即2PC)及TCC模式的分布式事务解决方案。

Seata的设计思想如下：

Seata的设计目标其一是对业务无侵入，因此从业务无侵入的2PC方案着手，在传统2PC的基础上演进，并解决2PC方案面临的问题。

Seata把一个分布式事务理解成一个包含了若干分支事务的全局事务。全局事务的职责是协调其下管辖的分支事务达成一致，要么一起成功提交，要么一起失败回滚。此外，通常分支事务本身就是一个关系数据库的本地事务

与 传统2PC 的模型类似，Seata定义了3个组件来协议分布式事务的处理过程：

• Transaction Coordinator (TC)： 事务协调器，它是独立的中间件，需要独立部署运行，它维护全局事务的运行状态，接收TM指令发起全局事务的提交与回滚，负责与RM通信协调各各分支事务的提交或回滚。

• Transaction Manager（TM）： 事务管理器，TM需要嵌入应用程序中工作，它负责开启一个全局事务，收集分支事务的事务状态，并最终向TC发起全局提交或全局回滚的指令。（jar包）

• Resource Manager (RM)： 控制分支事务，负责分支注册、状态汇报，并接收事务协调器TC的指令，驱动分支（本地）事务的提交和回滚。

还拿新用户注册送积分举例Seata的分布式事务过程：

先注册再送积分，所以开启全局事务的一方是用户服务，所以TM在用户服务。谁发起，谁是TM

一旦开始执行后，就生成一个全局事务和2个branch数据

具体的执行流程如下：

1. 用户服务的 TM 向 TC 申请开启一个全局事务，全局事务创建成功并生成一个全局唯一的XID。
2. 用户服务的 RM 向 TC 注册 分支事务，该分支事务在用户服务执行新增用户逻辑，并将其纳入 XID 对应全局事务的管辖。
3. 用户服务执行分支事务，向用户表插入一条记录。
4. 逻辑执行到远程调用积分服务时(XID 在微服务调用链路的上下文中传播)。积分服务的RM 向 TC 注册分支事务，该分支事务执行增加积分的逻辑，并将其纳入 XID 对应全局事务的管辖。
5. 积分服务执行分支事务，向积分记录表插入一条记录，执行完毕后，返回用户服务。
6. 用户服务分支事务执行完毕。
7. TM 向 TC 发起针对 XID 的全局提交或回滚决议。
8. TC 调度 XID 下管辖的全部分支事务完成提交或回滚请求。

Seata实现2PC与传统2PC的差别：

架构层次方面，传统2PC方案的 RM 实际上是在数据库层，RM 本质上就是数据库自身，通过 XA 协议实现，而Seata的 RM 是以jar包的形式作为中间件层部署在应用程序这一侧的。

两阶段提交方面，传统2PC无论第二阶段的决议是commit还是rollback，事务性资源的锁都要保持到阶段2完成才释放。而Seata的做法是在阶段1就将本地事务提交，这样就可以省去Phase2（第二阶段）持锁的时间，整体提高效率。

---

2、Seata实现2PC具体代码：

本示例通过Seata中间件实现分布式事务，模拟三个账户的转账交易过程：

两个账户在三个不同的银行(张三在bank1、李四在bank2)，bank1和bank2是两个个微服务。交易过程是，张三给李四转账指定金额。

上述交易步骤，要么一起成功，要么一起失败，必须是一个整体性的事务。

1、准备工作

包括bank1和bank2两个数据库。

微服务框架：spring-boot-2.1.3、spring-cloud-Greenwich.RELEASE

seata客户端（RM、TM）：spring-cloud-alibaba-seata-2.1.0.RELEASE

seata服务端(TC)：seata-server-0.7.1

微服务及数据库的关系 ：
dtx/dtx-seata-demo/seata-demo-bank1 银行1，操作张三账户， 连接数据库bank1 
dtx/dtx-seata-demo/seata-demo-bank2 银行2，操作李四账户，连接数据库bank2

服务注册中心：dtx/discover-server

本示例程序技术架构如下：
ban1操作后openFeign调用bank2

交互流程如下：
1、请求bank1进行转账，传入转账金额。
2、bank1减少转账金额，调用bank2，传入转账金额。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

2、创建数据库

-- 账户表
CREATE TABLE `account_info` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `account_name` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL COMMENT '户主姓名',
  `account_no` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL COMMENT '银行卡号',
  `account_password` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL COMMENT '帐户密码',
  `account_balance` double DEFAULT NULL COMMENT '帐户余额',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8 COLLATE=utf8_bin ROW_FORMAT=DYNAMIC;

-- undolog表，seata需要
CREATE TABLE `undo_log` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `branch_id` bigint(20) NOT NULL,
  `xid` varchar(100) NOT NULL,
  `context` varchar(128) NOT NULL,
  `rollback_info` longblob NOT NULL,
  `log_status` int(11) NOT NULL,
  `log_created` datetime NOT NULL,
  `log_modified` datetime NOT NULL,
  `ext` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=utf8;

INSERT INTO `bank1`.`account_info`(`id`, `account_name`, `account_no`, `account_password`, `account_balance`) VALUES (2, '张三', '1', NULL, 1000);

INSERT INTO `bank2`.`account_info`(`id`, `account_name`, `account_no`, `account_password`, `account_balance`) VALUES (3, '李四的账户', '2', NULL, 0);


-- 包括如下数据库：
-- bank1库，包含张三账户
-- bank2库，包含李四账户
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

3、下载TC（事务协调器）

seata-server-0.7.1.zip

下载解压，运行：/bin/seata-server.bat -p 8888 -m file

注：其中8888为服务端口号；file为启动模式，这里指seata服务将采用文件的方式存储信息。

出现“Server started…”的字样则表示启动成功。

4、注册中心准备

discover-server是服务注册中心，测试工程将自己注册至discover-server。

导入：资料\基础代码\dtx 父工程，此工程自带了discover-server（基于Eureka实现）

5、导入案例工程

dtx-seata-demo是seata的测试工程，根据业务需求需要创建两个dtx-seata-demo工程。

1. 导入dtx-seata-demo

   导入：资料\基础代码\dtx-seata-demo到父工程dtx下。两个测试工程如下：

   dtx/dtx-seata-demo/dtx-seata-demo-bank1 ，操作张三账户，连接数据库bank1

   dtx/dtx-seata-demo/dtx-seata-demo-bank2 ，操作李四账户，连接数据库bank2

2. 父工程maven依赖说明

   在dtx父工程中指定了SpringBoot和SpringCloud版本

   在dtx-seata-demo父工程中指定了spring-cloud-alibaba-dependencies的版本。

3. 配置seata

   在src/main/resource中，新增registry.conf、file.conf文件，内容可拷贝seata-server-0.7.1中的配置文件子。在registry.conf中registry.type使用file：

   在file.conf中更改service.vgroup_mapping.[springcloud服务名]-fescar-service-group = “default”，并修改 service.default.grouplist =[seata服务端地址]

   关于vgroup_mapping的配置：

   vgroup_mapping.事务分组服务名=Seata Server集群名称（默认名称为default） 
   default.grouplist = Seata Server集群地址
   -----------------------------------------------------------------------------------------
   在 org.springframework.cloud:spring-cloud-starter-alibaba-seata 的org.springframework.cloud.alibaba.seata.GlobalTransactionAutoConfiguration  类中，默认会使用${spring.application.name}-fescar-service-group 作为事务分组服务名注册到 Seata Server上，如果和file.conf 中的配置不一致，会提示 no available server to connect 错误
   
   也可以通过配置 spring.cloud.alibaba.seata.tx-service-group 修改后缀，但是必须和 file.conf 中的配置保持一致
   1
   2
   3
   4
   5
   6

6、创建代理源

新增DatabaseConfiguration.java，Seata的RM通过DataSourceProxy才能在业务代码的事务提交时，通过这个切入点，与TC进行通信交互、记录undo_log等。

7、seata正常执行流程

第一个程序开启全局事务，然后得到了事务ID，执行完逻辑后把内容写到undo_log表，然后提交事务给TC。
然后远程调用bank，此时RPC时携带上事务ID，这样bank2页知道了事务ID，bank2执行完逻辑后也存入undo_log表，然后汇报给TC，然后bank1提交全局事务，提交后删除undo_log。

8、seata回滚流程

回滚流程省略前的RM注册过程。

要点说明：

1. 每个RM使用DataSourceProxy连接数据库，其目的是使用ConnectionProxy，使用数据源和数据连接代理的目的就是在第一阶段将undo_log和业务数据放在一个本地事务提交，这样就保存了只要有业务操作就一定有undo_log。
2. 在第一阶段undo_log中存放了数据修改前和修改后的值，为事务回滚作好准备，所以第一阶段完成就已经将分支事务提交，也就释放了锁资源。
3. TM开启全局事务开始，将XID全局事务id放在事务上下文中，通过feign调用也将XID传入下游分支事务，每个分支事务将自己的Branch ID分支事务ID与XID关联。
4. 第二阶段全局事务提交，TC会通知各各分支参与者提交分支事务，在第一阶段就已经提交了分支事务，这里各各参与者只需要删除undo_log即可，并且可以异步执行，第二阶段很快可以完成。
5. 第二阶段全局事务回滚，TC会通知各各分支参与者回滚分支事务，通过 XID 和 Branch ID 找到相应的回滚日志，通过回滚日志生成反向的 SQL 并执行，以完成分支事务回滚到之前的状态，如果回滚失败则会重试回滚操作。

9、编写服务

1. dtx-seata-demo-bank1 张三转账 @GlobalTransactional 注解加在发起方
2. dtx-seata-demo-bank2 李四收账 无需加全局事务注解

事务ID的自动传递

将@GlobalTransactional注解标注在全局事务发起的Service实现方法上，开启全局事务：

GlobalTransactionalInterceptor会拦截@GlobalTransactional注解的方法，生成全局事务ID(XID)，XID会在整个分布式事务中传递。

在远程调用时，spring-cloud-alibaba-seata会拦截Feign调用将XID传递到下游服务。

10、测试场景

• 张三向李四转账成功。
• 李四事务失败，张三事务回滚成功。
• 张三事务失败，李四事务回滚成功。
• 分支事务超时测试。

3、总结：

本节讲解了传统2PC（基于数据库XA协议）和Seata实现2PC的两种2PC方案，由于Seata的0侵入性并且解决了传统2PC长期锁资源的问题，所以推荐采用Seata实现2PC。

Seata实现2PC要点：

1. 全局事务开始使用 @GlobalTransactional标识 。
2. 每个本地事务方案仍然使用@Transactional标识。
3. 每个数据都需要创建undo_log表，此表是seata保证本地事务一致性的关键。

2PC失效场景—协调者故障

2.2.4 2PC与3PC

待完善

2.2 TCC方案

1、理论基础:

2PC 和 3PC 都是数据库层面的，而 TCC 是业务层面的分布式事务，就像我前面说的分布式事务不仅仅包括数据库的操作，还包括发送短信等，这时候 TCC 就派上用场了！ TCC属于最终一致性事务，2PC属于强一致性事务。

什么是TCC事务？

TCC是Try、Confirm、Cancel三个词语的缩写，TCC要求每个分支事务实现三个操作：预处理Try、确认Confirm、撤销Cancel。Try操作做业务检查及资源预留，Confirm做业务确认操作，Cancel实现一个与Try相反的操作即回滚操作。TM首先发起所有的分支事务的try操作，任何一个分支事务的try操作执行失败，TM将会发起所有分支事务的Cancel操作，若try操作全部成功，TM将会发起所有分支事务的Confirm操作，其中Confirm/Cancel操作若执行失败，TM会进行重试。即try成功了，那么认为confirm/cancel一定会成功，所以失败会重试或者人工介入。

TCC事务流程：

TCC优点：

• TCC 不存在资源阻塞的问题，因为每个方法都直接进行事务的提交，一旦出现异常通过则 Cancel 来进行回滚补偿，这也就是常说的补偿性事务。

TCC缺点：

• TCC 对业务的侵入较大和业务紧耦合，需要根据特定的场景和业务逻辑来设计相应的操作。
• 需要每个服务编写try、confirm、cancel三个方法，考虑网络波动、服务宕机等原因，所以会有重试机制，所以需要保证这三个方法的幂等性，而且这三个方法有时还真不好写，同时也会增加开发量。

TCC框架解决方案：

Seata也支持TCC，但Seata（低版本）的TCC模式对Spring Cloud并没有提供支持。我们的目标是理解TCC的原理以及事务协调运作的过程，因此更请倾向于轻量级易于理解的框架，因此最终确定了Hmily。

Hmily分布式事务框架简介：

Hmily是一个高性能分布式事务TCC开源框架。基于Java语言来开发（JDK1.8），支持Dubbo，Spring Cloud等RPC框架进行分布式事务。它目前支持以下特性：

• 支持嵌套事务(Nested transaction support).
• 采用disruptor框架进行事务日志的异步读写，与RPC框架的性能毫无差别。
• 支持SpringBoot-starter 项目启动，使用简单。
• RPC框架支持 : dubbo,motan,springcloud。
• 本地事务存储支持 : redis,mongodb,zookeeper,file,mysql。
• 事务日志序列化支持 ：java，hessian，kryo，protostuff。
• 采用Aspect AOP 切面思想与Spring无缝集成，天然支持集群。
• RPC事务恢复，超时异常恢复等。

Hmily利用AOP对参与分布式事务的本地方法与远程方法进行拦截处理，通过多方拦截，事务参与者能透明的调用到另一方的Try、Confirm、Cancel方法；传递事务上下文；并记录事务日志，酌情进行补偿，重试等。

Hmily不需要事务协调服务，但需要提供一个数据库(mysql/mongodb/zookeeper/redis/file)来进行日志存储。

Hmily实现的TCC服务与普通的服务一样，只需要暴露一个接口，也就是它的Try业务。Confirm/Cancel业务逻辑，只是因为全局事务提交/回滚的需要才提供的，因此Confirm/Cancel业务只需要被Hmily TCC事务框架发现即可，不需要被调用它的其他业务服务所感知。

官网介绍：https://dromara.org/website/zh-cn/docs/hmily/index.html

TCC需要注意三种异常处理分别是空回滚、幂等、悬挂:

1）空回滚：
解释：在没有调用 TCC 资源 Try 方法的情况下，调用了二阶段的 Cancel 方法，Cancel 方法需要识别出这是一个空回滚，然后直接返回成功。

出现原因：当一个分支事务所在服务宕机或网络异常，分支事务调用记录为失败，这个时候其实是没有执行Try阶段，当故障恢复后，分布式事务进行回滚则会调用二阶段的Cancel方法，从而形成空回滚。

解决思路是关键就是要识别出这个空回滚。思路很简单就是需要知道一阶段是否执行，如果执行了，那就是正常回滚；如果没执行，那就是空回滚。前面已经说过TM在发起全局事务时生成全局事务记录，全局事务ID贯穿整个分布式事务调用链条。再额外增加一张分支事务记录表，其中有全局事务 ID 和分支事务 ID，第一阶段 Try 方法里会插入一条记录，表示一阶段执行了。Cancel 接口里读取该记录，如果该记录存在，则正常回滚；如果该记录不存在，则是空回滚。

2）幂等：
通过前面介绍已经了解到，为了保证TCC二阶段提交重试机制不会引发数据不一致，要求 TCC 的二阶段 Try、

Confirm 和 Cancel 接口保证幂等，这样不会重复使用或者释放资源。如果幂等控制没有做好，很有可能导致数据不一致等严重问题。

解决思路在上述“分支事务记录”中增加执行状态，每次执行前都查询该状态。

3）悬挂：
说明：悬挂就是对于一个分布式事务，其二阶段 Cancel 接口比 Try 接口先执行。

出现原因：在 RPC 调用分支事务try时，先注册分支事务，再执行RPC调用，如果此时 RPC 调用的网络发生拥堵，通常 RPC 调用是有超时时间的，RPC 超时以后，TM就会通知RM回滚该分布式事务，可能回滚完成后，RPC 请求才到达参与者真正执行，而一个 Try 方法预留的业务资源，只有该分布式事务才能使用，该分布式事务第一阶段预留的业务资源就再也没有人能够处理了，对于这种情况，我们就称为悬挂，即业务资源预留后没法继续处理

解决思路：如果二阶段执行完成，那一阶段就不能再继续执行。在执行一阶段事务时判断在该全局事务下，“分支事务记录”表中是否已经有二阶段事务记录，如果有则不执行Try。

2、Hmily实现TCC具体代码:

举例，场景为 A 转账 30 元给 B，A和B账户在不同的服务。

A账户：

try：
    try幂等校验
    try悬挂处理
    检查余额是否够30元
    扣减30元
confirm：
    空
cancel：
    cancel幂等校验
    cancel空回滚处理
    增加可用余额30元
1
2
3
4
5
6
7
8
9
10
11

B账户：

try：
    空
confirm：
    confirm幂等校验
    正式增加30元
cancel：
    空
1
2
3
4
5
6
7

1、准备工作：

微服务：spring-boot-2.1.3、spring-cloud-Greenwich.RELEASE

Hmily：hmily-springcloud.2.0.4-RELEASE

微服务及数据库的关系 ：
dtx/dtx-tcc-demo/dtx-tcc-demo-bank1 银行1，操作张三账户， 连接数据库bank1
dtx/dtx-tcc-demo/dtx-tcc-demo-bank2 银行2，操作李四账户，连接数据库bank2

服务注册中心：dtx/discover-serve
1
2
3
4
5
6
7
8
9

2、创建数据库：

-- 创建hmily数据库，用于存储hmily框架记录的数据。 数据库中表由hmily在执行过程中创建，用于记录事务执行过程中进行回滚或重试，每个服务会创建一张表，命名规则：Hmily_服务名称
CREATE DATABASE `hmily` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci';
1
2

-- 在bank1和bank2中创建try、confirm、cancel日志表：
CREATE TABLE `local_try_log` (
   `tx_no` varchar(64) NOT NULL COMMENT '事务id',
   `create_time` datetime DEFAULT NULL,
   PRIMARY KEY (`tx_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8
CREATE TABLE `local_confirm_log` (
   `tx_no` varchar(64) NOT NULL COMMENT '事务id',
   `create_time` datetime DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8
CREATE TABLE `local_cancel_log` (
   `tx_no` varchar(64) NOT NULL COMMENT '事务id',
   `create_time` datetime DEFAULT NULL,
   PRIMARY KEY (`tx_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

3、导入案例工程dtx-tcc-demo

引入依赖，在application-local.yml中添加Hmily配置：

org:
 dromara:
   hmily :
     serializer : kryo
     recoverDelayTime : 30
     retryMax : 30
     scheduledDelay : 30
     scheduledThreadMax :  10
     repositorySupport : db
     started: true  # 在bank2下为false
     hmilyDbConfig :
       driverClassName  : com.mysql.jdbc.Driver
       url :  jdbc:mysql://localhost:3306/hmily?useUnicode=true
       username : root
       password : 123456
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

启动类加@EnableAspectJAutoProxy，并扫描org.dromara.hmily包

4、编写服务

4.1 编写 dtx-tcc-demo-bank1服务实现 AccountInfoServiceImpl

@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {

    @Autowired
    AccountInfoDao accountInfoDao;

    @Autowired
    Bank2Client bank2Client;

    // 账户扣款，就是tcc的try方法

    /**
     * 	try幂等校验
     * 	try悬挂处理
     * 	检查余额是够扣减金额
     * 	扣减金额、
     */
    @Override
    @Transactional
    //只要标记@Hmily就是try方法，在注解中指定confirm、cancel两个方法的名字
    @Hmily(confirmMethod="commit",cancelMethod="rollback")
    public void updateAccountBalance(String accountNo, Double amount) {
        //获取全局事务id
        String transId = HmilyTransactionContextLocal.getInstance().get().getTransId();
        log.info("bank1 try begin 开始执行...xid:{}",transId);
        
        //幂等判断 判断local_try_log表中是否有try日志记录，如果有则不再执行
        if(accountInfoDao.isExistTry(transId)>0){
            log.info("bank1 try 已经执行，无需重复执行,xid:{}",transId);
            return ;
        }

        //try悬挂处理，如果cancel、confirm有一个已经执行了，try不再执行
        if(accountInfoDao.isExistConfirm(transId)>0 || accountInfoDao.isExistCancel(transId)>0){
            log.info("bank1 try悬挂处理  cancel或confirm已经执行，不允许执行try,xid:{}",transId);
            return ;
        }

        //扣减金额
        if(accountInfoDao.subtractAccountBalance(accountNo, amount)<=0){
            //扣减失败
            throw new RuntimeException("bank1 try 扣减金额失败,xid:{}"+transId);
        }
        //插入try执行记录,用于幂等判断
        accountInfoDao.addTry(transId);

        //远程调用李四，转账
        if(!bank2Client.transfer(amount)){
            throw new RuntimeException("bank1 远程调用李四微服务失败,xid:{}"+transId);
        }
        if(amount == 2){
            throw new RuntimeException("人为制造异常,xid:{}"+transId);
        }
        log.info("bank1 try end 结束执行...xid:{}",transId);
    }

    //confirm方法
    @Override
    @Transactional
    public void commit(String accountNo, Double amount){
        //获取全局事务id
        String transId = HmilyTransactionContextLocal.getInstance().get().getTransId();
        log.info("bank1 confirm begin 开始执行...xid:{},accountNo:{},amount:{}",transId,accountNo,amount);
    }


    /** cancel方法
     * 	cancel幂等校验
     * 	cancel空回滚处理
     * 	增加可用余额
     */
    @Override
    @Transactional
    public void rollback(String accountNo, Double amount){
        //获取全局事务id
        String transId = HmilyTransactionContextLocal.getInstance().get().getTransId();
        log.info("bank1 cancel begin 开始执行...xid:{}",transId);
        //	cancel幂等校验
        if(accountInfoDao.isExistCancel(transId)>0){
            log.info("bank1 cancel 已经执行，无需重复执行,xid:{}",transId);
            return ;
        }
        //cancel空回滚处理，如果try没有执行，cancel不允许执行
        if(accountInfoDao.isExistTry(transId)<=0){
            log.info("bank1 空回滚处理，try没有执行，不允许cancel执行,xid:{}",transId);
            return ;
        }
        //	增加可用余额
        accountInfoDao.addAccountBalance(accountNo,amount);
        //插入一条cancel的执行记录
        accountInfoDao.addCancel(transId);
        log.info("bank1 cancel end 结束执行...xid:{}",transId);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

feign调用dtx-tcc-demo-bank2服务,完成减钱操作：

@FeignClient(value="tcc-demo-bank2",fallback= Bank2ClientFallback.class)
public interface Bank2Client {
    //远程调用李四的微服务
    @GetMapping("/bank2/transfer")
    @Hmily  //Hmily整合feign会将事物id传递到调用服务下游，需要加这个注解
    public Boolean transfer(@RequestParam("amount") Double amount);
}
1
2
3
4
5
6
7

4.2 编写 dtx-tcc-demo-bank2服务实现 AccountInfoServiceImpl

@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {

    @Autowired
    AccountInfoDao accountInfoDao;

    @Override
    @Hmily(confirmMethod="confirmMethod", cancelMethod="cancelMethod")
    public void updateAccountBalance(String accountNo, Double amount) {
        //获取全局事务id
        String transId = HmilyTransactionContextLocal.getInstance().get().getTransId();
        log.info("bank2 try begin 开始执行...xid:{}",transId);
    }

    /**
     * confirm方法
     * 	confirm幂等校验
     * 	正式增加金额
     * @param accountNo
     * @param amount
     */
    @Transactional
    public void confirmMethod(String accountNo, Double amount){
        //获取全局事务id
        String transId = HmilyTransactionContextLocal.getInstance().get().getTransId();
        log.info("bank2 confirm begin 开始执行...xid:{}",transId);
        if(accountInfoDao.isExistConfirm(transId)>0){
            log.info("bank2 confirm 已经执行，无需重复执行...xid:{}",transId);
            return ;
        }
        //增加金额
        accountInfoDao.addAccountBalance(accountNo,amount);
        //增加一条confirm日志，用于幂等
        accountInfoDao.addConfirm(transId);
        log.info("bank2 confirm end 结束执行...xid:{}",transId);
    }



    /**
     * @param accountNo
     * @param amount
     */
    public void cancelMethod(String accountNo, Double amount){
        //获取全局事务id
        String transId = HmilyTransactionContextLocal.getInstance().get().getTransId();
        log.info("bank2 cancel begin 开始执行...xid:{}",transId);

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

5、场景测试：

• 张三向李四转账成功。
• 李四事务失败，张三事务回滚成功。
• 张三事务失败，李四事务回滚成功。
• 分支事务超时测试。

3、总结：

如果拿TCC事务的处理流程与2PC两阶段提交做比较，2PC通常都是在跨库的DB层面，而TCC则在应用层面的处理，需要通过业务逻辑来实现。这种分布式事务的实现方式的优势在于，可以让应用自己定义数据操作的粒度，使得降低锁冲突、提高吞吐量成为可能。

而不足之处则在于对应用的侵入性非常强，业务逻辑的每个分支都需要实现try、confirm、cancel三个操作。此外，其实现难度也比较大，需要按照网络状态、系统故障等不同的失败原因实现不同的回滚策略。

2.3 Rocket实现可靠消息最终一致性

1、本地事务消息表

2、rocketMq的事务消息

本实例通过RocketMQ中间件实现可靠消息最终一致性分布式事务，模拟两个账户的转账交易过程。

两个账户在分别在不同的银行(张三在bank1、李四在bank2)，bank1、bank2是两个微服务。

交易过程是，张三给李四转账指定金额。

上述交易步骤，张三扣减金额与给bank2发转账消息，两个操作必须是一个整体性的事务

1、准备工作

本示例程序组成部分如下：

包括bank1和bank2两个数据库。

rocketmq 服务端：RocketMQ-4.5.0

rocketmq 客户端：RocketMQ-Spring-Boot-starter.2.0.2-RELEASE

微服务框架：spring-boot-2.1.3、spring-cloud-Greenwich.RELEASE

微服务及数据库的关系 ：

dtx/dtx-txmsg-demo/dtx-txmsg-demo-bank1 银行1，操作张三账户， 连接数据库bank1

dtx/dtx-txmsg-demo/dtx-txmsg-demo-bank2 银行2，操作李四账户，连接数据库bank2

本示例程序技术架构如下：

交互流程如下：

1、Bank1向MQ Server发送转账消息

2、Bank1执行本地事务，扣减金额

3、Bank2接收消息，执行本地事务，添加金额
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

2、创建数据哭

-- 在bank1、bank2数据库中新增de_duplication，交易记录表(去重表)，用于交易幂等控制
DROP TABLE IF EXISTS `de_duplication`;
CREATE TABLE `de_duplication` (
    `tx_no` varchar(64) COLLATE utf8_bin NOT NULL,
    `create_time` datetime(0) NULL DEFAULT NULL,
    PRIMARY KEY (`tx_no`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;
1
2
3
4
5
6
7

3、启动mq

# 启动nameserver
/bin/mqnamesrv.cmd
# 启动broker
/bin/mqbroker.cmd ‐n 127.0.0.1:9876 autoCreateTopicEnable=true
1
2
3
4

4、导入dtx-txmsg-demo

dtx-txmsg-demo是本方案的测试工程，根据业务需求需要创建两个dtx-txmsg-demo工程

（1）导入dtx-txmsg-demo
导入：资料\基础代码\dtx-txmsg-demo到父工程dtx下。两个测试工程如下：

dtx/dtx-txmsg-demo/dtx-txmsg-demo-bank1 ，操作张三账户，连接数据库bank1
dtx/dtx-txmsg-demo/dtx-txmsg-demo-bank2  ，操作李四账户，连接数据库bank2

（2）父工程maven依赖说明

在dtx父工程中指定了SpringBoot和SpringCloud版本
在dtx-txmsg-demo父工程中指定了rocketmq-spring-boot-starter的版本。

（3）配置rocketMQ

在application-local.propertis中配置rocketMQ  nameServer地址及生产组：
其它详细配置见导入的基础工程。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

5、编写服务

dtx-txmsg-demo-bank1 张三转账

dtx-txmsg-demo-bank2 李四收钱

6、测试场景:

bank1本地事务失败，则bank1不发送转账消息。

bank2接收转账消息失败，会进行重试发送消息。

bank2多次消费同一个消息，实现幂等。

3、总结：

可靠消息最终一致性就是保证消息从生产方经过消息中间件传递到消费方的一致性，本案例使用了RocketMQ作为消息中间件，RocketMQ主要解决了两个功能：

1、本地事务与消息发送的原子性问题。

2、事务参与方接收消息的可靠性。

可靠消息最终一致性事务适合执行周期长且实时性要求不高的场景。引入消息机制后，同步的事务操作变为基于消息执行的异步操作, 避免了分布式事务中的同步阻塞操作的影响，并实现了两个服务的解耦。

2.4 尽最大努力通知

1、基础理论

最大努力通知也是一种解决分布式事务的方案，

交互流程:

1、账户系统调用充值系统接口

2、充值系统完成支付处理向账户系统发起充值结果通知（这个是重点）

若通知失败，则充值系统按策略进行重复通知

3、账户系统接收到充值结果通知修改充值状态。

4、账户系统未接收到通知会主动调用充值系统的接口查询充值结果。

如果实在通知不到，就提供查询接口，让用户主动去查询一遍

通过上边的例子我们总结最大努力通知方案的目标：

目标：发起通知方通过一定的机制最大努力将业务处理结果通知到接收方。具体包括：

1、有一定的消息重复通知机制。

​ 因为接收通知方可能没有接收到通知，此时要有一定的机制对消息重复通知。

2、消息校对机制。

​ 如果尽最大努力也没有通知到接收方，或者接收方消费消息后要再次消费，此时可由接收方主动向通知方查询消息信息来满足需求。

最大努力通知与可靠消息一致性有什么不同？

1、解决方案思想不同

可靠消息一致性，发起通知方需要保证将消息发出去，并且将消息发到接收通知方，消息的可靠性关键由发起通知方来保证。
最大努力通知，发起通知方尽最大的努力将业务处理结果通知为接收通知方，但是可能消息接收不到，此时需要接收通知方主动调用发起通知方的接口查询业务处理结果，通知的可靠性关键在接收通知方。

2、两者的业务应用场景不同

​ 可靠消息一致性关注的是交易过程的事务一致，以异步的方式完成交易。

最大努力通知关注的是交易后的通知事务，即将交易结果可靠的通知出去。

3、技术解决方向不同

可靠消息一致性要解决消息从发出到接收的一致性，即消息发出并且被接收到。

最大努力通知无法保证消息从发出到接收的一致性，只提供消息接收的可靠性机制。可靠机制是，最大努力的将消息通知给接收方，当消息无法被接收方接收时，由接收方主动查询消息（业务处理结果）。

解决方案：

通过对最大努力通知的理解，采用MQ的ack机制就可以实现最大努力通知。

方案1：

重点不在于谁发起调用，重点是将通知给谁

但是问题是接收方居然是个MQ，浏览器怎么会有MQ呢？

还有支付宝的MQ怎么会让你监控呢？

所以后面会有优化

本方案是利用MQ的ack机制由MQ向接收通知方发送通知，流程如下：

1、发起通知方将通知发给MQ。

使用普通消息机制将通知发给MQ。

注意：如果消息没有发出去可由接收通知方主动请求发起通知方查询业务执行结果。（后边会讲）

2、接收通知方监听 MQ。

3、接收通知方接收消息，业务处理完成回应ack。

4、接收通知方若没有回应ack则MQ会重复通知。

MQ会按照间隔1min、5min、10min、30min、1h、2h、5h、10h的方式，逐步拉大通知间隔 （如果MQ采用rocketMq，在broker中可进行配置），直到达到通知要求的时间窗口上限。

5、接收通知方可通过消息校对接口来校对消息的一致性。

方案2：

本方案也是利用MQ的ack机制，与方案1不同的是应用程序向接收通知方发送通知，如下图：

交互流程如下：

1、发起通知方将通知发给MQ。

使用可靠消息一致方案中的事务消息保证本地事务与消息的原子性，最终将通知先发给MQ。

2、通知程序监听 MQ，接收MQ的消息。

方案1中接收通知方直接监听MQ，方案2中由通知程序监听MQ。

通知程序若没有回应ack则MQ会重复通知。

3、通知程序通过互联网接口协议（如http、webservice）调用接收通知方案接口，完成通知。

通知程序调用接收通知方案接口成功就表示通知成功，即消费MQ消息成功，MQ将不再向通知程序投递通知消息。

4、接收通知方可通过消息校对接口来校对消息的一致性。

方案1和方案2的不同点：

1、方案1中接收通知方与MQ接口，即接收通知方案监听 MQ，此方案主要应用与内部应用之间的通知。
2、方案2中由通知程序与MQ接口，通知程序监听MQ，收到MQ的消息后由通知程序通过互联网接口协议调用接收通知方。此方案主要应用于外部应用之间的通知，例如支付宝、微信的支付结果通知。

2、RocketMQ实现最大努力通知型事务

案例说明：

本实例通过RocketMQ中间件实现最大努力通知型分布式事务，模拟充值过程。

本案例有账户系统和充值系统两个微服务，其中

• 账户系统的数据库是bank1数据库，其中有张三账户。
• 充值系统的数据库是bank1_pay数据库，记录了账户的充值记录。

交互流程如下：

1、用户请求充值系统进行充值。

2、充值系统完成充值将充值结果发给MQ。

3、账户系统监听MQ，接收充值结果通知，如果接收不到消息，MQ会重复发送通知。接收到充值结果通知账户系统增加充值金额。

4、账户系统也可以主动查询充值系统的充值结果查询接口，增加金额。

1、准备工作

本示例程序组成部分如下：

包括bank1和bank1_pay两个数据库。

rocketmq 服务端：RocketMQ-4.5.0

rocketmq 客户端：RocketMQ-Spring-Boot-starter.2.0.2-RELEASE

微服务框架：spring-boot-2.1.3、spring-cloud-Greenwich.RELEASE

微服务及数据库的关系 ：

dtx/dtx-notifymsg-demo/dtx-notifymsg-demo-bank1 银行1，操作张三账户， 连接数据库bank1

dtx/dtx-notifymsg-demo/dtx-notifymsg-demo-pay 银行2，操作充值记录，连接数据库bank1_pay

交互流程如下：

1、用户请求充值系统进行充值。

2、充值系统完成充值将充值结果发给MQ。

3、账户系统监听MQ，接收充值结果通知，如果接收不到消息，MQ会重复发送通知。接收到充值结果通知账户系统增加充值金额。

4、账户系统也可以主动查询充值系统的充值结果查询接口，增加金额。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

2、创建数据库

CREATE DATABASE `bank1_pay` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci';
CREATE TABLE `account_pay` (
    `id` varchar(64) COLLATE utf8_bin NOT NULL,
    `account_no` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '账号',
    `pay_amount` double NULL DEFAULT NULL COMMENT '充值余额',
    `result` varchar(20) COLLATE utf8_bin DEFAULT NULL COMMENT '充值结果:success，fail',
    PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;
1
2
3
4
5
6
7
8

3、导入dtx-notifymsg-demo

dtx-notifydemo-bank1 本地账户系统

实现如下功能：

1. 监听MQ，接收充值结果，根据充值结果完成账户金额修改。
2. 主动查询充值系统，根据充值结果完成账户金额修改。

dtx-notifymsg-demo-pay 第三方充值系统

实现如下功能：

1. 充值接口
2. 充值完成要通知
3. 充值结果查询接口

4、测试场景：

1. 充值系统充值成功，账户系统主动查询充值结果，修改账户金额。
2. 充值系统充值成功，发送消息，账户系统接收消息，修改账户金额。
3. 账户系统修改账户金额幂等测试。

3、总结：

最大努力通知方案是分布式事务中对一致性要求最低的一种,适用于一些最终一致性时间敏感度低的业务；最大努力通知方案需要实现如下功能：

1、消息重复通知机制。

2、消息校对机制。

系统的充值结果查询接口，增加金额。

**2、创建数据库**

```mysql
CREATE DATABASE `bank1_pay` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci';
CREATE TABLE `account_pay` (
    `id` varchar(64) COLLATE utf8_bin NOT NULL,
    `account_no` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '账号',
    `pay_amount` double NULL DEFAULT NULL COMMENT '充值余额',
    `result` varchar(20) COLLATE utf8_bin DEFAULT NULL COMMENT '充值结果:success，fail',
    PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;
1
2
3
4
5
6
7
8
9
10
11
12

3、导入dtx-notifymsg-demo

dtx-notifydemo-bank1 本地账户系统

实现如下功能：

1. 监听MQ，接收充值结果，根据充值结果完成账户金额修改。
2. 主动查询充值系统，根据充值结果完成账户金额修改。

dtx-notifymsg-demo-pay 第三方充值系统

实现如下功能：

1. 充值接口
2. 充值完成要通知
3. 充值结果查询接口

4、测试场景：

1. 充值系统充值成功，账户系统主动查询充值结果，修改账户金额。
2. 充值系统充值成功，发送消息，账户系统接收消息，修改账户金额。
3. 账户系统修改账户金额幂等测试。

3、总结：

最大努力通知方案是分布式事务中对一致性要求最低的一种,适用于一些最终一致性时间敏感度低的业务；最大努力通知方案需要实现如下功能：

1、消息重复通知机制。

2、消息校对机制。