分布式事务 TCC-Transaction 源码分析 —— TCC 实现
本文主要基于 TCC-Transaction 1.2.3.3 正式版
1. 概述
2. TCC 原理
3. TCC-Transaction 原理
4. 事务与参与者
4.1 事务
4.2 参与者
5. 事务管理器
5.1 发起根事务
5.2 传播发起分支事务
5.3 传播获取分支事务
5.4 提交事务
5.5 回滚事务
5.6 添加参与者到事务
6. 事务拦截器
6.1 Compensable
6.2 可补偿事务拦截器
6.3 资源协调者拦截器
666. 彩蛋
友情提示：欢迎关注公众号【芋道源码】。😈关注后，拉你进【源码圈】微信群和【芋艿】搞基嗨皮。
友情提示：欢迎关注公众号【芋道源码】。😈关注后，拉你进【源码圈】微信群和【芋艿】】搞基嗨皮。
友情提示：欢迎关注公众号【芋道源码】。😈关注后，拉你进【源码圈】微信群和【芋艿】】搞基嗨皮。
1. 概述
本文分享 TCC 实现。主要涉及如下三个 Maven 项目：
tcc-transaction-core：tcc-transaction 底层实现。tcc-transaction-api：tcc-transaction 使用 API。tcc-transaction-spring：tcc-transaction Spring 支持。
你行好事会因为得到赞赏而愉悦
同理，开源项目贡献者会因为 Star 而更加有动力
为 TCC-Transaction 点赞！传送门
OK，开始我们的第一段 TCC 旅程吧。
ps：笔者假设你已经阅读过《tcc-transaction 官方文档 —— 使用指南1.2.x》。
ps2：未特殊说明的情况下，本文事务指的是 TCC事务。
2. TCC 原理
FROM https://support.hwclouds.com/devg-servicestage/zh-cn_topic_0056814426.html
TCC事务
为了解决在事务运行过程中大颗粒度资源锁定的问题，业界提出一种新的事务模型，它是基于业务层面的事务定义。锁粒度完全由业务自己控制。它本质是一种补偿的思路。它把事务运行过程分成 Try、Confirm / Cancel 两个阶段。在每个阶段的逻辑由业务代码控制。这样就事务的锁粒度可以完全自由控制。业务可以在牺牲隔离性的情况下，获取更高的性能。
Try 阶段
完成所有业务检查( 一致性 )
预留必须业务资源( 准隔离性 )
Try ：尝试执行业务
Confirm / Cancel 阶段：
释放 Try 阶段预留的业务资源
Cancel 操作满足幂等性
真正执行业务
不做任务业务检查
Confirm 操作满足幂等性
Confirm ：确认执行业务
Cancel ：取消执行业务
Confirm 与 Cancel 互斥
整体流程如下图：
红框部分功能由
tcc-transaction-core实现：启动业务活动
登记业务操作
提交 / 回滚业务活动
黄框部分功能由
tcc-transaction-http-sample实现( 官方提供的示例项目 )：Try 操作
Confirm 操作
Cancel 操作
与 2PC协议 比较：
位于业务服务层而非自愿层
没有单独的准备( Prepare )阶段，Try 操作兼备自愿操作与准备能力
Try 操作可以灵活选择业务资源的锁定粒度
较高开发成本
参考资料：
《支付宝运营架构中柔性事务指的是什么？》
《分布式事务的典型处理方式:2PC、TCC、异步确保和最大努力型》
3. TCC-Transaction 原理
在 TCC 里，一个业务活动可以有多个事务，每个业务操作归属于不同的事务，即一个事务可以包含多个业务操作。TCC-Transaction 将每个业务操作抽象成事务参与者，每个事务可以包含多个参与者。
参与者需要声明 try / confirm / cancel 三个类型的方法，和 TCC 的操作一一对应。在程序里，通过 @Compensable 注解标记在 try 方法上，并填写对应的 confirm / cancel 方法，示例代码如下：
// try
(confirmMethod = "confirmRecord", cancelMethod = "cancelRecord", transactionContextEditor = MethodTransactionContextEditor.class)
public String record(TransactionContext transactionContext, CapitalTradeOrderDto tradeOrderDto) {}
// confirm
public void confirmRecord(TransactionContext transactionContext, CapitalTradeOrderDto tradeOrderDto) {}
// cancel
public void cancelRecord(TransactionContext transactionContext, CapitalTradeOrderDto tradeOrderDto) {}
在示例代码中，我们看到 TransactionContext，事务上下文，这个是怎么生成的呢？这里先卖一个关子。
TCC-Transaction 有两个拦截器，通过对 @Compensable AOP 切面( 参与者 try 方法 )进行拦截，透明化对参与者 confirm / cancel 方法调用，从而实现 TCC 。简化流程如下图：
第一个拦截器，可补偿事务拦截器，实现如下功能：
在 Try 阶段，对事务的发起、传播。
在 Confirm / Cancel 阶段，对事务提交或回滚。
为什么会有对事务的传播呢？在远程调用服务的参与者时，会通过"参数"( 需要序列化 )的形式传递事务给远程参与者。
第二个拦截器，资源协调者拦截器，实现如下功能：
在 Try 阶段，添加参与者到事务中。当事务上下文不存在时，进行创建。
实际拦截器对事务的处理会比上图复杂一些，在本文「6. 事务拦截器」详细解析。
在 TCC-Transaction 代码实现上，组件分层如下图：
本文按照如下顺序分享：
「4. 事务拦截器」
「5. 事务管理器」
「6. 事务管理器」
内容是自下而上的方式分享，每个组件可以更加整体的被认识。当然这可能对你理解时产生一脸闷逼，所以推荐两种阅读方式：
简读 x 1 + 深读 x 1
倒着读，发现未分享的方法，全文检索该方法。
事务存储器在《TCC-Transaction 源码解析 —— 事务存储于恢复》详细解析。
事务恢复在《TCC-Transaction 源码解析 —— 事务恢复》详细解析。
4. 事务与参与者
在 TCC 里，一个事务( org.mengyun.tcctransaction.Transaction ) 可以有多个参与者( org.mengyun.tcctransaction.Participant )参与业务活动。类图关系如下( 打开大图 )：
4.1 事务
Transaction 实现代码如下：
public class Transaction implements Serializable {
private static final long serialVersionUID = 7291423944314337931L;
/**
* 事务编号
*/
private TransactionXid xid;
/**
* 事务状态
*/
private TransactionStatus status;
/**
* 事务类型
*/
private TransactionType transactionType;
/**
* 重试次数
*/
private volatile int retriedCount = 0;
/**
* 创建时间
*/
private Date createTime = new Date();
/**
* 最后更新时间
*/
private Date lastUpdateTime = new Date();
/**
* 版本号
*/
private long version = 1;
/**
* 参与者集合
*/
private List<Participant> participants = new ArrayList<Participant>();
/**
* 附带属性映射
*/
private Map<String, Object> attachments = new ConcurrentHashMap<String, Object>();
/**
* 添加参与者
*
* @param participant 参与者
*/
public void enlistParticipant(Participant participant) {
participants.add(participant);
}
/**
* 提交 TCC 事务
*/
public void commit() {
for (Participant participant : participants) {
participant.commit();
}
}
/**
* 回滚 TCC 事务
*/
public void rollback() {
for (Participant participant : participants) {
participant.rollback();
}
}
}
xid，事务编号( TransactionXid )，用于唯一标识一个事务。使用 UUID 算法生成，保证唯一性。
org.mengyun.tcctransaction.api.TransactionXid实现javax.transaction.xa.Xid接口，实现代码如下：public class TransactionXid implements Xid, Serializable {
private static final long serialVersionUID = -6817267250789142043L;/**}
* xid 格式标识符
*/
private int formatId = 1;
/**
* 全局事务编号
*/
private byte[] globalTransactionId;
/**
* 分支事务编号
*/
private byte[] branchQualifier;TODO 为什么要继承 Xid 接口？
status，事务状态( TransactionStatus )。
org.mengyun.tcctransaction.api.TransactionStatus实现代码如下：public enum TransactionStatus {
/**
* 尝试中状态
*/
TRYING(1),
/**
* 确认中状态
*/
CONFIRMING(2),
/**
* 取消中状态
*/
CANCELLING(3);private int id;}transactionType，事务类型( TransactionType )。
org.mengyun.tcctransaction.common.TransactionType实现代码如下：public enum TransactionType {/**}
* 根事务
*/
ROOT(1),
/**
* 分支事务
*/
BRANCH(2);
int id;在「6.2 可补偿事务拦截器」有详细解析，可以看到看到这两种事务是如何发起。
retriedCount，重试次数。在 TCC 过程中，可能参与者异常崩溃，这个时候会进行重试直到成功或超过最大次数。在《TCC-Transaction 源码解析 —— 事务恢复》详细解析。
version，版本号，用于乐观锁更新事务。在《TCC-Transaction 源码解析 —— 事务存储器》详细解析。
attachments，附带属性映射。在《TCC-Transaction 源码解析 —— Dubbo 支持》详细解析。
提供
#enlistParticipant()方法，添加事务参与者。提供
#commit()方法，调用参与者们提交事务。提供
#rollback()方法，调用参与者回滚事务。
4.2 参与者
Participant 实现代码如下：
public class Participant implements Serializable {
private static final long serialVersionUID = 4127729421281425247L;
/**
* 事务编号
*/
private TransactionXid xid;
/**
* 确认执行业务方法调用上下文
*/
private InvocationContext confirmInvocationContext;
/**
* 取消执行业务方法
*/
private InvocationContext cancelInvocationContext;
/**
* 执行器
*/
private Terminator terminator = new Terminator();
/**
* 事务上下文编辑
*/
Class<? extends TransactionContextEditor> transactionContextEditorClass;
/**
* 提交事务
*/
public void commit() {
terminator.invoke(new TransactionContext(xid, TransactionStatus.CONFIRMING.getId()), confirmInvocationContext, transactionContextEditorClass);
}
/**
* 回滚事务
*/
public void rollback() {
terminator.invoke(new TransactionContext(xid, TransactionStatus.CANCELLING.getId()), cancelInvocationContext, transactionContextEditorClass);
}
}
xid，参与者事务编号。通过
TransactionXid.globalTransactionId属性，关联上其所属的事务。当参与者进行远程调用时，远程的分支事务的事务编号等于该参与者的事务编号。通过事务编号的关联，TCC Confirm / Cancel 阶段，使用参与者的事务编号和远程的分支事务进行关联，从而实现事务的提交和回滚，在「5.2 传播发起分支事务」 + 「6.2 可补偿事务拦截器」可以看到具体实现。confirmInvocationContext，确认执行业务方法调用上下文( InvocationContext )。
org.mengyun.tcctransaction.InvocationContext实现代码如下：public class InvocationContext implements Serializable {private static final long serialVersionUID = -7969140711432461165L;}
/**
* 类
*/
private Class targetClass;
/**
* 方法名
*/
private String methodName;
/**
* 参数类型数组
*/
private Class[] parameterTypes;
/**
* 参数数组
*/
private Object[] args;InvocationContext，执行方法调用上下文，记录类、方法名、参数类型数组、参数数组。通过这些属性，可以执行提交 / 回滚事务。在
org.mengyun.tcctransaction.Terminator会看到具体的代码实现。本质上，TCC 通过多个参与者的 try / confirm / cancel 方法，实现事务的最终一致性。cancelInvocationContext，取消执行业务方法调用上下文( InvocationContext )。
terminator，执行器( Terminator )。
org.mengyun.tcctransaction.Terminator实现代码如下：public class Terminator implements Serializable {private static final long serialVersionUID = -164958655471605778L;}
public Object invoke(TransactionContext transactionContext, InvocationContext invocationContext, Class<? extends TransactionContextEditor> transactionContextEditorClass) {
if (StringUtils.isNotEmpty(invocationContext.getMethodName())) {
try {
// 获得 参与者对象
Object target = FactoryBuilder.factoryOf(invocationContext.getTargetClass()).getInstance();
// 获得 方法
Method method = target.getClass().getMethod(invocationContext.getMethodName(), invocationContext.getParameterTypes());
// 设置 事务上下文 到方法参数
FactoryBuilder.factoryOf(transactionContextEditorClass).getInstance().set(transactionContext, target, method, invocationContext.getArgs());
// 执行方法
return method.invoke(target, invocationContext.getArgs());
} catch (Exception e) {
throw new SystemException(e);
}
}
return null;
}FactoryBuilder，工厂 Builder，感兴趣的同学点击链接查看，已经添加完整中文代码注释。
TransactionContextEditor，在本文「6.1 Compensable」详细解析。
transactionContextEditorClass，事务上下文编辑，在「6.1 Compensable」详细解析。
提交
#commit()方法，提交参与者自己的事务。提交
#rollback()方法，回滚参与者自己的事务。
5. 事务管理器
org.mengyun.tcctransaction.TransactionManager，事务管理器，提供事务的获取、发起、提交、回滚，参与者的新增等等方法。
5.1 发起根事务
提供 begin() 方法，发起根事务。该方法在调用方法类型为 MethodType.ROOT 并且 事务处于 Try 阶段被调用。MethodType 在「6.2 可补偿事务拦截器」详细解析。
实现代码如下：
// TransactionManager.java
/**
* 发起根事务
*
* @return 事务
*/
public Transaction begin() {
// 创建 根事务
Transaction transaction = new Transaction(TransactionType.ROOT);
// 存储 事务
transactionRepository.create(transaction);
// 注册 事务
registerTransaction(transaction);
return transaction;
}
调用 Transaction 构造方法，创建根事务。实现代码如下：
// Transaction.java
/**
* 创建指定类型的事务
*
* @param transactionType 事务类型
*/
public Transaction(TransactionType transactionType) {
this.xid = new TransactionXid();
this.status = TransactionStatus.TRYING; // 尝试中状态
this.transactionType = transactionType;
}目前该构造方法只有
TransactionManager#begin()在调用，即只创建根事务。调用
TransactionRepository#crete()方法，存储事务。调用
#registerTransaction(...)方法，注册事务到当前线程事务队列。实现代码如下：// TransactionManager.java
/**
* 当前线程事务队列
*/
private static final ThreadLocal<Deque<Transaction>> CURRENT = new ThreadLocal<Deque<Transaction>>();
/**
* 注册事务到当前线程事务队列
*
* @param transaction 事务
*/
private void registerTransaction(Transaction transaction) {
if (CURRENT.get() == null) {
CURRENT.set(new LinkedList<Transaction>());
}
CURRENT.get().push(transaction); // 添加到头部
}可能有同学会比较好奇，为什么使用队列存储当前线程事务？TCC-Transaction 支持多个的事务独立存在，后创建的事务先提交，类似 Spring 的
org.springframework.transaction.annotation.Propagation.REQUIRES_NEW。在下文，很快我们就会看到 TCC-Transaction 自己的org.mengyun.tcctransaction.api.Propagation。
5.2 传播发起分支事务
调用 #propagationNewBegin(...) 方法，传播发起分支事务。该方法在调用方法类型为 MethodType.PROVIDER 并且 事务处于 Try 阶段被调用。MethodType 在「6.2 可补偿事务拦截器」详细解析。
实现代码如下：
/**
* 传播发起分支事务
*
* @param transactionContext 事务上下文
* @return 分支事务
*/
public Transaction propagationNewBegin(TransactionContext transactionContext) {
// 创建 分支事务
Transaction transaction = new Transaction(transactionContext);
// 存储 事务
transactionRepository.create(transaction);
// 注册 事务
registerTransaction(transaction);
return transaction;
}
调用 Transaction 构造方法，创建分支事务。实现代码如下：
/**
* 创建分支事务
*
* @param transactionContext 事务上下文
*/
public Transaction(TransactionContext transactionContext) {
this.xid = transactionContext.getXid(); // 事务上下文的 xid
this.status = TransactionStatus.TRYING; // 尝试中状态
this.transactionType = TransactionType.BRANCH; // 分支事务
}分支事务使用传播的事务上下文的事务编号。
调用
TransactionRepository#crete()方法，存储事务。为什么要存储分支事务，在「6.3 资源协调者拦截器」详细解析。调用
#registerTransaction(...)方法，注册事务到当前线程事务队列。
5.3 传播获取分支事务
调用 #propagationExistBegin(...) 方法，传播发起分支事务。该方法在调用方法类型为 MethodType.PROVIDER 并且 事务处于 Confirm / Cancel 阶段被调用。MethodType 在「6.2 可补偿事务拦截器」详细解析。
实现代码如下：
/**
* 传播获取分支事务
*
* @param transactionContext 事务上下文
* @return 分支事务
* @throws NoExistedTransactionException 当事务不存在时
*/
public Transaction propagationExistBegin(TransactionContext transactionContext) throws NoExistedTransactionException {
// 查询 事务
Transaction transaction = transactionRepository.findByXid(transactionContext.getXid());
if (transaction != null) {
// 设置 事务 状态
transaction.changeStatus(TransactionStatus.valueOf(transactionContext.getStatus()));
// 注册 事务
registerTransaction(transaction);
return transaction;
} else {
throw new NoExistedTransactionException();
}
}
调用
TransactionRepository#findByXid()方法，查询事务。调用
Transaction#changeStatus(...)方法，设置事务状态为 CONFIRMING 或 CANCELLING。调用
#registerTransaction(...)方法，注册事务到当前线程事务队列。为什么此处是分支事务呢？结合
#propagationNewBegin(...)思考下。
5.4 提交事务
调用 #commit(...) 方法，提交事务。该方法在事务处于 Confirm / Cancel 阶段被调用。
实现代码如下：
/**
* 提交事务
*/
public void commit() {
// 获取 事务
Transaction transaction = getCurrentTransaction();
// 设置 事务状态 为 CONFIRMING
transaction.changeStatus(TransactionStatus.CONFIRMING);
// 更新 事务
transactionRepository.update(transaction);
try {
// 提交 事务
transaction.commit();
// 删除 事务
transactionRepository.delete(transaction);
} catch (Throwable commitException) {
logger.error("compensable transaction confirm failed.", commitException);
throw new ConfirmingException(commitException);
}
}
调用
#getCurrentTransaction()方法， 获取事务。实现代码如下：public Transaction getCurrentTransaction() {
if (isTransactionActive()) {
return CURRENT.get().peek(); // 获得头部元素
}
return null;
}
public boolean isTransactionActive() {
Deque<Transaction> transactions = CURRENT.get();
return transactions != null && !transactions.isEmpty();
}为什么获得队列头部元素呢？该元素即是上文调用
#registerTransaction(...)注册到队列头部。调用
Transaction#changeStatus(...)方法， 设置事务状态为 CONFIRMING。调用
TransactionRepository#update(...)方法， 更新事务。调用
Transaction#commit(...)方法， 提交事务。调用
TransactionRepository#delete(...)方法，删除事务。
5.5 回滚事务
调用 #rollback(...) 方法，取消事务，和 #commit() 方法基本类似。该方法在事务处于 Confirm / Cancel 阶段被调用。
实现代码如下：
/**
* 回滚事务
*/
public void rollback() {
// 获取 事务
Transaction transaction = getCurrentTransaction();
// 设置 事务状态 为 CANCELLING
transaction.changeStatus(TransactionStatus.CANCELLING);
// 更新 事务
transactionRepository.update(transaction);
try {
// 提交 事务
transaction.rollback();
// 删除 事务
transactionRepository.delete(transaction);
} catch (Throwable rollbackException) {
logger.error("compensable transaction rollback failed.", rollbackException);
throw new CancellingException(rollbackException);
}
}
调用
#getCurrentTransaction()方法，获取事务。调用
Transaction#changeStatus(...)方法， 设置事务状态为 CANCELLING。调用
TransactionRepository#update(...)方法， 更新事务。调用
Transaction#rollback(...)方法， 回滚事务。调用
TransactionRepository#delete(...)方法，删除事务。
5.6 添加参与者到事务
调用 #enlistParticipant(...) 方法，添加参与者到事务。该方法在事务处于 Try 阶段被调用，在「6.3 资源协调者拦截器」有详细解析。
实现代码如下：
/**
* 添加参与者到事务
*
* @param participant 参与者
*/
public void enlistParticipant(Participant participant) {
// 获取 事务
Transaction transaction = this.getCurrentTransaction();
// 添加参与者
transaction.enlistParticipant(participant);
// 更新 事务
transactionRepository.update(transaction);
}
调用
#getCurrentTransaction()方法，获取事务。调用
Transaction#enlistParticipant(...)方法， 添加参与者到事务。调用
TransactionRepository#update(...)方法， 更新事务。
6. 事务拦截器
TCC-Transaction 基于 org.mengyun.tcctransaction.api.@Compensable + org.aspectj.lang.annotation.@Aspect 注解 AOP 切面实现业务方法的 TCC 事务声明拦截，同 Spring 的 org.springframework.transaction.annotation.@Transactional 的实现。
TCC-Transaction 有两个拦截器：
org.mengyun.tcctransaction.interceptor.CompensableTransactionInterceptor，可补偿事务拦截器。org.mengyun.tcctransaction.interceptor.ResourceCoordinatorInterceptor，资源协调者拦截器。
在分享拦截器的实现之前，我们先一起看看 @Compensable 注解。
6.1 Compensable
@Compensable，标记可补偿的方法注解。实现代码如下：
public Compensable {
/**
* 传播级别
*/
Propagation propagation() default Propagation.REQUIRED;
/**
* 确认执行业务方法
*/
String confirmMethod() default "";
/**
* 取消执行业务方法
*/
String cancelMethod() default "";
/**
* 事务上下文编辑
*/
Class<? extends TransactionContextEditor> transactionContextEditor() default DefaultTransactionContextEditor.class;
}
propagation，传播级别( Propagation )，默认 Propagation.REQUIRED。和 Spring 的 Propagation 除了缺少几个属性，基本一致。实现代码如下：
public enum Propagation {/**}
* 支持当前事务，如果当前没有事务，就新建一个事务。
*/
REQUIRED(0),
/**
* 支持当前事务，如果当前没有事务，就以非事务方式执行。
*/
SUPPORTS(1),
/**
* 支持当前事务，如果当前没有事务，就抛出异常。
*/
MANDATORY(2),
/**
* 新建事务，如果当前存在事务，把当前事务挂起。
*/
REQUIRES_NEW(3);
private final int value;confirmMethod，确认执行业务方法名。
cancelMethod，取消执行业务方法名。
TransactionContextEditor，事务上下文编辑器( TransactionContextEditor )，用于设置和获得事务上下文( TransactionContext )，在「6.3 资源协调者拦截器」可以看到被调用，此处只看它的代码实现。
org.mengyun.tcctransaction.api.TransactionContextEditor接口代码如下：public interface TransactionContextEditor {/**}
* 从参数中获得事务上下文
*
* @param target 对象
* @param method 方法
* @param args 参数
* @return 事务上下文
*/
TransactionContext get(Object target, Method method, Object[] args);
/**
* 设置事务上下文到参数中
*
* @param transactionContext 事务上下文
* @param target 对象
* @param method 方法
* @param args 参数
*/
void set(TransactionContext transactionContext, Object target, Method method, Object[] args);x
DefaultTransactionContextEditor，默认事务上下文编辑器实现。实现代码如下：
class DefaultTransactionContextEditor implements TransactionContextEditor {
public TransactionContext get(Object target, Method method, Object[] args) {
int position = getTransactionContextParamPosition(method.getParameterTypes());
if (position >= 0) {
return (TransactionContext) args[position];
}
return null;
}
public void set(TransactionContext transactionContext, Object target, Method method, Object[] args) {
int position = getTransactionContextParamPosition(method.getParameterTypes());
if (position >= 0) {
args[position] = transactionContext; // 设置方法参数
}
}
/**
* 获得事务上下文在方法参数里的位置
*
* @param parameterTypes 参数类型集合
* @return 位置
*/
public static int getTransactionContextParamPosition(Class<?>[] parameterTypes) {
int position = -1;
for (int i = 0; i < parameterTypes.length; i++) {
if (parameterTypes[i].equals(org.mengyun.tcctransaction.api.TransactionContext.class)) {
position = i;
break;
}
}
return position;
}NullableTransactionContextEditor，无事务上下文编辑器实现。实现代码如下：
```Java
class NullableTransactionContextEditor implements TransactionContextEditor {
public TransactionContext get(Object target, Method method, Object[] args)
{
return nul
l;
ide
public void set(TransactionContext transactionContext, Object target, Method method, Object[] arg
s
)
{
}
}```
DubboTransactionContextEditor，Dubbo 事务上下文编辑器实现，通过 Dubbo 隐式传参方式获得事务上下文，在《TCC-Transaction 源码解析 —— Dubbo 支持》详细解析。
6.2 可补偿事务拦截器
先一起来看下可补偿事务拦截器对应的切面 org.mengyun.tcctransaction.interceptor.CompensableTransactionAspect，实现代码如下：
public abstract class CompensableTransactionAspect {
private CompensableTransactionInterceptor compensableTransactionInterceptor;
public void setCompensableTransactionInterceptor(CompensableTransactionInterceptor compensableTransactionInterceptor) {
this.compensableTransactionInterceptor = compensableTransactionInterceptor;
}
("@annotation(org.mengyun.tcctransaction.api.Compensable)")
public void compensableService() {
}
("compensableService()")
public Object interceptCompensableMethod(ProceedingJoinPoint pjp) throws Throwable {
return compensableTransactionInterceptor.interceptCompensableMethod(pjp);
}
public abstract int getOrder();
}
通过
org.aspectj.lang.annotation.@Pointcut+org.aspectj.lang.annotation.@Around注解，配置对 @Compensable 注解的方法进行拦截，调用CompensableTransactionInterceptor#interceptCompensableMethod(...)方法进行处理。
CompensableTransactionInterceptor 实现代码如下：
public class CompensableTransactionInterceptor {
private TransactionManager transactionManager;
private Set<Class<? extends Exception>> delayCancelExceptions;
public Object interceptCompensableMethod(ProceedingJoinPoint pjp) throws Throwable {
// 获得带 @Compensable 注解的方法
Method method = CompensableMethodUtils.getCompensableMethod(pjp);
//
Compensable compensable = method.getAnnotation(Compensable.class);
Propagation propagation = compensable.propagation();
// 获得 事务上下文
TransactionContext transactionContext = FactoryBuilder.factoryOf(compensable.transactionContextEditor()).getInstance().get(pjp.getTarget(), method, pjp.getArgs());
// 当前线程是否在事务中
boolean isTransactionActive = transactionManager.isTransactionActive();
// 判断事务上下文是否合法
if (!TransactionUtils.isLegalTransactionContext(isTransactionActive, propagation, transactionContext)) {
throw new SystemException("no active compensable transaction while propagation is mandatory for method " + method.getName());
}
// 计算方法类型
MethodType methodType = CompensableMethodUtils.calculateMethodType(propagation, isTransactionActive, transactionContext);
// 处理
switch (methodType) {
case ROOT:
return rootMethodProceed(pjp);
case PROVIDER:
return providerMethodProceed(pjp, transactionContext);
default:
return pjp.proceed();
}
}
}
调用
CompensableMethodUtils#getCompensableMethod(...)方法，获得带 @Compensable 注解的方法。实现代码如下：// CompensableMethodUtils.java
/**
* 获得带 @Compensable 注解的方法
*
* @param pjp 切面点
* @return 方法
*/
public static Method getCompensableMethod(ProceedingJoinPoint pjp) {
Method method = ((MethodSignature) (pjp.getSignature())).getMethod(); // 代理方法对象
if (method.getAnnotation(Compensable.class) == null) {
try {
method = pjp.getTarget().getClass().getMethod(method.getName(), method.getParameterTypes()); // 实际方法对象
} catch (NoSuchMethodException e) {
return null;
}
}
return method;
}调用
TransactionContextEditor#get(...)方法，从参数中获得事务上下文。为什么从参数中可以获得事务上下文呢？在「6.3 资源协调者拦截器」揭晓答案。调用
TransactionManager#isTransactionActive()方法，当前线程是否在事务中。实现代码如下：// TransactionManager.java
private static final ThreadLocal<Deque<Transaction>> CURRENT = new ThreadLocal<Deque<Transaction>>();
public boolean isTransactionActive() {
Deque<Transaction> transactions = CURRENT.get();
return transactions != null && !transactions.isEmpty();
}调用
TransactionUtils#isLegalTransactionContext(...)方法，判断事务上下文是否合法。实现代码如下：// TransactionUtils.java
/**
* 判断事务上下文是否合法
* 在 Propagation.MANDATORY 必须有在事务内
*
* @param isTransactionActive 是否
* @param propagation 传播级别
* @param transactionContext 事务上下文
* @return 是否合法
*/
public static boolean isLegalTransactionContext(boolean isTransactionActive, Propagation propagation, TransactionContext transactionContext) {
if (propagation.equals(Propagation.MANDATORY) && !isTransactionActive && transactionContext == null) {
return false;
}
return true;
}当传播级别为 Propagation.MANDATORY 时，要求必须在事务中。
调用
CompensableMethodUtils#calculateMethodType(...)方法，计算方法类型。实现代码如下：/**
* 计算方法类型
*
* @param propagation 传播级别
* @param isTransactionActive 是否事务开启
* @param transactionContext 事务上下文
* @return 方法类型
*/
public static MethodType calculateMethodType(Propagation propagation, boolean isTransactionActive, TransactionContext transactionContext) {
if ((propagation.equals(Propagation.REQUIRED) && !isTransactionActive && transactionContext == null) // Propagation.REQUIRED：支持当前事务，当前没有事务，就新建一个事务。
|| propagation.equals(Propagation.REQUIRES_NEW)) { // Propagation.REQUIRES_NEW：新建事务，如果当前存在事务，把当前事务挂起。
return MethodType.ROOT;
} else if ((propagation.equals(Propagation.REQUIRED) // Propagation.REQUIRED：支持当前事务
|| propagation.equals(Propagation.MANDATORY)) // Propagation.MANDATORY：支持当前事务
&& !isTransactionActive && transactionContext != null) {
return MethodType.PROVIDER;
} else {
return MethodType.NORMAL;
}
}事务传播级别为 Propagation.REQUIRED，并且当前没有事务。
事务传播级别为 Propagation.REQUIRES_NEW，新建事务，如果当前存在事务，把当前事务挂起。此时，事务管理器的当前线程事务队列可能会存在多个事务。
计算方法类型( MethodType )的目的，可以根据不同方法类型，做不同的事务处理。
方法类型为 MethodType.ROOT 时，发起根事务，判断条件如下二选一：
方法类型为 MethodType.ROOT 时，发起分支事务，判断条件如下二选一：
* 事务传播级别为 Propagation.REQUIRED，并且当前不存在事务，并且方法参数传递了事务上下文。
* 事务传播级别为 Propagation.PROVIDER，并且当前不存在事务，并且方法参数传递了事务上下文。
* 当前不存在事务，方法参数传递了事务上下文是什么意思？当跨服务远程调用时，被调用服务本身( 服务提供者 )不在事务中，通过传递事务上下文参数，融入当前事务。方法类型为 MethodType.Normal 时，不进行事务处理。
MethodType.CONSUMER 项目已经不再使用，猜测已废弃。
当方法类型为 MethodType.ROOT 时，调用
#rootMethodProceed(...)方法，发起 TCC 整体流程。实现代码如下：private Object rootMethodProceed(ProceedingJoinPoint pjp) throws Throwable {
Object returnValue;
Transaction transaction = null;
try {
// 发起根事务
transaction = transactionManager.begin();
// 执行方法原逻辑
try {
returnValue = pjp.proceed();
} catch (Throwable tryingException) {
if (isDelayCancelException(tryingException)) { // 是否延迟回滚
} else {
logger.warn(String.format("compensable transaction trying failed. transaction content:%s", JSON.toJSONString(transaction)), tryingException);
// 回滚事务
transactionManager.rollback();
}
throw tryingException;
}
// 提交事务
transactionManager.commit();
} finally {
// 将事务从当前线程事务队列移除
transactionManager.cleanAfterCompletion(transaction);
}
return returnValue;
}x
调用
#transactionManager()方法，发起根事务，TCC Try 阶段开始。调用
ProceedingJoinPoint#proceed()方法，执行方法原逻辑( 即 Try 逻辑 )。当原逻辑执行异常时，TCC Try 阶段失败，调用
TransactionManager#rollback(...)方法，TCC Cancel 阶段，回滚事务。此处#isDelayCancelException(...)方法，判断异常是否为延迟取消回滚异常，部分异常不适合立即回滚事务，在《TCC-Transaction 源码分析 —— 事务恢复》详细解析。当原逻辑执行成功时，TCC Try 阶段成功，调用
TransactionManager#commit(...)方法，TCC Confirm 阶段，提交事务。调用
TransactionManager#cleanAfterCompletion(...)方法，将事务从当前线程事务队列移除，避免线程冲突。实现代码如下：// TransactionManager.java
public void cleanAfterCompletion(Transaction transaction) {
if (isTransactionActive() && transaction != null) {
Transaction currentTransaction = getCurrentTransaction();
if (currentTransaction == transaction) {
CURRENT.get().pop();
} else {
throw new SystemException("Illegal transaction when clean after completion");
}
}
}当方法类型为 Propagation.PROVIDER 时，服务提供者参与 TCC 整体流程。实现代码如下：
private Object providerMethodProceed(ProceedingJoinPoint pjp, TransactionContext transactionContext) throws Throwable {
Transaction transaction = null;
try {
switch (TransactionStatus.valueOf(transactionContext.getStatus())) {
case TRYING:
// 传播发起分支事务
transaction = transactionManager.propagationNewBegin(transactionContext);
return pjp.proceed();
case CONFIRMING:
try {
// 传播获取分支事务
transaction = transactionManager.propagationExistBegin(transactionContext);
// 提交事务
transactionManager.commit();
} catch (NoExistedTransactionException excepton) {
//the transaction has been commit,ignore it.
}
break;
case CANCELLING:
try {
// 传播获取分支事务
transaction = transactionManager.propagationExistBegin(transactionContext);
// 回滚事务
transactionManager.rollback();
} catch (NoExistedTransactionException exception) {
//the transaction has been rollback,ignore it.
}
break;
}
} finally {
// 将事务从当前线程事务队列移除
transactionManager.cleanAfterCompletion(transaction);
}
// 返回空值
Method method = ((MethodSignature) (pjp.getSignature())).getMethod();
return ReflectionUtils.getNullValue(method.getReturnType());
}为什么要传播发起分支事务？在根事务进行 Confirm / Cancel 时，调用根事务上的参与者们提交或回滚事务时，进行远程服务方法调用的参与者，可以通过自己的事务编号关联上传播的分支事务( 两者的事务编号相等 )，进行事务的提交或回滚。
当事务处于 TransactionStatus.TRYING 时，调用
TransactionManager#propagationExistBegin(...)方法，传播发起分支事务。发起分支事务完成后，调用ProceedingJoinPoint#proceed()方法，执行方法原逻辑( 即 Try 逻辑 )。当事务处于 TransactionStatus.CONFIRMING 时，调用
TransactionManager#commit()方法，提交事务。当事务处于 TransactionStatus.CANCELLING 时，调用
TransactionManager#rollback()方法，提交事务。调用
TransactionManager#cleanAfterCompletion(...)方法，将事务从当前线程事务队列移除，避免线程冲突。当事务处于 TransactionStatus.CONFIRMING / TransactionStatus.CANCELLING 时，调用
ReflectionUtils#getNullValue(...)方法，返回空值。为什么返回空值？Confirm / Cancel 相关方法，是通过 AOP 切面调用，只调用，不处理返回值，但是又不能没有返回值，因此直接返回空。实现代码如下：public static Object getNullValue(Class type) {
// 处理基本类型
if (boolean.class.equals(type)) {
return false;
} else if (byte.class.equals(type)) {
return 0;
} else if (short.class.equals(type)) {
return 0;
} else if (int.class.equals(type)) {
return 0;
} else if (long.class.equals(type)) {
return 0;
} else if (float.class.equals(type)) {
return 0;
} else if (double.class.equals(type)) {
return 0;
}
// 处理对象
return null;
}当方法类型为 Propagation.NORMAL 时，执行方法原逻辑，不进行事务处理。
6.3 资源协调者拦截器
先一起来看下资源协调者拦截器 对应的切面 org.mengyun.tcctransaction.interceptor.CompensableTransactionAspect，实现代码如下：
public abstract class ResourceCoordinatorAspect {
private ResourceCoordinatorInterceptor resourceCoordinatorInterceptor;
("@annotation(org.mengyun.tcctransaction.api.Compensable)")
public void transactionContextCall() {
}
("transactionContextCall()")
public Object interceptTransactionContextMethod(ProceedingJoinPoint pjp) throws Throwable {
return resourceCoordinatorInterceptor.interceptTransactionContextMethod(pjp);
}
public void setResourceCoordinatorInterceptor(ResourceCoordinatorInterceptor resourceCoordinatorInterceptor) {
this.resourceCoordinatorInterceptor = resourceCoordinatorInterceptor;
}
public abstract int getOrder();
}
通过
org.aspectj.lang.annotation.@Pointcut+org.aspectj.lang.annotation.@Around注解，配置对 @Compensable 注解的方法进行拦截，调用ResourceCoordinatorInterceptor#interceptTransactionContextMethod(...)方法进行处理。
ResourceCoordinatorInterceptor 实现代码如下：
public class ResourceCoordinatorInterceptor {
private TransactionManager transactionManager;
public Object interceptTransactionContextMethod(ProceedingJoinPoint pjp) throws Throwable {
Transaction transaction = transactionManager.getCurrentTransaction();
if (transaction != null) {
switch (transaction.getStatus()) {
case TRYING:
// 添加事务参与者
enlistParticipant(pjp);
break;
case CONFIRMING:
break;
case CANCELLING:
break;
}
}
// 执行方法原逻辑
return pjp.proceed(pjp.getArgs());
}
}
当事务处于 TransactionStatus.TRYING 时，调用
#enlistParticipant(...)方法，添加事务参与者。调用
ProceedingJoinPoint#proceed(...)方法，执行方法原逻辑。
ResourceCoordinatorInterceptor#enlistParticipant() 实现代码如下：
private void enlistParticipant(ProceedingJoinPoint pjp) throws IllegalAccessException, InstantiationException {
// 获得 @Compensable 注解
Method method = CompensableMethodUtils.getCompensableMethod(pjp);
if (method == null) {
throw new RuntimeException(String.format("join point not found method, point is : %s", pjp.getSignature().getName()));
}
Compensable compensable = method.getAnnotation(Compensable.class);
// 获得 确认执行业务方法 和 取消执行业务方法
String confirmMethodName = compensable.confirmMethod();
String cancelMethodName = compensable.cancelMethod();
// 获取 当前线程事务第一个(头部)元素
Transaction transaction = transactionManager.getCurrentTransaction();
// 创建 事务编号
TransactionXid xid = new TransactionXid(transaction.getXid().getGlobalTransactionId());
// TODO
if (FactoryBuilder.factoryOf(compensable.transactionContextEditor()).getInstance().get(pjp.getTarget(), method, pjp.getArgs()) == null) {
FactoryBuilder.factoryOf(compensable.transactionContextEditor()).getInstance().set(new TransactionContext(xid, TransactionStatus.TRYING.getId()), pjp.getTarget(), ((MethodSignature) pjp.getSignature()).getMethod(), pjp.getArgs());
}
// 获得类
Class targetClass = ReflectionUtils.getDeclaringType(pjp.getTarget().getClass(), method.getName(), method.getParameterTypes());
// 创建 确认执行方法调用上下文 和 取消执行方法调用上下文
InvocationContext confirmInvocation = new InvocationContext(targetClass,
confirmMethodName,
method.getParameterTypes(), pjp.getArgs());
InvocationContext cancelInvocation = new InvocationContext(targetClass,
cancelMethodName,
method.getParameterTypes(), pjp.getArgs());
// 创建 事务参与者
Participant participant =
new Participant(
xid,
confirmInvocation,
cancelInvocation,
compensable.transactionContextEditor());
// 添加 事务参与者 到 事务
transactionManager.enlistParticipant(participant);
}
调用
CompensableMethodUtils#getCompensableMethod(...)方法，获得带 @Compensable 注解的方法。调用
#getCurrentTransaction()方法， 获取事务。调用 TransactionXid 构造方法，创建分支事务编号。实现代码如下：
/**
* 全局事务编号
*/
private byte[] globalTransactionId;
/**
* 分支事务编号
*/
private byte[] branchQualifier;
public TransactionXid(byte[] globalTransactionId) {
this.globalTransactionId = globalTransactionId;
branchQualifier = uuidToByteArray(UUID.randomUUID()); // 生成 分支事务编号
}分支事务编号(
branchQualifier) 需要生成。TODO TransactionContext 和 Participant 的关系。
调用
ReflectionUtils#getDeclaringType(...)方法，获得声明 @Compensable 方法的实际类。实现代码如下：public static Class getDeclaringType(Class aClass, String methodName, Class<?>[] parameterTypes) {
Method method;
Class findClass = aClass;
do {
Class[] clazzes = findClass.getInterfaces();
for (Class clazz : clazzes) {
try {
method = clazz.getDeclaredMethod(methodName, parameterTypes);
} catch (NoSuchMethodException e) {
method = null;
}
if (method != null) {
return clazz;
}
}
findClass = findClass.getSuperclass();
} while (!findClass.equals(Object.class));
return aClass;
}调用 InvocationContext 构造方法，分别创建确认执行方法调用上下文和取消执行方法调用上下文。实现代码如下：
/**
* 类
*/
private Class targetClass;
/**
* 方法名
*/
private String methodName;
/**
* 参数类型数组
*/
private Class[] parameterTypes;
/**
* 参数数组
*/
private Object[] args;
public InvocationContext(Class targetClass, String methodName, Class[] parameterTypes, Object... args) {
this.methodName = methodName;
this.parameterTypes = parameterTypes;
this.targetClass = targetClass;
this.args = args;
}调用 Participant 构造方法，创建事务参与者。实现代码如下：
public class Participant implements Serializable {private static final long serialVersionUID = 4127729421281425247L;}
/**
* 事务编号
*/
private TransactionXid xid;
/**
* 确认执行业务方法调用上下文
*/
private InvocationContext confirmInvocationContext;
/**
* 取消执行业务方法
*/
private InvocationContext cancelInvocationContext;
/**
* 执行器
*/
private Terminator terminator = new Terminator();
/**
* 事务上下文编辑
*/
Class<? extends TransactionContextEditor> transactionContextEditorClass;
public Participant() {
}
public Participant(TransactionXid xid, InvocationContext confirmInvocationContext, InvocationContext cancelInvocationContext, Class<? extends TransactionContextEditor> transactionContextEditorClass) {
this.xid = xid;
this.confirmInvocationContext = confirmInvocationContext;
this.cancelInvocationContext = cancelInvocationContext;
this.transactionContextEditorClass = transactionContextEditorClass;
}调用
TransactionManager#enlistParticipant(...)方法，添加事务参与者到事务。
666. 彩蛋
受限于本人的能力，蛮多处表达不够清晰或者易懂，非常抱歉。如果你对任何地方有任何疑问，欢迎添加本人微信号( wangwenbin-server )，期待与你的交流。不限于 TCC，也可以是分布式事务，也可以是微服务，以及等等。
外送一本武林秘籍：带中文注释的 TCC-Transaction 仓库地址，目前正在慢慢完善。传送门：https://github.com/YunaiV/tcc-transaction。
再送一本葵花宝典：《TCC型分布式事务原理和实现》系列。
胖友，分享一个朋友圈可好？