Seata 源码分析 - tmrm 中 xid 传递过程

Posted 2022-03-14 小毕超

一、Seata

前面文章讲解了对 Seata 的 AT 和 TCC 模式的使用，本篇文章为大家讲解下 Seata 中 TM、RM 中 xid 传递过程，如果不了解 Seata 中的 xid，可以理解为全局的事物 ID，我们都知道 Seata 中分了三个角色，TC、TM、RM，其中 TC 为全局事物的协调者，TM 则为全局事物的发起者，RM 为全局事物的参与者，其中 TM 和 RM 我们可以看作一个组，有发起者必定有参与者，不然也没必要使用分布式事物了。那 TC 怎么知道谁和谁是一个组的呢，那就是本篇文章说的 xid，TC 主要通过 xid 来对当前的不同的连接来分组处理。

那 xid 怎么来的怎么传递的呢，下面我梳理了下过程：

1. 首先 TM 注册到 TC 中， 要发起全局事物时，先向 TC 发送一个通知，然后TC 就会生成一个唯一的 ID 返还给 TM，这个ID 就是 xid。
2. TM 收到 xid 后放入当前线程的 ThreadLocal 中存放，在业务逻辑中我们使用 OpenFeign 调用其他服务的接口时，Seata 重写了 feign客户端，如果是RestTemplate的方式，Seata也写了请求拦截器，将当前 ThreadLocal 中的 xid 放入 header 中进行传递。
3. RM 收到请求后，首先在拦截器中尝试获取 header 中的 xid ，如果获取成功就将 xid 再放入当前 ThreadLocal 中。
4. 然后 RM 通知 TC 自己是该 xid 的事物参与者，也就是注册该分支事务。
5. 后面不管是提交事物和回滚事物，TC 通过该 xid， 都能准确的通知相应的服务了。

从上面简单的梳理的过程来看 xid 在Seata 分布式事物中，是尤为重要的，起到了桥梁的作用，下面我们一起看下在 Seata 源码中的实现。

在开始前首先将 Seata 的源码拉取下来，或者在 idea中查看 seata-all 包中的代码也可以，下面是 Github 中 Seata 源码的地址：

https://github.com/seata/seata

二、源码分析

在分析前我们需要分析下，在 TM 中我们一般是如何开启一个 全局事物的呢，其中最为明显和常用的应该就是使用 GlobalTransactional 注解了，如果是 AT 模式，在配置好 Seata 环境后，然后在需要全局事物的方法上加上 GlobalTransactional 注解，下面基本就已经可以出现分布式事物的效果了。

既然是使用了 GlobalTransactional 注解，那我们就从该注解下手分析，在 java 中使注解生效的方式有哪些呢，一般都要通过扫描包的方式，然后再判断类上面是否有注解，然后再获取注解中的参数等。那在 Spring 中呢，我们要在方法前后加上某些同一的逻辑，应该一下就想到的就是 AOP 了吧，那 Spring 中我们要实现 AOP 现在常用的方式有哪些呢，一个就是使用注解 @Aspect 的方式，还有一种就是实现 MethodInterceptor 接口，都可以达到 AOP 的效果。

上面简单分析了，我们就要留意哪个类中是否加了 @Aspect 注解，并切入点是GlobalTransactional 注解，或者实现了 MethodInterceptor 接口，并且在invoke 中还获取了GlobalTransactional 注解。

在 idea 中全局搜索 GlobalTransactional ：其中有一个类的命名就非常抢眼GlobalTransactionalInterceptor，并且该类就实现了 MethodInterceptor 接口：

下面进入到该类中，既然是实现了MethodInterceptor 接口，那么我们就要看下该类下的 invoke 方法中干了什么事情：

在这里确实获取了方法上是否有GlobalTransactional 注解，如果有的话，就会创建一个新的 AspectTransactional 类，并将注解中的参数赋进去，然后执行handleGlobalTransaction方法：

下面点到 handleGlobalTransaction方法中，在该方法中主要执行了transactionalTemplate.execute，也就是执行全局事物，传入了一个TransactionalExecutor事物执行器：
在该执行器中，核心方法是 getTransactionInfo，主要包装了当前事物的信息 ，包括超时时间、名称、传播行为等都放入了TransactionInfo对象中：

这个方法何时执行的呢，下面还是要看 transactionalTemplate.execute 方法，点击来之后，第一步做的就是调用上面的 getTransactionInfo 方法，获取当前事物的信息：

上面还标出来了一个方法 GlobalTransactionContext.getCurrent()，在该方法上官方也给了注解，获取当前事物，如果不是空，这个角色就是GlobalTransactionRole.Participant，就是事物的参与者，点进入看下源码：

首先从当前的 ThreadLocal 中获取 xid，获取不到就返回 null，如果有的话就新建一个DefaultGlobalTransaction 对象，并将角色设为 GlobalTransactionRole.Participant 表示事物的参与者。

下面再回到 transactionalTemplate.execute 方法，接着向下看，下面的事物的传播行为这里就先略过，下面可以看到一个判断：

上面说到 GlobalTransactionContext.getCurrent()方法时， xid 获取不到的话，就返回null，那 tx 也就是null，此时会执行 GlobalTransactionContext.createNew()方法，点进入看下：

直接创建了一个默认的DefaultGlobalTransaction且使用的无参构造函数，再看下无参构造函数呢：

创建了一个角色为GlobalTransactionRole.Launcher 事物的发起者的对象，从这里就已经区分出了当前时 TM 还是 RM 角色。

下面再回到transactionalTemplate.execute方法中，接着下面有一个特别明显的方法 beginTransaction，并且该方法执行后就执行了 business.execute() 方法，而 business.execute() 方法就是执行的业务事物方法：

在该方法执行后，如果没有错误，有执行了commitTransaction方法，很明显beginTransaction方法开启全局事物，commitTransaction提交全局事物，如果异常使用completeTransactionAfterThrowing方法进行回滚。

但是到现在我们还没有看到获取 xid 的地方，那肯定再beginTransaction方法中了，进入该方法看下：

其中前后的触发通知先不看，直接再到tx.begin方法中，上面创建的就是DefaultGlobalTransaction 这里直接到 DefaultGlobalTransaction 的begin中查看 ：


在这里就看到了，我们想要看到的东西 xid ，是通过transactionManager.begin获取到的，并下面将 xid 存入了当前的 ThreadLocal 中，下面点到transactionManager.begin中查看是如何获取的xid，进入到DefaultTransactionManager 下的 begin 方法中：


这个地方就很明了了，发送了个请求获取到的 xid，也就是发送请求给 TC ，TC生成全局唯一的 xid 后返回给这里，这里发送请求就是使用的Netty了，可以到 syncCall 方法中看下：

到这里大家应该明白了 xid 的生成，那是怎么发送给 RM 的呢？上面再梳理过程的时候提到 Seata 重写了 OpenFeign 客户端，将xid 放入了 header中进行传播，这个重写的客户端就是 SeataFeignClient，这个在 spring-starter-alibaba-seata包中：

其中主要的方法为 getModifyRequest ，将 ThreadLocal 中的 xid，放入了请求header 中：

到这基本上 TM 端的 xid 的处理过程大致就明白了，下面看下 RM 端的，上面提到RM端使用拦截器的方式，拦截请求，查看当前header 中是否有 xid ，如果有就存入当前的 ThreadLocal 中，这个拦截器同样也在 spring-starter-alibaba-seata包中：

在这个类中，在请求前拦截preHandle中，进行了xid 的绑定，也就是从header中获取到，放入ThreadLocal 中，请求后拦截afterCompletion，又进行了 xid 的解绑：


这样 RPC 进来的请求，如果是全局事物的请求，就会将 全局事物ID xid 存入当前 ThreadLocal 中，这样就和最初的分析所呼应了。

喜欢的小伙伴可以关注我的个人微信公众号，获取更多学习资料！