猿创征文｜手把手教你微服务分布式事务与Seata框架源码分析

一，卷首语

将自己热爱与信仰的技术,持续不辍地传递。—— 步道师

二，Seata介绍

1，Seata 是什么?

Seata 是一款开源的分布式事务解决方案，致力于提供高性能和简单易用的分布式事务服务。

下面是一个分布式事务在Seata中的执行流程(后面我们会详细讲解)：

• TM 向 TC 申请开启一个全局事务，全局事务创建成功并生成一个全局唯一的XID。
• XID在微服务调用链路的上下文中传播。
• RM 向 TC 注册分支事务，接着执行这个分支事务并提交（重点：RM在第一阶段就已经执行了本地事务的提交/回滚），最后将执行结果汇报给TC。
• TM 根据 TC 中所有的分支事务的执行情况，发起全局提交或回滚决议。
• TC 调度XID下全部分支事务完成提交或回滚请求。

2，seata提供4种事务模式

在了解Seata的分布式事务解决方案是业务层面，只依赖于单台数据库的事务能力。Seata框架中一个分布式事务包含3中角色：
Transaction Coordinator (TC)： 事务协调器，维护全局事务的运行状态，负责协调并驱动全局事务的提交或回滚。
Transaction Manager ™： 控制全局事务的边界，负责开启一个全局事务，并最终发起全局提交或全局回滚的决议。
Resource Manager (RM)： 控制分支事务，负责分支注册、状态汇报，并接收事务协调器的指令，驱动分支（本地）事务的提交和回滚。

2.1 AT模式

基于支持本地 ACID 事务的关系型数据库。
Java 应用，通过 JDBC 访问数据库。

2.1.1 什么是ACID?

严格意义上的事务实现应该是具备原子性、一致性、隔离性和持久性，简称ACID。

• 原子性（(Atomicity) : 指的是一个事务（transaction）中的所有操作，要么全部完成，要么全部不完成，不会结束在中间某个环节。
• 一致性(Consistency): 指的是在事务开始之前和事务结束以后，数据库的完整性没有被破坏。
• 隔离性(lsolation): 指的是多个事务并发执行的时候不会互相干扰，即一个事务内部的数据对于其他事务来说是隔离的。
• 持久性(Durability): 指的是一个事务完成了之后数据就被永远保存下来，之后的其他操作或故障都不会对事务的结果产生影响。

2.1.2 2PC

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

这里把这两种情况在这里给大家简单的描述了下

2.2 TCC模式

一个分布式的全局事务，整体是 两阶段提交 的模型。全局事务是由若干分支事务组成的，分支事务要满足 两阶段提交 的模型要求，即需要每个分支事务都具备自己的：
一阶段 prepare 行为
二阶段 commit 或 rollback 行为

根据两阶段行为模式的不同，我们将分支事务划分为 Automatic (Branch) Transaction Mode 和 Manual (Branch) Transaction Mode.

AT 模式基于 支持本地 ACID 事务 的 关系型数据库：

• 一阶段 prepare 行为：在本地事务中，一并提交业务数据更新和相应回滚日志记录。

• 二阶段 commit 行为：马上成功结束，自动 异步批量清理回滚日志。

• 二阶段 rollback 行为：通过回滚日志，自动 生成补偿操作，完成数据回滚。
  相应的，TCC 模式，不依赖于底层数据资源的事务支持：

• 一阶段 prepare 行为：调用 自定义 的 prepare 逻辑。

• 二阶段 commit 行为：调用 自定义 的 commit 逻辑。

• 二阶段 rollback 行为：调用 自定义 的 rollback 逻辑。
  所谓 TCC 模式，是指支持把 自定义 的分支事务纳入到全局事务的管理中。

2.3 SAGA模式

Saga模式是SEATA提供的长事务解决方案，在Saga模式中，业务流程中每个参与者都提交本地事务，当出现某一个参与者失败则补偿前面已经成功的参与者，一阶段正向服务和二阶段补偿服务都由业务开发实现。

适用场景:

• 业务流程长、业务流程多
• 参与者包含其它公司或遗留系统服务，无法提供 TCC 模式要求的三个接口

优势:

• 一阶段提交本地事务，无锁，高性能
• 事件驱动架构，参与者可异步执行，高吞吐
• 补偿服务易于实现

缺点：

• 不保证隔离性

2.4 XA模式

在 Seata 定义的分布式事务框架内，利用事务资源（数据库、消息服务等）对 XA 协议的支持，以 XA 协议的机制来管理分支事务的一种 事务模式。

执行阶段：

• 可回滚：业务 SQL 操作放在 XA 分支中进行，由资源对 XA 协议的支持来保证 可回滚
• 持久化：XA 分支完成后，执行 XA prepare，同样，由资源对 XA 协议的支持来保证 持久化（即，之后任何意外都不会造成无法回滚的情况）

完成阶段：

• 分支提交：执行 XA 分支的 commit
• 分支回滚：执行 XA 分支的 rollback

三，Seata 安装

1，下载seata

官网下载地址： https://seata.io/zh-cn/blog/download.html

2，安装seata

解压seata-server-1.5.2安装包。解压出来我们进入bin目录会看到相应的启动脚本(由于这里是在Windows演示)

在 Linux/Mac 下

sh ./bin/seata-server.sh
1

在 Windows 下

bin\seata-server.bat
1

高可用部署：
Seata 的高可用依赖于注册中心、配置中心和数据库来实现。使用nacos和redis为例

这里我们看到是基于文件的方式，后面我们会有详细的课程讲解基于nacos注册中心的安装配置等

3，运行seata

• windows：bin目录下双击seata-server.bat启动

• linux：命令行启动 seata-server.sh -h 127.0.0.1 -p 8091

启动参数：

• -h: 注册到注册中心的ip
• -p: Server rpc 监听端口
• -m: 全局事务会话信息存储模式，nacos, consul, apollo, zk, etcd3，优先读取启动参数
• -n: Server node，多个Server时，需区分各自节点，用于生成不同区间的transactionId，以免冲突
• -e: 环境配置
  

四，项目环境搭建

1，增加 Maven 依赖

首先，您需要将 nacos-client 的 Maven 依赖添加到您的项目 pom.xml 文件中

    io.seata</groupId>
    seata-spring-boot-starter</artifactId>
    最新版</version>
</dependency>

    com.alibaba.nacos</groupId>
    nacos-client</artifactId>
    1.2.0及以上版本</version>
</dependency>
1
2
3
4
5
6
7
8
9
10

2，Client端配置中心

在 application.yml 中加入对应的配置中心,其余配置参考

seata:
  config:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848
      group : "SEATA_GROUP"
      namespace: ""
      username: "XXXXXX"
1
2
3
4
5
6
7
8

3，Server端配置中心

在 registry.conf 中加入对应配置中心,其余配置参考

config {
  type = "nacos"

  nacos {
    serverAddr = "127.0.0.1:8848"
    group = "SEATA_GROUP"
    namespace = ""
    username = "nacos"
  }
}
1
2
3
4
5
6
7
8
9
10

4，上传配置至Nacos配置中心

seata:
  config:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848
      group : "SEATA_GROUP"
      namespace: ""
      dataId: "seataServer.properties"
      username: "nacos"
1
2
3
4
5
6
7
8
9

五，源码分析

今天咱们开始咱们的主题了，在了解以上问题之后我们是不是比较好奇他的运行流程。了解执行流程我们就要从官网，或者源码入手来了解

1，启动类

// 【排除SpringBoot启动时自动加载GlobalTransactionAutoConfiguration.class】
@SpringBootApplication(exclude = {GlobalTransactionAutoConfiguration.class})
@EnableFeignClients   // 远程调用
@EnableDiscoveryClient   // 开启nacos服务
public class GatewayApplication{
   public static void main(String[] args) {
     ErosApplication.run(ProjectNameConstatnt.GATEWAY_NAME,"consume",GatewayApplication.class,args);
  }
}
1
2
3
4
5
6
7
8
9

2，配置类

@GlobalTransactional(name="XXXXX",rollbackFor = RuntimeException.class)
1

3，@GlobalTransactional

方法上加了@GlobalTransactional，Seata通过aop检测到之后，就会使用TM和TC通信，注册全局事务。
在@GlobalTransactional涵括的代码中，不管是本服务中的sql操作，还是feign调用别的服务的sql操作，只要sql操作满足如下：insert操作，delete操作，update操作，select for update操作。 就会被seata增强，使用RM与TC通信，注册分支事务。
全局事务：包括开启事务、提交、回滚、获取当前状态等方法。

public interface GlobalTransaction {

    /**
     * 开启一个全局事务（使用默认的事务名和超时时间）
     */
    void begin() throws TransactionException;

    /**
     * 开启一个全局事务，并指定超时时间（使用默认的事务名）
     */
    void begin(int timeout) throws TransactionException;

    /**
     * 开启一个全局事务，并指定事务名和超时时间
     */
    void begin(int timeout, String name) throws TransactionException;

    /**
     * 全局提交
     */
    void commit() throws TransactionException;

    /**
     * 全局回滚
     */
    void rollback() throws TransactionException;

    /**
     * 获取事务的当前状态
     */
    GlobalStatus getStatus() throws TransactionException;

    /**
     * 获取事务的 XID
     */
    String getXid();

}
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

GlobalTransactionScanner继承自AbstractAutoProxyCreator，在这里拦截到加了@GlobalTransactional的方法。
GlobalTransactionScanner

• AbstractAutoProxyCreator：wrapIfNecessary（aop的核心），getAdvicesAndAdvisorsForBean（拦截器）
• InitializingBean：afterPropertiesSet（初始化TM,RM）

3.1 初始化TM,RM

@Override
    public void afterPropertiesSet() {
        //是否禁止了全局事务
        if (disableGlobalTransaction) {
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Global transaction is disabled.");
            }
            return;
        }
        //初始化netty 客户端（TM  RM）
        initClient();
    }
1
2
3
4
5
6
7
8
9
10
11
12

初始化TM,RM

private void initClient() {  
        //init TM
        TMClient.init(applicationId, txServiceGroup);
        
        //init RM
        RMClient.init(applicationId, txServiceGroup);
    }
1
2
3
4
5
6
7

GlobalTransactionalInterceptor#invoke

3.2 TransactionalTemplate

GlobalTransaction 和 GlobalTransactionContext API 把一个业务服务的调用包装成带有分布式事务支持的服务。

public class TransactionalTemplate {

    public Object execute(TransactionalExecutor business) throws TransactionalExecutor.ExecutionException {

        // 1. 获取当前全局事务实例或创建新的实例
        GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();

        // 2. 开启全局事务
        try {
            tx.begin(business.timeout(), business.name());

        } catch (TransactionException txe) {
            // 2.1 开启失败
            throw new TransactionalExecutor.ExecutionException(tx, txe,
                TransactionalExecutor.Code.BeginFailure);

        }

        Object rs = null;
        try {
            // 3. 调用业务服务
            rs = business.execute();

        } catch (Throwable ex) {

            // 业务调用本身的异常
            try {
                // 全局回滚
                tx.rollback();

                // 3.1 全局回滚成功：抛出原始业务异常
                throw new TransactionalExecutor.ExecutionException(tx, TransactionalExecutor.Code.RollbackDone, ex);

            } catch (TransactionException txe) {
                // 3.2 全局回滚失败：
                throw new TransactionalExecutor.ExecutionException(tx, txe,
                    TransactionalExecutor.Code.RollbackFailure, ex);

            }

        }

        // 4. 全局提交
        try {
            tx.commit();

        } catch (TransactionException txe) {
            // 4.1 全局提交失败：
            throw new TransactionalExecutor.ExecutionException(tx, txe,
                TransactionalExecutor.Code.CommitFailure);

        }
        return rs;
    }

}
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

DataSourceProxy#getConnection

@Override
    public ConnectionProxy getConnection() throws SQLException {
        Connection targetConnection = targetDataSource.getConnection();
        return new ConnectionProxy(this, targetConnection);
    }
1
2
3
4
5

ConnectionProxy#doCommit

private void doCommit() throws SQLException {
		//处理@GlobalTransaction的分支事务
        if (context.inGlobalTransaction()) {
            processGlobalTransactionCommit();
        } 
        //处理@GlobalLock，即检查一下是否可以获取全局锁
		else if (context.isGlobalLockRequire()) {
            processLocalCommitWithGlobalLocks();
        } else {
            targetConnection.commit();
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12

ConnectionProxy#processGlobalTransactionCommit

模板的异常方法

class ExecutionException extends Exception {

        // 发生异常的事务实例
        private GlobalTransaction transaction;

        // 异常编码：
        // BeginFailure（开启事务失败）
        // CommitFailure（全局提交失败）
        // RollbackFailure（全局回滚失败）
        // RollbackDone（全局回滚成功）
        private Code code;

        // 触发回滚的业务原始异常
        private Throwable originalException;
1
2
3
4
5
6
7
8
9
10
11
12
13
14

       以上就是我们今天的教程，如果本文对你有所帮助，欢迎关注点赞，分享给您身边的朋友。您的鼓励就是对我的最大动力。

• 注：本文部分图片素材来源于 互联网 。