分布式事务实战

转载本文需注明出处：微信公众号EAWorld，违者必究。

1.1事务原理

在讲分布式事务之前，先聊一下事务。简单讲事务是数据库管理系统执行过程中的一个逻辑单元，它能保证要么一组数据库操作全部执行成功，要么全部失败，而做到这些的原理就是事务的ACID四大特性。

• A. Atomic原子性的简称，事务作为一个整体来执行，要么全部成功，要么全部失败。

• C. Consistency一致性的简称，事务应确保数据从一个一致的状态转变为另一个一致的状态。

• I.  Isolation隔离性的简称，多个事务并发执行时，一个事务的执行不影响其他事务的执行。

• D．Durability持久性的检查，已提交的事务修改数据会被持久保存。

1.2传统单机数据库事务

在传统单体应用架构中，我们的业务数据通常都是存储在一个数据库中的，应用中的各个模块对数据库直接进行操作。在这种场景中，事务是由数据库提供的基于ACID特性来保证的。

例如，在一个用户购物下单的场景中，涉及到用户、订单、支付、库存等模块的一系列协同操作，如果其中一个模块出现问题，我们就可以通过数据库提供的事务特性来保证本次下单操作要么都成功，要么都失败。因为这些模块用的是同一个数据库，所处的是同一个事务管理器，不需要做额外的其他操作就能保证事务的特性。

从广义上来讲，分布式事务其实也是事务，只是区别于单机事务不同之处是：由于业务上的定义和系统微服务架构的设计，很多大型的业务流程都被拆分成了多个单一的基础服务，而为了保证每个微服务都能独立进行开发和部署运行，通常都会采用一个微服务一个数据库的架构配套，然后将内部服务进行封装，以Rest api方式对外暴露。这样以往基于数据库来实现的数据操作，就变成了多个对外提供微服务的微服务系统之间的协同操作。在这种情况下，原有的单机事务方式已经不能够使用了，因为多个服务就意味着存在多个事务管理器和多个资源，单个微服务的本地事务管理器只能保证本地事务的ACID，为了在多个服务之间能保证业务的事务性，参与分布式事务的微服务通常会依托协调器来完成相关的一致性协调操作。

那我们在微服务系统实际开发中，如何去实现协调器以处理分布式事务呢，这里的解决方案是采用华为提供的servicecomb-pack框架来解决这一问题。

2. 分布式事务解决方案：

2.1补偿方式

在讲servicecomb-pack之前先了解两个概念：不完美补偿(saga)和完美补偿(tcc)。

1. saga：不完美补偿，一般在系统中我们会专门为业务逻辑对应写一个补偿逻辑，如果业务逻辑执行失败，就会去执行这个补偿逻辑，我们称这个补偿逻辑为反向操作，这个反向操作同样会留下操作痕迹，例如：在银行系统中，客户去ATM取钱，银行会先对用户账户进行扣款操作，如果本次取钱不成功，银行系统会发出一笔冲正操作，将之前扣除的款项打回用户账户，这个冲正操作在交易记录里面是开源查询到的。

2. tcc：完美补偿，cancel阶段会彻底清楚之前的业务逻辑操作，用户是感知不到的。例如：在一个交易平台去发起交易，首先在try阶段不会直接去扣除账户余额，而且去检查用户的额度并刷新额度，然后在confirm阶段才去真正操作账户。如果出现异常，那么在cancel阶段就需要去执行业务逻辑来取消try阶段产生的后果，释放在try阶段被占用的额度。整个过程只有等confirm执行完毕，交易才算完成。

2.2servicecomb-pack

servicecomb-pack出自于华为微服务框架servicecomb，是一个开源的分布式事务最终一致性解决方案，该项目已交由Apache软件基金会孵化，目前已经在apache毕业了。0.3.0版本之前叫servicecomb-saga，现版本已经改名为servicecomb-pack。

servicecomb-pack架构主要包含两个组件：alpha和Omega

• alpha：alpha其实就是一个server端，需要用户自行编译运行，它的作用就是上述中的分布式事务协调器，主要作用是和Omega客户端进行通讯，接收omega发过来的事务事件，然后进行持久化存储事务以及修改协调子事务的状态，从而保证全局事务中的所有子事务状态都一致，即要么全执行完成，要么全执行失败。

• omega：Omega端其实可以看成是一个微服务中内嵌的agent，主要作用是监控本地子事务的执行情况并向alpha-server端发送子事务执行事件以及传递全局事务ID，并在异常情况下会根据alpha下发的操作事件进行相应的补偿操作。

Omega会以切面编程的方式向应用程序注入相关的处理模块，帮助我们构建分布式事务调用的上下文。Omega在事务处理初始阶段处理事务的相关准备的操作，在事务执行完毕做一些清理的操作，例如创建分布式事务起始事件，以及相关的子事件，根据事务的执行的成功或者失败生产相关的事务终止或者失败事件。这样带来的好处是用户的代码只需要添加几个annotation 来描述分布式事务执行范围，以及与本地的事务处理恢复的相关函数信息，Omega就能通过切面注入的代码能够追踪与本地事务的执行情况。Omega会将本地事务执行的情况以事件的方式通知给Alpha。由于单个Omega不可能知晓一个分布式事务下其他参与服务的执行情况， 这样就需要Alpha扮演一个十分重要的协调者的角色。Alpha将收集到的分布式事务事件信息整理汇总，通过分析这些事件之间的关系可以了解到分布式事务的执行情况， Alpha通过向Omega下发相关的执行指令由Omega执行相关提交或恢复操作，实现分布式事务的最终一致性。

在了解的Pack实现的部分细节之后， 我们可以从下图进一步了解ServiceComb Pack架构下，Alpha与Omega内部各模块之间的关系图[1]。

整个架构分为三个部分，一个是Alpha协调器，另外一个就是注入到微服务实例中的Omega，以及Alpha与Omega之间的交互协议， 目前ServiceComb Pack支持Saga 以及TCC两种分布式事务协调协议实现。

Omega包含了与分析用户分布式事务逻辑相关的事务注解模块（Transaction Annotation）以及事务拦截器（Transaction Interceptor）；分布式事务执行相关的事务上下文（Transaction Context），事务回调（Transaction Callback) ，事务执行器（Transaction Executor）；以及负责与Alpha进行通讯的事务传输（Transaction Transport）模块。

• 事务注解模块是分布式事务的用户界面，用户将这些标注添加到自己的业务代码之上用以描述与分布式事务相关的信息，这样Omega就可以按照分布式事务的协调要求进行相关的处理。如果大家扩展自己的分布式事务，也可以通过定义自己的事务标注来实现。

• 事务拦截器这个模块我们可以借助AOP手段，在用户标注的代码基础上添加相关的拦截代码，获取到与分布式事务以及本地事务执行相关的信息，并借助事务传输模块与Alpha进行通讯传递事件。

• 事务上下文为Omega内部提供了一个传递事务调用信息的一个手段，借助前面提到的全局事务ID以及本地事务ID的对应关系，Alpha可以很容易检索到与一个分布式事务相关的所有本地事务事件信息。

• 事务执行器主要是为了处理事务调用超时设计的模块。由于Alpha与Omega之间的连接有可能不可靠，Alpha端很难判断Omega本地事务执行超时是由Alpha与Omega直接的网络引起的还是Omega自身调用的问题，因此设计了事务执行器来监控Omega的本地的执行情况，简化Omega的超时操作。目前Omega的缺省实现是直接调用事务方法，由Alpha的后台服务通过扫描事件表的方式来确定事务执行时间是否超时。

• 事务回调在Omega与Alpha建立连接的时候就会向Alpha进行注册，当Alpha需要进行相关的协调操作的时候，会直接调用Omega注册的回调方法进行通信。由于微服务实例在云化场景启停会很频繁，我们不能假设Alpha一直能找到原有注册上的事务回调， 因此我们建议微服务实例是无状态的，这样Alpha只需要根据服务名就能找到对应的Omega进行通信。

• 事务传输模块负责Omega与Alpha之间的通讯，在具体的实现过程中，Pack通过定义相关的Grpc描述接口文件定义了TCC 以及Saga的事务交互方法， 同时也定义了与交互相关的事件[2]。 

3.1 alpha-server配置

3.1.1编译alpha-server

1. 环境准备

• JDK1.8

• Maven3.x

2. 源码获取

Github地址：https://github.com/apache/servicecomb-pack

$ git clone：https://github.com/apache/servicecomb-pack.git

$ git checkout 0.4.0

3. 修改配置文件

找到alpha-server/src/main/resource/application.yaml，修改datasource信息为本地信息即可

4. 本地构建alpha-server

$ cd servicecomb-pack

$ mvn clean install -DskipTests -Pspring-boot-2

在执行完命令后，可在alpha/alpha-server/target/saga/alpha-server-${version}-exec.jar中找到alpha-server的可执行jar包

5. 初始化数据库

可在alpha\alpha-server\src\main\resources目录下找到schema-mysql.sql和schema-postgresql.sql两个sql文件，可自行根据所选数据库进行初始化即可。

6. 启动alpha-server

java -Dspring.profiles.active=prd -D"spring.datasource.url=jdbc:postgresql://${host_address}:5432/saga?useSSL=false" -jar alpha-server-${saga_version}-exec.jar

*注意：请在执行命令前将${saga_version}和${host_address}更改为实际值

至此，alpha-server全局事务管理器已经启动成功。

3.1.2替换postgresql为mysql

目前alpha-server支持pg和mysql两种数据库，默认为pg，如需改为mysql，需要进行如下操作：

1. 安装并运行mysql

2. 修改pom文件，添加依赖

alpha-server/pom.xml，添加mysql依赖

ependency>
  <groupId>mysql</groupId>
  <artifactId>mysql-connector-java</artifactId>
  <scope>runtime</scope>
  <version>8.0.15</version>
</dependency>

3. 修改配置文件

找到alpha-server/src/main/resource/application.yaml，修改datasource信息为本地信息即可

spring:
  profiles: mysql
  datasource:
    username: ${username}
    password: ${password}
    url: jdbc:mysql://${host_address}:${port}/${database_name}?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8&useSSL=false
    platform: mysql
    continue-on-error: false
    driver-class-name: com.mysql.cj.jdbc.Driver

（左右滑动查看全部代码）

4. 本地构建alpha-server（和上面步骤一致）

5. 启动alpha-server

java -Dspring.profiles.active=mysql -Dloader.path=./plugins -D"spring.datasource.url=jdbc:mysql://${host_address}:3306/${database_name}? serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8&useSSL=false " -jar alpha-server-${saga_version}-exec.jar

3.2 Omega配置

配置完alpha-server之后，就相当于分布式事务的协调器已经配置完成，剩下的就是omega的配置，也就是在实际开发中如何运用servicecomb-pack去处理分布式事务。本次讲解会结合一个实际案例：购物系统中的下单流程和删除产品流程来分别讲解saga模式和tcc模式如何使用的。

3.2.1 环境准备

本次案例：购物系统是采用分布式微服务架构，整体分为三个微服务应用：orderManage订单管理应用、productManage产品管理应用、stockManage库存管理应用

1. 添加依赖

分别在三个应用的pom文件中添加Omega所需的依赖：

<dependency>
    <groupId>org.apache.servicecomb.pack</groupId>
    <artifactId>omega-spring-starter</artifactId>
    <version>${servicecomb-pack.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.servicecomb.pack</groupId>
    <artifactId>omega-transport-resttemplate</artifactId>
    <version>${servicecomb-pack.version}</version>
</dependency>
<!--非必需 -->
<dependency>
    <groupId>org.apache.servicecomb.pack</groupId>
    <artifactId>omega-spring-cloud-consul-starter</artifactId>
    <version>${servicecomb-pack.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.servicecomb.pack</groupId>
    <artifactId>omega-spring-cloud-eureka-starter</artifactId>
    <version>${servicecomb-pack.version}</version>
</dependency>

*注意：请将${servicecomb-pack.version}更改为实际的版本号(推荐版本为0.4.0)

*注意：如需做集群，omega-spring-cloud-consul-starter和omega-spring-cloud-eureka-starter二选一，视项目的注册中心而定。

2. 修改配置文件

分别在三个应用的application.yml配置文件中添加alpha-server配置，具体配置如下：

#配置alpha-server地址
alpha:
  cluster:
    address: 10.15.15.172:8080
omega:
  enabled: true

注意：application.name一定不要过长，因为instanceId的格式是application.name+IP,并且长度为36，否则alpha-server事务持久化会报错

以上两个属性配置为必填，因为alpha-server会依据application.name去查找对应的Omega，其他应用配置自行添加，address可根据alpha-server中的配置实际添加

至此，环境准备已经完毕，下面开始进行应用代码编写。

3.2.2 saga模式代码编写

在本次案例中，我们以一个下单流程来讲解saga模式下代码是如何编写的。下单流程包括：点击下单、查询库存、支付、更新库存；订单应用作为起始服务，调用库存应用和产品应用，这两个应用对应的服务作为参与服务(子事务)，在订单应用下单，订单应用使用rest template向产品应用发起调用校验产品库存，然后订单应用向库存应用发起支付请求(子事务1)，支付成功后订单应用再向库存应用发起请求更新库存(子事务2)。

1. @SagaStart

首先需要在应用代码中描述出saga事务的边界，作为分布式事务的起始点，因此我们需要在订单应用中的createOrder()方法上添加该注解@SagaStart：

2. @ Compensable

支付对应补偿方法：

更新库存：

更新库存补偿方法：

*注意：实现的服务和补偿方法必须满足幂等的要求

*注意：默认情况下，超时需要显示声明

*注意：若全局事务起点与子事务重合，需同时声明@SagaStart和@Compensable注解

*注意：补偿方法的入参必须与try方法入参一致，否则启动时会报错(alpha-server找不到补偿方法)

3.2.3 tcc模式代码编写

下面我们会以删除库存流程来讲解tcc模式是如何编写代码的。删除库存流程：由产品应用发起(分布式事务起始)，调用库存应用删除对应产品的库存信息(tcc子事务)。

本次调用使用的是feign的方式，因此需要在产品应用中的pom文件添加相应的依赖：

1. @TccStart

我们以产品应用中的delete方法作为分布式事务起始点，因此在该方法上添加注解@TccStart:

2. @ Participate

在子事务所处的方法上添加该注解，并通过confirmMethod 以及cancelMethod属性定义相关确认以及取消方法名。这里需要注意的是这里提到的confirm，cancel方法的参数必须和try方法的相同。

Cancel逻辑：

*注意：confirm和cancel方法的入参必须和try方法一致

*注意：目前tcc模式还不支持timeout

3.2.4事件信息获取

默认情况下，8080端口用来处理Omega处发起的grpc请求，用来做事务上下文等操作；而8090端口则用于处理查询alpha处的事件信息。

1. saga-事件信息查询api

统计所有事件状态：

http://${alpha-server.address:port}/saga/stats

统计最近事件状态：

http://${alpha-server.address:port}/saga/recent

根据事件状态查询事件列表：

http://${alpha-server.address:port}/saga/transactions

根据服务名称查询对应的分布式事件列表：

http://${alpha-server.address:port}/saga/findTransactions

2. tcc-事件信息查询api

Tcc目前没有提供正式的查询接口。但是有测试接口，在AlphaTccEventController中，可自行根据测试接口修改源码，重新编译即可。

目前alpha-server提供的事件查询api不多，若有其他需求，用户可自行编写接口对数据库进行查询。

本文所有观点都出自个人见解，疏漏、错误之处在所难免，欢迎大家指正，希望能够与大家一起交流和进步。

[1]引用自：

http://servicecomb.apache.org/cn/docs/distributed-transaction-of-services-1/

[2]引用自：

http://servicecomb.apache.org/cn/docs/distributed-transaction-of-services-1/