Fescar源码学习--事物管理者TM（服务调用方）

r囧r小猫 2022-04-23 05:38 124阅读 0赞

之前我们已经在博客[《分布式事务--Fescar》][--Fescar]中了解学习到Fescar相关的架构，接下来我们分别用几篇博客分别来介绍一下Fescar的 TM、RM 和 TC之间的交互流程。

TM、RM和TC之间的交互流程图：

![Architecture][]

简单角色理解：

TC： Fesacr-server应用

TM：dubbo服务调用方

RM：dubbo服务提供方

### 一、示例 ###

官方示例：[https://github.com/alibaba/fescar/wiki/Quick-Start][https_github.com_alibaba_fescar_wiki_Quick-Start]

调用方：

ublic class BusinessServiceImpl implements BusinessService {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(BusinessService.class);
    
        private StorageService storageService;
        private OrderService orderService;
    
        @Override
        @GlobalTransactional(timeoutMills = 300000, name = "dubbo-demo-tx")
        public void purchase(String userId, String commodityCode, int orderCount) {
            storageService.deduct(commodityCode, orderCount);
    
            orderService.create(userId, commodityCode, orderCount);
    
            System.out.println("hello fescar");
    
        }
    
        public void setStorageService(StorageService storageService) {
            this.storageService = storageService;
        }
    
        public void setOrderService(OrderService orderService) {
            this.orderService = orderService;
        }
    
        public static void main(String[] args) {
    
    
    
            ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
                new String[] {"dubbo-business.xml"});
            final BusinessService business = (BusinessService)context.getBean("business");
    
            LOGGER.info("Main business begin ... xid: " + RootContext.getXID());
            business.purchase("U100001", "C00321", 2);
            LOGGER.info("Main business end ... xid: " + RootContext.getXID());
    
    
        }
    }

在示例代码中我们看到了注解@GlobalTransactional，注解的方法：

@GlobalTransactional(timeoutMills = 300000, name = "dubbo-demo-tx")
        public void purchase(String userId, String commodityCode, int orderCount) {
            storageService.deduct(commodityCode, orderCount);
    
            orderService.create(userId, commodityCode, orderCount);
    
            System.out.println("hello fescar");
    
        }

这样我们可以猜测到在真正执行purchase方法前后Fescar肯定做了很多处理操作，当然Fescar也是利用Spring AOP来实现的，实现类为GlobalTransactionalInterceptor，在GlobalTransactionalInterceptor中我们可以看到如下操作：

@Override
        public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
            final GlobalTransactional anno = getAnnotation(methodInvocation.getMethod());
            if (anno != null) {
                try {
                    return transactionalTemplate.execute(new TransactionalExecutor() {
                        @Override
                        public Object execute() throws Throwable {
                            return methodInvocation.proceed();
                        }
    
                        @Override
                        public int timeout() {
                            return anno.timeoutMills();
                        }
    
                        @Override
                        public String name() {
                            if (anno.name() != null) {
                                return anno.name();
                            }
                            return formatMethod(methodInvocation.getMethod());
                        }
                    });
                } catch (TransactionalExecutor.ExecutionException e) {
                    TransactionalExecutor.Code code = e.getCode();
                    switch (code) {
                        case RollbackDone:
                            throw e.getOriginalException();
                        case BeginFailure:
                            failureHandler.onBeginFailure(e.getTransaction(), e.getCause());
                            throw e.getCause();
                        case CommitFailure:
                            failureHandler.onCommitFailure(e.getTransaction(), e.getCause());
                            throw e.getCause();
                        case RollbackFailure:
                            failureHandler.onRollbackFailure(e.getTransaction(), e.getCause());
                            throw e.getCause();
                        default:
                            throw new ShouldNeverHappenException("Unknown TransactionalExecutor.Code: " + code);
    
                    }
                }
    
            }
            return methodInvocation.proceed();
        }

在真正调用方法之前首先调用了TransactionalTemplate的execute方法，在execute中主要调用了一下操作：

（1）tx.begin(business.timeout(), business.name()); 启动事务，在这里面的操作就是想TC（Fesacr-server应用）申请一个全局的事物XID

（2）rs = business.execute();，调用本地方法，当调用远程dubbo服务时会利用dubbo的Filter机制将xid传递到RM服务中

（3）tx.rollback();调用异常时事务回滚，会向TC发送全局事务XID，回滚所有RM的事务。

（4）tx.commit(); 事务提交，会向TC发送全局事物XID，提交所有RM的事务。

通过以上过程分析Fescar将分布式事务在TM层做了抽象处理，简化为单一事务，只需要开启事务、调用、提交事务或回滚，并且事务开启、提交或回滚只需要将操作提交到TC即可，由TC对象去管理各个RM的分片事务。

public Object execute(TransactionalExecutor business) throws TransactionalExecutor.ExecutionException {
    
            // 1. get or create a transaction
            GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
    
            // 2. begin transaction
            try {
                tx.begin(business.timeout(), business.name());
    
            } catch (TransactionException txe) {
                throw new TransactionalExecutor.ExecutionException(tx, txe,
                    TransactionalExecutor.Code.BeginFailure);
    
            }
    
            Object rs = null;
            try {
    
                // Do Your Business
                rs = business.execute();
    
            } catch (Throwable ex) {
    
                // 3. any business exception, rollback.
                try {
                    tx.rollback();
    
                    // 3.1 Successfully rolled back
                    throw new TransactionalExecutor.ExecutionException(tx, TransactionalExecutor.Code.RollbackDone, ex);
    
                } catch (TransactionException txe) {
                    // 3.2 Failed to rollback
                    throw new TransactionalExecutor.ExecutionException(tx, txe,
                        TransactionalExecutor.Code.RollbackFailure, ex);
    
                }
    
            }
    
            // 4. everything is fine, commit.
            try {
                tx.commit();
    
            } catch (TransactionException txe) {
                // 4.1 Failed to commit
                throw new TransactionalExecutor.ExecutionException(tx, txe,
                    TransactionalExecutor.Code.CommitFailure);
    
            }
            return rs;
        }

begin、rollback和commit方法最终在DefaultTransactionManager中远程调用将指令发送到TC（Fesacr-server应用）中。

@Override
        public String begin(String applicationId, String transactionServiceGroup, String name, int timeout) throws TransactionException {
            GlobalBeginRequest request = new GlobalBeginRequest();
            request.setTransactionName(name);
            request.setTimeout(timeout);
            GlobalBeginResponse response = (GlobalBeginResponse) syncCall(request);
            return response.getXid();
        }
    
        @Override
        public GlobalStatus commit(String xid) throws TransactionException {
            long txId = XID.getTransactionId(xid);
            GlobalCommitRequest globalCommit = new GlobalCommitRequest();
            globalCommit.setTransactionId(txId);
            GlobalCommitResponse response = (GlobalCommitResponse) syncCall(globalCommit);
            return response.getGlobalStatus();
        }
    
        @Override
        public GlobalStatus rollback(String xid) throws TransactionException {
            long txId = XID.getTransactionId(xid);
            GlobalRollbackRequest globalRollback = new GlobalRollbackRequest();
            globalRollback.setTransactionId(txId);
            GlobalRollbackResponse response = (GlobalRollbackResponse) syncCall(globalRollback);
            return response.getGlobalStatus();
        }

在syncCall方法中，begin、rollback和commit是同样的参数，不过是给TC发送的一个状态指令

private AbstractTransactionResponse syncCall(AbstractTransactionRequest request) throws TransactionException {
            try {
                return (AbstractTransactionResponse) TmRpcClient.getInstance().sendMsgWithResponse(request);
            } catch (TimeoutException toe) {
                throw new TransactionException(TransactionExceptionCode.IO, toe);
            }
        }

目前Fescar只支持Dubbo作为RPC调用，实现分布式事务，TM和RM之间全局事务XID中传递目前是利用dubbo的Filter来实现的，这样每个RM的分片事务有一个统一的事务XID，这样dubbo的服务提供者（RM对象）在被RPC调用时可以获取全局事务XID。

@Activate(group = { Constants.PROVIDER, Constants.CONSUMER }, order = 100)
    public class TransactionPropagationFilter implements Filter {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(TransactionPropagationFilter.class);
    
        @Override
        public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
            String xid = RootContext.getXID();
            String rpcXid = RpcContext.getContext().getAttachment(RootContext.KEY_XID);
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("xid in RootContext[" + xid + "] xid in RpcContext[" + rpcXid + "]");
            }
            boolean bind = false;
            if (xid != null) {
                RpcContext.getContext().setAttachment(RootContext.KEY_XID, xid);
            } else {
                if (rpcXid != null) {
                    RootContext.bind(rpcXid);
                    bind = true;
                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("bind[" + rpcXid + "] to RootContext");
                    }
                }
            }
            try {
                return invoker.invoke(invocation);
    
            } finally {
                if (bind) {
                    String unbindXid = RootContext.unbind();
                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("unbind[" + unbindXid + "] from RootContext");
                    }
                    if (!rpcXid.equalsIgnoreCase(unbindXid)) {
                        LOGGER.warn("xid in change during RPC from " + rpcXid + " to " + unbindXid);
                        if (unbindXid != null) {
                            RootContext.bind(unbindXid);
                            LOGGER.warn("bind [" + unbindXid + "] back to RootContext");
                        }
                    }
                }
            }
        }
    }

简单流程图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70][]

[--Fescar]: https://blog.csdn.net/qq924862077/article/details/86556559
[Architecture]: /images/20220323/5e778a6ee9314b50ae11e28c382abe53.png
[https_github.com_alibaba_fescar_wiki_Quick-Start]: https://github.com/alibaba/fescar/wiki/Quick-Start
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70]: /images/20220323/cab46ffa55c54ebc94f376062876438b.png