tcc分布式事务源码解析系列(五)之项目实战

接上一篇，我们已经分析了在整个消费的调用流程，现在只差发起真实的rpc远端调用了，这篇文章，我们一起进入提供者的调用流程吧！

我们发起 accountService.payment(accountDTO); 的调用，在提供方，我们可以看到其实现类为AccountServiceImpl:

/** * 扣款支付 * * @param accountDTO 参数dto * @return true */ @Override @Tcc(confirmMethod = "confirm", cancelMethod = "cancel") public boolean payment(AccountDTO accountDTO) { final AccountDO accountDO = accountMapper.findByUserId(accountDTO.getUserId()); accountDO.setBalance(accountDO.getBalance().subtract(accountDTO.getAmount())); accountDO.setFreezeAmount(accountDO.getFreezeAmount().add(accountDTO.getAmount())); accountDO.setUpdateTime(new Date()); final int update = accountMapper.update(accountDO); if (update != 1) { throw new TccRuntimeException("资金不足！"); } return Boolean.TRUE; }

public boolean confirm(AccountDTO accountDTO) {

1   LOGGER.debug("============执行确认付款接口===============");
2
3   final AccountDO accountDO = accountMapper.findByUserId(accountDTO.getUserId());
4   accountDO.setFreezeAmount(accountDO.getFreezeAmount().subtract(accountDTO.getAmount()));
5   accountDO.setUpdateTime(new Date());
6   accountMapper.update(accountDO);
7   return Boolean.TRUE;

}

public boolean cancel(AccountDTO accountDTO) {

1   LOGGER.debug("============执行取消付款接口===============");
2   final AccountDO accountDO = accountMapper.findByUserId(accountDTO.getUserId());
3   accountDO.setBalance(accountDO.getBalance().add(accountDTO.getAmount()));
4   accountDO.setFreezeAmount(accountDO.getFreezeAmount().subtract(accountDTO.getAmount()));
5   accountDO.setUpdateTime(new Date());
6   accountMapper.update(accountDO);
7   return Boolean.TRUE;

}

我们发现它也有@Tcc注解，并且提供了confrim，cancel等真实的方法。通过前面一篇的分析，我们知道，他是springBean的一个实现类，同样会走切面。

经过 TccTransactionFactoryServiceImpl 的 factoryOf方法，我们可以知道他会返回 ProviderTccTransactionHandler

@Override public Class factoryOf(TccTransactionContext context) throws Throwable {

1   //如果事务还没开启或者 tcc事务上下文是空， 那么应该进入发起调用
2   if (!tccTransactionManager.isBegin() && Objects.isNull(context)) {
3       return StartTccTransactionHandler.class;
4   } else if (tccTransactionManager.isBegin() && Objects.isNull(context)) {
5       return ConsumeTccTransactionIHandler.class;
6   } else if (Objects.nonNull(context)) {
7       return ProviderTccTransactionHandler.class;
8   }
9   return ConsumeTccTransactionIHandler.class;

}

最终我们来到 ProviderTccTransactionHandler.handler 方法：

/** * 分布式事务提供者处理接口 * 根据tcc事务上下文的状态来执行相对应的方法 * * @param point point 切点 * @param context context * @return Object * @throws Throwable 异常 */ @Override public Object handler(ProceedingJoinPoint point, TccTransactionContext context) throws Throwable { TccTransaction tccTransaction = null; try { switch (TccActionEnum.acquireByCode(context.getAction())) { case TRYING: try { //创建事务信息 tccTransaction = tccTransactionManager.providerBegin(context); //发起方法调用 return point.proceed(); } catch (Throwable throwable) { tccTransactionManager.removeTccTransaction(tccTransaction); throw throwable;

1               }
2           case CONFIRMING:
3               //如果是confirm 通过之前保存的事务信息 进行反射调用
4               final TccTransaction acquire = tccTransactionManager.acquire(context);
5               tccTransactionManager.confirm();
6               break;
7           case CANCELING:
8               //如果是调用CANCELING 通过之前保存的事务信息 进行反射调用
9               tccTransactionManager.acquire(context);
10               tccTransactionManager.cancel();
11               break;
12           default:
13               break;
14       }
15   } finally {
16       tccTransactionManager.remove();
17   }
18   Method method = ((MethodSignature) (point.getSignature())).getMethod();
19   return getDefaultValue(method.getReturnType());

}

TccTransactionContext 就是通过rpc json序列化后传过来的对象，此时我们知道是在try阶段，所以我们进入try

try { //创建事务信息 tccTransaction = tccTransactionManager.providerBegin(context); //发起方法调用 return point.proceed(); } catch (Throwable throwable) { tccTransactionManager.removeTccTransaction(tccTransaction); throw throwable;

}

首先我们会创建提供者的事务信息，并把他存起来，再把它存入threadlocal中，接着发起 point.proceed() 调用的时候，我们会进入 TccCoordinatorMethodAspect,由于是在try阶段最终会进入：

/** * 获取调用接口的协调方法并封装 * * @param point 切点 */ private void registerParticipant(ProceedingJoinPoint point, String transId) throws NoSuchMethodException {

1   MethodSignature signature = (MethodSignature) point.getSignature();
2   Method method = signature.getMethod();
3
4   Class<?> clazz = point.getTarget().getClass();
5
6   Object[] args = point.getArgs();
7
8   final Tcc tcc = method.getAnnotation(Tcc.class);
9
10   //获取协调方法
11   String confirmMethodName = tcc.confirmMethod();
12
13  /* if (StringUtils.isBlank(confirmMethodName)) {
14       confirmMethodName = method.getName();
15   }*/
16
17   String cancelMethodName = tcc.cancelMethod();
18
19  /* if (StringUtils.isBlank(cancelMethodName)) {
20       cancelMethodName = method.getName();
21   }

*/ //设置模式 final TccPatternEnum pattern = tcc.pattern();

1   tccTransactionManager.getCurrentTransaction().setPattern(pattern.getCode());
2
3
4   TccInvocation confirmInvocation = null;
5   if (StringUtils.isNoneBlank(confirmMethodName)) {
6       confirmInvocation = new TccInvocation(clazz,
7               confirmMethodName, method.getParameterTypes(), args);
8   }
9
10   TccInvocation cancelInvocation = null;
11   if (StringUtils.isNoneBlank(cancelMethodName)) {
12       cancelInvocation = new TccInvocation(clazz,
13               cancelMethodName,
14               method.getParameterTypes(), args);
15   }
16
17
18   //封装调用点
19   final Participant participant = new Participant(
20           transId,
21           confirmInvocation,
22           cancelInvocation);
23
24   tccTransactionManager.enlistParticipant(participant);

}

这里获取真实的confrim，cancel方法并存入当前的事务信息中。然后发起真实的业务调用，即执行payment方法：

@Override @Tcc(confirmMethod = "confirm", cancelMethod = "cancel") public boolean payment(AccountDTO accountDTO) { final AccountDO accountDO = accountMapper.findByUserId(accountDTO.getUserId()); accountDO.setBalance(accountDO.getBalance().subtract(accountDTO.getAmount())); accountDO.setFreezeAmount(accountDO.getFreezeAmount().add(accountDTO.getAmount())); accountDO.setUpdateTime(new Date()); final int update = accountMapper.update(accountDO); if (update != 1) { throw new TccRuntimeException("资金不足！"); } return Boolean.TRUE; }
当我们执行完该方法后，会返回，还记得我是在哪里来执行这个方法的吗？对，当然是切面，我们是在切面里执行的，我们是在 PaymentServiceImpl.makePayment 切面里面执行的！请要理解这一点，执行完后，我们发起了 inventoryService.decrease(inventoryDTO) 调用他的调用原理和上面一模一样，只是在不同的模块里面执行。当 makePayment 方法执行完后，我们该怎么执行？你还记得 StartTccTransactionHandler吗，它可一直在那等呢。。我们再来回顾下他的代码:

@Override public Object handler(ProceedingJoinPoint point, TccTransactionContext context) throws Throwable { Object returnValue; try { tccTransactionManager.begin(); try { //发起调用执行try方法 returnValue = point.proceed();
```
1       } catch (Throwable throwable) {
2           //异常执行cancel
3
4           tccTransactionManager.cancel();
5
6           throw throwable;
7       }
8       //try成功执行confirm confirm 失败的话，那就只能走本地补偿
9       tccTransactionManager.confirm();
10   } finally {
11       tccTransactionManager.remove();
12   }
13   return returnValue;
```
}
说到底，我们走了这么久，其实到这里，我们才执行完 returnValue = point.proceed(); 这一句代码。

没有异常

我们会执行 tccTransactionManager.confirm(); 我们跟进去看代码:

/** * 调用confirm方法这里主要如果是发起者调用这里调用远端的还是原来的方法，不过上下文设置了调用confirm * 那么远端的服务则会调用confirm方法。。 */ void confirm() throws TccRuntimeException {

1    LogUtil.debug(LOGGER, () -> "开始执行tcc confirm 方法！start");
2
3    final TccTransaction currentTransaction = getCurrentTransaction();
4
5    if (Objects.isNull(currentTransaction)) {
6        return;
7    }
8
9    currentTransaction.setStatus(TccActionEnum.CONFIRMING.getCode());
10
11    coordinatorCommand.execute(new CoordinatorAction(CoordinatorActionEnum.UPDATE, currentTransaction));
12
13    final List<Participant> participants = currentTransaction.getParticipants();
14    List<Participant> participantList = Lists.newArrayListWithCapacity(participants.size());
15    boolean success = true;
16    Participant fail = null;
17    if (CollectionUtils.isNotEmpty(participants)) {
18        for (Participant participant : participants) {
19            try {
20                TccTransactionContext context = new TccTransactionContext();
21                context.setAction(TccActionEnum.CONFIRMING.getCode());
22                context.setTransId(participant.getTransId());
23                TransactionContextLocal.getInstance().set(context);
24                //通过反射调用rpc的confrim方法
25                executeParticipantMethod(participant.getConfirmTccInvocation());
26                participantList.add(participant);
27            } catch (Exception e) {
28                LogUtil.error(LOGGER, "执行confirm方法异常:{}", () -> e);
29                success = false;
30                fail = participant;
31                break;
32            }
33        }
34    }
35    executeHandler(success, currentTransaction, fail, participantList, participants);
36}
37private void executeHandler(boolean success, final TccTransaction currentTransaction, Participant fail,
38                            List<Participant> participantList, final List<Participant> participants) {
39    if (success) {
40        TransactionContextLocal.getInstance().remove();
41        coordinatorCommand.execute(new CoordinatorAction(CoordinatorActionEnum.DELETE, currentTransaction));
42    } else {
43        //获取还没执行的，或者执行失败的
44        final List<Participant> updateList =
45                participants.stream().skip(participantList.size()).collect(Collectors.toList());
46        currentTransaction.setParticipants(updateList);
47        coordinatorCommand.execute(new CoordinatorAction(CoordinatorActionEnum.UPDATE, currentTransaction));
48        assert fail != null;
49        throw new TccRuntimeException(fail.getConfirmTccInvocation().toString());
50    }
51}
52
53private void executeParticipantMethod(TccInvocation tccInvocation) throws Exception {
54   if (Objects.nonNull(tccInvocation)) {
55       final Class clazz = tccInvocation.getTargetClass();
56       final String method = tccInvocation.getMethodName();
57       final Object[] args = tccInvocation.getArgs();
58       final Class[] parameterTypes = tccInvocation.getParameterTypes();
59       final Object bean = SpringBeanUtils.getInstance().getBean(clazz);
60       MethodUtils.invokeMethod(bean, method, args, parameterTypes);
61
62   }

}

这段代码的逻辑，简单理解起来，首先更新当前事务状态（confrim），获取当前事务的调用点的confrim方法，设置上下文，发起反射调用！

其实这里通过调试我们发现，发起confrim的方法为 AccountService.payment(AccountDTO accountDTO) ，不过设置的上下文状态为confrim, 当我们发起反射调用的时候，我们会走到 ProviderTccTransactionHandler.handler 方法,这个方法或许你还有印象，我们再看一下它的代码:

@Override public Object handler(ProceedingJoinPoint point, TccTransactionContext context) throws Throwable { TccTransaction tccTransaction = null; try { switch (TccActionEnum.acquireByCode(context.getAction())) { case TRYING: try { //创建事务信息 tccTransaction = tccTransactionManager.providerBegin(context); //发起方法调用 return point.proceed(); } catch (Throwable throwable) { tccTransactionManager.removeTccTransaction(tccTransaction); throw throwable;

1                }
2            case CONFIRMING:
3                //如果是confirm 通过之前保存的事务信息 进行反射调用
4                final TccTransaction acquire = tccTransactionManager.acquire(context);
5                tccTransactionManager.confirm();
6                break;
7            case CANCELING:
8                //如果是调用CANCELING 通过之前保存的事务信息 进行反射调用
9                tccTransactionManager.acquire(context);
10                tccTransactionManager.cancel();
11                break;
12            default:
13                break;
14        }
15    } finally {
16        tccTransactionManager.remove();
17    }
18    Method method = ((MethodSignature) (point.getSignature())).getMethod();
19    return getDefaultValue(method.getReturnType());
20}

这里因为上下文设置的状态为:CONFIRMING ,所以会执行:

//如果是confirm 通过之前保存的事务信息进行反射调用 final TccTransaction acquire = tccTransactionManager.acquire(context); tccTransactionManager.confirm(); break;
我们跟踪 tccTransactionManager.confirm(); 会发现和之前是一个方法，这时候，你要知道，这个方法是在account微服务里面执行
所以它最后会执行 AccountServiceImpl.confirm 方法，进行了付款确认。

tcc分布式事务源码解析系列(五)之项目实战

接上一篇，我们已经分析了在整个消费的调用流程，现在只差发起真实的rpc远端调用了，这篇文章，我们一起进入提供者的调用流程吧！

没有异常

同理cancel方法也和上面描述的一样的原理执行。

到这里，我们就解析完了，整个tcc过程执行的流程，大家关键要理解AOP，理解好了切面思想，其实是很简单的事情了，如果有任何疑问和问题，欢迎加入QQ群：162614487 进行讨论

评论区

相关推荐