【微服务的绊脚石--分布式事务】 Seata-AT模式深入分

发表于

前言

在上一篇[微服务的绊脚石–分布式事务] SEATA解决方案介绍中，介绍了微服务架构的问题之一：分布式事务，以及业界常见的解决方案。这一篇，针对我和身边同事在学习Seata过程中遇到的各种问题，结合当前最新的版本Seata 1.4.2的代码实现，跟大家一起深入了解一下Seata。主要的问题有：

  1. Seata的核心竞争力是什么？
  2. Seata内部有哪些模块？
  3. AT 与 XA 方案有什么不同？
  4. 具体Commit/Rollback流程是怎样的？
  5. AT模式是如何初始化的？
  6. RM为什么不需要@GlobalTransactional？
  7. @GlobalLock有什么用？
  8. 为什么ExceptionHandler会导致全局事务失效？

Seata主要的竞争力是什么？

上一篇中我们介绍过2PC/XA，TCC和SAGA方案。2PC/XA的优点是对业务代码无侵入，但是它的缺点也是很明显：必须要求数据库对 XA 协议的支持，且由于 XA 协议自身的特点，它会造成事务资源长时间得不到释放，锁定周期长，而且在应用层上面无法干预，因此性能很差，属于杀敌一千自损八百。而TCC和SAGA方案都是业务侵入式的，提交逻辑的实现必然伴随着回滚逻辑（或者补偿逻辑）的实现，这样的代码会变得非常臃肿，维护成本高。

据阿里工程师介绍，AT模式作为 Seata的默认模式，虽然也是类似于 XA 方案的两段式提交方案，但一开始就是冲着业务无侵入性与高性能方向走，这正是我们对解决分布式事务问题迫切的需求。

Seata内部有哪些模块？

Seata-AT 的设计思路是将一个分布式事务作为一个全局事务，在下面挂若干个分支事务，而一个分支事务是一个满足 ACID 的本地事务，因此我们可以像操作本地事务一样操作分布式事务。

Seata 内部定义了 3个模块来处理全局事务和分支事务，这三个组件分别是：

  • Transaction Coordinator (TC)：事务协调器，维护全局事务的运行状态，独立部署，负责协调并驱动全局事务的提交或回滚。
  • Transaction Manager (TM)：控制全局事务的边界，负责开启一个全局事务，并最终发起全局提交或全局回滚的决议。
  • Resource Manager (RM)：控制分支事务，负责分支注册、状态汇报，并接收事务协调器的指令，驱动分支（本地）事务的提交和回滚。

image.png

全局事务的执行步骤：

  1. TM 向 TC 申请开启一个全局事务，TC 创建全局事务后返回全局唯一的 XID，XID 会在全局事务的上下文中传播
  2. RM 向 TC 注册分支事务，该分支事务归属于拥有相同 XID 的全局事务；
  3. TM 向 TC 发起全局提交或回滚；
  4. TC 调度 XID 下的分支事务完成提交或者回滚。

AT 与 XA 方案有什么不同？

Seata 的事务提交方式跟 XA 协议的两段式提交在总体思路上来说基本是一致的，那它们之间有什么不同呢？

我们都知道 XA 协议它依赖的是数据库层面来保障事务的一致性，也即是说 XA 的各个分支事务是在数据库层面上驱动的，由于 XA 的各个分支事务需要有 XA 的驱动程序，一方面会导致数据库与 XA 驱动耦合，另一方面它会导致各个分支的事务资源锁定周期长，这也是它没有在互联网公司流行的重要因素。

基于 XA 协议以上的问题，Seata 另辟蹊径，既然在依赖数据库层会导致这么多问题，那我就从应用层做手脚，这还得从 Seata 的 RM 模块说起，前面也说过 RM 的主要作用了，其实 RM 在内部做了对数据库对象，如DataSource， Connection， Statement做了一层代理。

image.png

Seata 对数据源做了代理，所以我们在使用 Seata 时，实际上用的数据源是 Seata 自带的数据源代理 DataSourceProxy，Seata 在这层代理中加入了很多逻辑，主要是解析 SQL，把业务数据在更新前后的数据镜像组织成回滚日志，并将 UndoLog 日志插入 undo_log 表中，保证每条更新数据的业务 sql 都有对应的回滚日志存在。

这样做的好处就是，本地事务执行完可以立即释放本地事务锁定的资源，然后向 TC 上报分支状态。当 TM 决定全局提交时，就不需要同步协调处理了，TC 会异步调度各个 RM 分支事务删除对应的 UndoLog 日志即可，这个步骤非常快速地可以完成；当 TM 决议全局回滚时，RM 收到 TC 发送的回滚请求，RM 通过 XID 找到对应的 UndoLog 回滚日志，然后执行回滚日志完成回滚操作。

image.png
如上图所示，XA 方案的 RM 是放在数据库层的，它依赖了数据库的 XA 驱动程序。

image.png

而Seata 的 RM 实际上是已中间件的形式放在应用层，不用依赖数据库对协议的支持，完全剥离了分布式事务方案对数据库在协议支持上的要求。

具体Commit/Rollback流程是怎样的？

image.png

概括来讲，AT 模式的工作流程分为两个阶段。一阶段进行业务 SQL 执行，并通过 SQL 拦截、SQL 改写等过程生成修改数据前后的快照（Image），并作为 UndoLog 和业务修改在同一个本地事务中提交。

如果一阶段成功那么二阶段仅仅异步删除刚刚插入的 UndoLog；如果二阶段失败则通过 UndoLog 生成反向 SQL 语句回滚一阶段的数据修改。其中关键的 SQL 解析和拼接工作借助了 Druid Parser 中的代码，这部分本文并不涉及，感兴趣的小伙伴可以去翻看源码，并不是很复杂。

下面，我们以上一篇中的order_tbl为例来说明整个 AT 分支的工作过程。

业务表：order_tbl

Field Type Key
id int PRI
user_id varchar(255)
commodity_code varchar(255)
count int
money int

AT 分支事务的业务逻辑：

1
sql复制代码insert into order_tbl values (12, '1002', '2001', 1, 5);

一阶段

image.png

过程：

  1. 解析 SQL：得到 SQL 的类型（INSERT），表（order_tbl），条件等相关的信息。查询前镜像：根据解析得到的条件信息，生成查询语句，定位数据。如果是更新数据，会根据更新语句的条件，生成如下SQL查询前镜像。
1
bash复制代码select id, user_id, commodity_code， count， money from product where id = 12;
  1. 执行业务 SQL：插入这条记录。
  2. 查询后镜像：根据前镜像的结果，通过 主键 定位数据。
1
bash复制代码select id, user_id, commodity_code， count， money from product where id = 12;

得到后镜像：

id user_id commodity_code count money
1 1002 2001 1 5
  1. 插入回滚日志：把前后镜像数据以及业务 SQL 相关的信息组成一条回滚日志记录，插入到 UNDO_LOG 表中。
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
less复制代码{
    "@class": "io.seata.rm.datasource.undo.BranchUndoLog",
    "xid": "172.27.0.2:8091:18207193960247322",
    "branchId": 18207193960247324,
    "sqlUndoLogs": [
        "java.util.ArrayList",
        [
            {
                "@class": "io.seata.rm.datasource.undo.SQLUndoLog",
                "sqlType": "INSERT",
                "tableName": "order_tbl",
                "beforeImage": {
                    "@class": "io.seata.rm.datasource.sql.struct.TableRecords$EmptyTableRecords",
                    "tableName": "order_tbl",
                    "rows": [
                        "java.util.ArrayList",
                        []
                    ]
                },
                "afterImage": {
                    "@class": "io.seata.rm.datasource.sql.struct.TableRecords",
                    "tableName": "order_tbl",
                    "rows": [
                        "java.util.ArrayList",
                        [
                            {
                                "@class": "io.seata.rm.datasource.sql.struct.Row",
                                "fields": [
                                    "java.util.ArrayList",
                                    [
                                        {
                                            "@class": "io.seata.rm.datasource.sql.struct.Field",
                                            "name": "id",
                                            "keyType": "PRIMARY_KEY",
                                            "type": 4,
                                            "value": 12
                                        },
                                        {
                                            "@class": "io.seata.rm.datasource.sql.struct.Field",
                                            "name": "user_id",
                                            "keyType": "NULL",
                                            "type": 12,
                                            "value": "1002"
                                        },
                                        {
                                            "@class": "io.seata.rm.datasource.sql.struct.Field",
                                            "name": "commodity_code",
                                            "keyType": "NULL",
                                            "type": 12,
                                            "value": "2001"
                                        },
                                        {
                                            "@class": "io.seata.rm.datasource.sql.struct.Field",
                                            "name": "count",
                                            "keyType": "NULL",
                                            "type": 4,
                                            "value": 1
                                        },
                                        {
                                            "@class": "io.seata.rm.datasource.sql.struct.Field",
                                            "name": "money",
                                            "keyType": "NULL",
                                            "type": 4,
                                            "value": 5
                                        }
                                    ]
                                ]
                            }
                        ]
                    ]
                }
            }
        ]
    ]
}
  1. 提交前，向 TC 注册分支：申请 order_tbl 表中，主键值等于 12 的记录的 全局锁 。
  2. 本地事务提交：业务数据的更新和前面步骤中生成的 UndoLog 一并提交。
  3. 将本地事务提交的结果上报给 TC。

二阶段-回滚

image.png

  1. 收到 TC 的分支回滚请求，开启一个本地事务，通过 XID 和 Branch ID 查找到相应的 UndoLog 记录。
  2. 数据校验：拿 UndoLog 中的后镜与当前数据进行比较，如果有不同，说明数据被当前全局事务之外的动作做了修改。这种情况，需要根据配置策略来做处理。
  3. 根据 UndoLog 中的前镜像和业务 SQL 的相关信息生成并执行回滚的语句：
1
ini复制代码delete from order_tbl where id = 12;
  1. 删除UndoLog。
  2. 提交本地事务。
  3. 把本地事务的执行结果（即分支事务回滚的结果）上报给 TC。

二阶段-提交

image.png

  1. 收到 TC 的分支提交请求，把请求放入一个异步任务的队列中，马上返回提交成功的结果给 TC。查找UndoLog。
  2. 批量地删除相应 UndoLog 记录。
  3. 开始提交本地事务。
  4. 向TC汇报本地事务结果。

AT模式是如何初始化的？

整个全局事务是有TM负责开启的，上一篇的代码中是在BusinessService中开始全局事务的，我们注意到这里有一个@GlobalTransactional注解，详细的io.seata.spring.annotation.GlobalTransactional的代码可以参考这里。

1
2
3
4
5
6
7
java复制代码//BusinessService
@GlobalTransactional
public void purchase(String userId, String commodityCode, int orderCount) {
    LOGGER.info("purchase begin ... xid: " + RootContext.getXID()); 
    storageClient.deduct(commodityCode, orderCount); 
    orderClient.create(userId, commodityCode, orderCount); 
}

在同一个包下，还有一个io.seata.spring.annotation.GlobalTransactionScanner，它继承了org.springframework.aop.framework.autoproxy.AbstractAutoProxyCreator,在它的initClient()方法中，对TM和RM进行了初始化。

1
2
3
4
5
6
7
8
9
10
java复制代码    private void initClient() {
        // ...
        //init TM
        TMClient.init(applicationId, txServiceGroup, accessKey, secretKey);
        // ...
        //init RM
        RMClient.init(applicationId, txServiceGroup);
        // ...
        registerSpringShutdownHook();
    }

TM的初始化

在TMClient 的init 方法中获取了io.seata.core.rpc.netty.TmNettyRemotingClient的实例，用于处理各种与服务端的消息交互。

1
2
3
4
5
java复制代码// TMClient
public static void init(String applicationId, String transactionServiceGroup, String accessKey, String secretKey) {
    TmNettyRemotingClient tmNettyRemotingClient = TmNettyRemotingClient.getInstance(applicationId, transactionServiceGroup, accessKey, secretKey);
    tmNettyRemotingClient.init();
}

TmNettyRemotingClient继承了io.seata.core.rpc.netty.AbstractNettyRemotingClient，在AbstractNettyRemotingClient的init方法中，设置了定时任务用于定时重发 RegisterTMRequest（RM 客户端会发送 RegisterRMRequest）请求尝试连接服务端，具体逻辑是： NettyClientChannelManager 中的 channels 中缓存了客户端 channel，如果此时 channels 不存在或者已过期，那么就会尝试连接服务端以重新获取 channel 并将其缓存到 channels 中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
java复制代码@Override
    public void init() {
        timerExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                clientChannelManager.reconnect(getTransactionServiceGroup());
            }
        }, SCHEDULE_DELAY_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);
        if (NettyClientConfig.isEnableClientBatchSendRequest()) {
            mergeSendExecutorService = new ThreadPoolExecutor(MAX_MERGE_SEND_THREAD,
                MAX_MERGE_SEND_THREAD,
                KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));
            mergeSendExecutorService.submit(new MergedSendRunnable());
        }
        super.init();
        clientBootstrap.start();
    }

RM的初始化

在RMClient的init方法中，RmNettyRemotingClient.getInstance 处理逻辑与 TM 大致相同；ResourceManager 是 RM 资源管理器，负责分支事务的注册、提交、上报、以及回滚操作，以及全局锁的查询操作，DefaultResourceManager 会持有当前所有的 RM 资源管理器，进行统一调用处理。

TransactionMessageHandler 是 RM 消息处理器，用于负责处理从 TC 发送过来的指令，并对分支进行分支提交、分支回滚，以及 UndoLog 删除操作；最后 init 方法跟 TM 逻辑也大体一致；DefaultRMHandler 封装了 RM 分支事务的一些具体操作逻辑。

1
2
3
4
5
6
java复制代码public static void init(String applicationId, String transactionServiceGroup) {
    RmNettyRemotingClient rmNettyRemotingClient = RmNettyRemotingClient.getInstance(applicationId, transactionServiceGroup);
    rmNettyRemotingClient.setResourceManager(DefaultResourceManager.get());
    rmNettyRemotingClient.setTransactionMessageHandler(DefaultRMHandler.get());
    rmNettyRemotingClient.init();
}

添加拦截器

在GlobalTransactionScanner的wrapIfNecessary方法中，会扫描带有@GlobalTransactional，@GlobalLock等注解的方法，并添加对应的拦截器。

  1. 判断是否存在对应的注解
  2. 创建拦截器
  3. 将拦截器添加到目标对象
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
java复制代码/**
 * The following will be scanned, and added corresponding interceptor
 */
@Override
protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
    try {
        synchronized (PROXYED_SET) {
            if (PROXYED_SET.contains(beanName)) {
                return bean;
            }
            interceptor = null;
            //check TCC proxy
            if (TCCBeanParserUtils.isTccAutoProxy(bean, beanName, applicationContext)) {
                //TCC interceptor, proxy bean of sofa:reference/dubbo:reference, and LocalTCC
                interceptor = new TccActionInterceptor(TCCBeanParserUtils.getRemotingDesc(beanName));
                ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
                    (ConfigurationChangeListener)interceptor);
            } else {
                Class<?> serviceInterface = SpringProxyUtils.findTargetClass(bean);
                Class<?>[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean);

                //#### <1> ####
                if (!existsAnnotation(new Class[]{serviceInterface})
                    && !existsAnnotation(interfacesIfJdk)) {
                    return bean;
                }
                
                if (globalTransactionalInterceptor == null) {
                    //#### <2> ####
                    globalTransactionalInterceptor = new GlobalTransactionalInterceptor(failureHandlerHook);
                    ConfigurationCache.addConfigListener(
                        ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
                        (ConfigurationChangeListener)globalTransactionalInterceptor);
                }
                interceptor = globalTransactionalInterceptor;
            }

            LOGGER.info("Bean[{}] with name [{}] would use interceptor [{}]", bean.getClass().getName(), beanName, interceptor.getClass().getName());
            if (!AopUtils.isAopProxy(bean)) {
                bean = super.wrapIfNecessary(bean, beanName, cacheKey);
            } else {
                //#### <3> ####
                AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean);
                Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));
                for (Advisor avr : advisor) {
                    advised.addAdvisor(0, avr);
                }
            }
            PROXYED_SET.add(beanName);
            return bean;
        }
    } catch (Exception exx) {
        throw new RuntimeException(exx);
    }
}

事务处理

以全局事务@GlobalTransactional为例，在io.seata.spring.annotation.GlobalTransactionalInterceptor的invoke方法中，handleGlobalTransaction方法，在该方法中又调用了io.seata.tm.api.TransactionalTemplate的execute方法。

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
java复制代码@Override
public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
    Class<?> targetClass =
        methodInvocation.getThis() != null ? AopUtils.getTargetClass(methodInvocation.getThis()) : null;
    Method specificMethod = ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass);
    if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) {
        final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
        final GlobalTransactional globalTransactionalAnnotation =
            getAnnotation(method, targetClass, GlobalTransactional.class);
        final GlobalLock globalLockAnnotation = getAnnotation(method, targetClass, GlobalLock.class);
        boolean localDisable = disable || (degradeCheck && degradeNum >= degradeCheckAllowTimes);
        if (!localDisable) {
            if (globalTransactionalAnnotation != null) {
                return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);
            } else if (globalLockAnnotation != null) {
                return handleGlobalLock(methodInvocation, globalLockAnnotation);
            }
        }
    }
    return methodInvocation.proceed();
}

Object handleGlobalTransaction(final MethodInvocation methodInvocation,
    final GlobalTransactional globalTrxAnno) throws Throwable {
    boolean succeed = true;
    try {
        return transactionalTemplate.execute(new TransactionalExecutor() {
            @Override
            public Object execute() throws Throwable {
                return methodInvocation.proceed();
            }
            // ...
        });
    }
    // ...
}

在TransactionalTemplate的execute方法中执行了具体事务处理，比如开启事务、提交、回滚等等。

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
java复制代码public Object execute(TransactionalExecutor business) throws Throwable {
    // 1. Get transactionInfo
    TransactionInfo txInfo = business.getTransactionInfo();
    if (txInfo == null) {
        throw new ShouldNeverHappenException("transactionInfo does not exist");
    }
    // 1.1 Get current transaction, if not null, the tx role is 'GlobalTransactionRole.Participant'.
    GlobalTransaction tx = GlobalTransactionContext.getCurrent();

    // 1.2 Handle the transaction propagation.
    Propagation propagation = txInfo.getPropagation();
    SuspendedResourcesHolder suspendedResourcesHolder = null;
    try {
        // ...

        // 1.3 If null, create new transaction with role 'GlobalTransactionRole.Launcher'.
        if (tx == null) {
            tx = GlobalTransactionContext.createNew();
        }

        // set current tx config to holder
        GlobalLockConfig previousConfig = replaceGlobalLockConfig(txInfo);

        try {
            // 2. If the tx role is 'GlobalTransactionRole.Launcher', send the request of beginTransaction to TC,
            //    else do nothing. Of course, the hooks will still be triggered.
            beginTransaction(txInfo, tx);

            Object rs;
            try {
                // Do Your Business
                rs = business.execute();
            } catch (Throwable ex) {
                // 3. The needed business exception to rollback.
                completeTransactionAfterThrowing(txInfo, tx, ex);
                throw ex;
            }

            // 4. everything is fine, commit.
            commitTransaction(tx);

            return rs;
        } finally {
            //5. clear
            resumeGlobalLockConfig(previousConfig);
            triggerAfterCompletion();
            cleanUp();
        }
    } finally {
        // If the transaction is suspended, resume it.
        if (suspendedResourcesHolder != null) {
            tx.resume(suspendedResourcesHolder);
        }
    }
}

RM为什么不需要@GlobalTransactional？

在上一篇的代码示例中，我们只在TM端的BusinessService 方法上添加了@GlobalTransactional注解,而在下游微服务中并没有添加任何注解，为什么也可以当做全局事务处理呢？

事务上下文

我们先来看看 Seata 的事务上下文，它是由 RootContext 来管理的。

应用开启一个全局事务后，RootContext 会自动绑定该事务的 XID，事务结束（提交或回滚完成），RootContext 会自动解绑 XID。

1
2
3
4
5
ini复制代码// 绑定 XID
RootContext.bind(xid);

// 解绑 XID
String xid = RootContext.unbind();

应用可以通过 RootContext 的 API 接口来获取当前运行时的全局事务 XID。

1
2
ini复制代码// 获取 XID
String xid = RootContext.getXID();

应用是否运行在一个全局事务的上下文中，就是通过 RootContext 是否绑定 XID 来判定的。

1
2
3
csharp复制代码    public static boolean inGlobalTransaction() {
        return CONTEXT_HOLDER.get(KEY_XID) != null;
    }

事务传播

Seata 全局事务的传播机制就是指事务上下文的传播，根本上，就是 XID 的应用运行时的传播方式。

1. 服务内部的事务传播

默认的，RootContext 的实现是基于 ThreadLocal 的，即 XID 绑定在当前线程上下文中。

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
java复制代码public class ThreadLocalContextCore implements ContextCore {

    private ThreadLocal<Map<String, String>> threadLocal = new ThreadLocal<Map<String, String>>() {
        @Override
        protected Map<String, String> initialValue() {
            return new HashMap<String, String>();
        }

    };

    @Override
    public String put(String key, String value) {
        return threadLocal.get().put(key, value);
    }

    @Override
    public String get(String key) {
        return threadLocal.get().get(key);
    }

    @Override
    public String remove(String key) {
        return threadLocal.get().remove(key);
    }
}

所以服务内部的 XID 传播通常是天然的通过同一个线程的调用链路串连起来的。默认不做任何处理，事务的上下文就是传播下去的。

如果希望挂起事务上下文，则需要通过 RootContext 提供的 API 来实现：

1
2
3
4
5
6
7
scss复制代码// 挂起（暂停）
String xid = RootContext.unbind();

// TODO: 运行在全局事务外的业务逻辑

// 恢复全局事务上下文
RootContext.bind(xid);

2. 跨服务调用的事务传播

通过上述基本原理，我们可以很容易理解：

跨服务调用场景下的事务传播，本质上就是要把 XID 通过服务调用传递到服务提供方，并绑定到 RootContext 中去。

只要能做到这点，理论上 Seata 可以支持任意的微服务框架。

我们注意到，在Common模块的SeataFilter中，从Http请求Header中获取了全局事务ID XID，并将其设置到了事务上下文RootContext中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码@Override
public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws IOException, ServletException {
    HttpServletRequest req = (HttpServletRequest) servletRequest;
    String xid = req.getHeader(RootContext.KEY_XID.toLowerCase());
    boolean isBind = false;
    if (StringUtils.isNotBlank(xid)) {
        RootContext.bind(xid);
        isBind = true;
    }
    try {
        filterChain.doFilter(servletRequest, servletResponse);
    } finally {
        if (isBind) {
            RootContext.unbind();
        }
    }
}

而在SeataRestTemplateInterceptor中，先从RootContext获取了XID，然后设置到了Http请求头中，这样下游的RM就能通过SeataFilter获取到XID了。

1
2
3
4
5
6
7
8
9
10
java复制代码@Override
public ClientHttpResponse intercept(HttpRequest httpRequest, byte[] bytes, ClientHttpRequestExecution clientHttpRequestExecution) throws IOException {
    HttpRequestWrapper requestWrapper = new HttpRequestWrapper(httpRequest);
    String xid = RootContext.getXID();
    if (StringUtils.isNotEmpty(xid)) {
        requestWrapper.getHeaders().add(RootContext.KEY_XID, xid);
    }

    return clientHttpRequestExecution.execute(requestWrapper, bytes);
}

@GlobalLock有什么用？

前面初始化部分的代码中，也出现了对@GlobalLock的处理，那么它的用法跟 @GlobalTransactional 有什么区别呢？

如果是用 @GlobalLock 修饰的业务方法，虽然该方法并非某个全局事务下的分支事务，但是它对数据资源的操作也需要先查询全局锁，如果存在其他 Seata 全局事务正在修改，则该方法也需等待。所以，如果想要 Seata 全局事务执行期间，数据库不会被其他事务修改，则该方法需要强制添加 GlobalLock 注解，来将其纳入 Seata 分布式事务的管理范围。

功能有点类似于 Spring 的 @Transactional 注解，如果你希望开启事务，那么必须添加该注解，如果你没有添加那么事务功能自然不生效，业务可能出 BUG；Seata 也一样，如果你希望某个不在全局事务下的 SQL 操作不影响 AT 分布式事务，那么必须添加 GlobalLock 注解。

全局锁的组成和作用

Seata AT 模式的全局锁主要由表名加操作行的主键两个部分组成，通过在服务端保存全局锁的方法保证：

  1. 全局事务之前的写隔离
  2. 全局事务与被 GlobalLock 修饰方法间的写隔离性

全局锁的注册

当客户端在进行一阶段本地事务提交前，会先向服务端注册分支事务，此时会将修改行的表名、主键信息封装成全局锁一并发送到服务端进行保存，如果服务端保存时发现已经存在其他全局事务锁定了这些行主键，则抛出全局锁冲突异常，客户端循环等待并重试。

全局锁的查询

被 @GlobalLock 修饰的方法虽然不在某个全局事务下，但是其在提交事务前也会进行全局锁查询，如果发现全局锁正在被其他全局事务持有，则自身也会循环等待。

全局锁的释放

由于二阶段提交是异步进行的，当服务端向客户端发送 branch commit 请求后，客户端仅仅是将分支提交信息插入内存队列即返回，服务端只要判断这个流程没有异常就会释放全局锁。因此，可以说如果一阶段成功则在二阶段一开始就会释放全局锁，不会锁定到二阶段提交流程结束。

VTh76.png

但是如果一阶段失败二阶段进行回滚，则由于回滚是同步进行的，全局锁直到二阶段回滚完成才会被释放。

为什么ExceptionHandler会导致全局事务失效？

在SpringBoot项目中，我们经常使用@ControllerAdvice来构造ExceptionHandler，用于处理各种异常。有的时候会因此导致全局事务无法回滚，这是为什么呢？

以本文的 ApiExceptionHandler为例，

首先我们来看TM的执行，上文我们提到实际的全局事务处理，比如开启全局事务，提交，回滚，是在TransactionalTemplate的execute方法中实现的，所以为了能够回滚，就必须保证业务处理business.execute()会抛出异常。

1
2
3
4
5
6
7
8
java复制代码try {
    // Do Your Business
    rs = business.execute();
} catch (Throwable ex) {
    // 3. The needed business exception to rollback.
    completeTransactionAfterThrowing(txInfo, tx, ex);
    throw ex;
}

本文的样例代码是使用RestTemplate来调用其他服务的API，查看RestTemplate相关的代码发现，具体的错误处理实在org.springframework.web.client.DefaultResponseErrorHandler中完成的。

1
2
3
4
5
6
7
8
9
java复制代码public void debit(String userId, BigDecimal orderMoney) {
    String url = "http://127.0.0.1:8083?userId=" + userId + "&orderMoney=" + orderMoney;
    try {
        restTemplate.getForEntity(url, Void.class);
    } catch (Exception e) {
        log.error("debit url {} ,error:", url, e);
        throw new RuntimeException();
    }
}

在DefaultResponseErrorHandler的 hasError方法中判断返回的Response是否有错，如果有错就会调用handleError 方法。如果HTTP响应的状态是4xx或者5xx，就会被判定为有错，然后在handleError 方法中抛出异常。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
java复制代码public boolean isError() {
   return (is4xxClientError() || is5xxServerError());
}

protected void handleError(ClientHttpResponse response, HttpStatus statusCode) throws IOException {
   String statusText = response.getStatusText();
   HttpHeaders headers = response.getHeaders();
   byte[] body = getResponseBody(response);
   Charset charset = getCharset(response);
   String message = getErrorMessage(statusCode.value(), statusText, body, charset);

   switch (statusCode.series()) {
      case CLIENT_ERROR:
         throw HttpClientErrorException.create(message, statusCode, statusText, headers, body, charset);
      case SERVER_ERROR:
         throw HttpServerErrorException.create(message, statusCode, statusText, headers, body, charset);
      default:
         throw new UnknownHttpStatusCodeException(message, statusCode.value(), statusText, headers, body, charset);
   }
}

所以，在下游服务出现异常需要Rollback时，如果是使用RestTemplate来调用下游API，那么一定要保证，返回的HTTP状态是4xx或者5xx。如果使用其他方式调用API，也需要保证出错信息能反馈到TM端，并在TM端抛出异常。

参考

seata.io/zh-cn/docs/…

seata.io/zh-cn/blog/…

mp.weixin.qq.com/s/Pypkm5C9a…

chenjiayang.me/2019/06/29/…

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%