分布式事务中间件 Fescar—RM 模块源码解读

时间 2019-12-05

原文原文链接

前言

在SOA、微服务架构流行的年代，许多复杂业务上须要支持多资源占用场景，而在分布式系统中由于某个资源不足而致使其它资源占用回滚的系统设计一直是个难点。我所在的团队也遇到了这个问题，为解决这个问题上，团队采用的是阿里开源的分布式中间件Fescar的解决方案，并详细了解了Fescar内部的工做原理，解决在使用Fescar中间件过程当中的一些疑虑的地方，也为后续团队在继续使用该中间件奠基理论基础。git

目前分布式事务解决方案基本是围绕两阶段提交模式来设计的，按对业务是有侵入分为：对业务无侵入的基于XA协议的方案，但须要数据库支持XA协议而且性能较低；对业务有侵入的方案包括：TCC等。Fescar就是基于两阶段提交模式设计的，以高效且对业务零侵入的方式，解决微服务场景下面临的分布式事务问题。Fescar设计上将总体分红三个大模块，即TM、RM、TC，具体解释以下：github

TM（Transaction Manager）：全局事务管理器，控制全局事务边界，负责全局事务开启、全局提交、全局回滚。
RM（Resource Manager）：资源管理器，控制分支事务，负责分支注册、状态汇报，并接收事务协调器的指令，驱动分支（本地）事务的提交和回滚。
TC（Transaction Coordinator）：事务协调器，维护全局事务的运行状态，负责协调并驱动全局事务的提交或回滚。

本文将深刻到Fescar的RM模块源码去介绍Fescar是如何在完成分支提交和回滚的基础上又作到零侵入，进而极大方便业务方进行业务系统开发。sql

1、从配置开始解读

上图是Fescar源码examples模块dubbo-order-service.xml内的配置，数据源采用druid的DruidDataSource，但实际jdbcTemplate执行时并非用该数据源，而用的是Fescar对DruidDataSource的代理DataSourceProxy，因此，与RM相关的代码逻辑基本上都是从DataSourceProxy这个代理数据源开始的。数据库

Fescar采用2PC来完成分支事务的提交与回滚，具体怎么作到的呢，下面就分别介绍Phase一、Phase2具体作了些什么。缓存

2、Phase1—分支（本地）事务执行

Fescar将一个本地事务作为一个分布式事务分支，因此若干个分布在不一样微服务中的本地事务共同组成了一个全局事务，结构以下。
微信

那么，一个本地事务中SQL是如何执行呢？在Spring中，本质上都是从jdbcTemplate开始的，好比下面的SQL语句：架构

jdbcTemplate.update("update storage_tbl set count = count - ? where commodity_code = ?", new Object[] {count, commodityCode});

通常JdbcTemplate执行流程以下图所示：
app

因为在配置中，JdbcTemplate数据源被配置成了Fescar实现DataSourceProxy，进而控制了后续的数据库链接使用的是Fescar提供的ConnectionProxy，Statment使用的是Fescar实现的StatmentProxy，最终Fescar就瓜熟蒂落地实现了在本地事务执行先后增长所须要的逻辑，好比：完成分支事务的快照记录和分支事务执行状态的上报等等。异步

DataSourceProxy获取ConnectionProxy:
async

ConnectionProxy获取StatmentProxy：

在获取到StatmentProxy后，能够调用excute方法执行sql了

而真正excute实现逻辑以下：

首先会检查当前本地事务是否处于全局事务中，若是不处于，直接使用默认的Statment执行，避免因引入Fescar致使非全局事务中的SQL执行性能降低。
解析Sql，有缓存机制，由于有些sql解析会比较耗时，可能会致使在应用启动后刚开始的那段时间里处理全局事务中的sql执行效率下降。
对于INSERT、UPDATE、DELETE、SELECT..FOR UPDATE这四种类型的sql会专门实现的SQL执行器进行处理，其它SQL直接是默认的Statment执行。
返回执行结果，若有异常则直接抛给上层业务代码进行处理。

再来看一下关键的INSERT、UPDATE、DELETE、SELECT..FOR UPDATE这四种类型的sql如何执行的，先看一下具体类图结构：

为结省篇幅，选择UpdateExecutor实现源码看一下，先看入口BaseTransactionalExecutor.execute，该方法将ConnectionProxy与Xid（事务ID）进行绑定,这样后续判断当前本地事务是否处理全局事务中只须要看ConnectionProxy中Xid是否为空。

而后，执行AbstractDMLBaseExecutor中实现的doExecute方法

基本逻辑以下：

先判断是否为Auto-Commit模式
若是非Auto-Commit模式，则先查询Update前对应行记录的快照beforeImage，再执行Update语句，完成后再查询Update后对应行记录的快照afterImage，最后将beforeImage、afterImage生成UndoLog追加到Connection上下文ConnectionContext中。（注：获取beforeImage、afterImage方法在UpdateExecutor类下，通常是构造一条Select...For Update语句获取执行先后的行记录，同时会检查是否有全局锁冲突，具体可参考源码）
若是是Auto-Commit模式，先将提交模式设置成非自动Commit，再执行2中的逻辑，再执行connectionProxy.commit()方法，因为执行2过程和commit时均可能会出现全局锁冲突问题，增长了一个循环等待重试逻辑，最后将connection的模式设置成Auto-Commit模式

若是本地事务执行过程当中发生异常，业务上层会接收到该异常，至因而给TM模块返回成功仍是失败，由业务上层实现决定，若是返回失败，则TM裁决对全局事务进行回滚；若是本地事务执行过程未发生异常，不论是非Auto-Commit仍是Auto-Commit模式，最后都会调用connectionProxy.commit()对本地事务进行提交，在这里会建立分支事务、上报分支事务的状态以及将UndoLog持久化到undo_log表中，具体代码以下图：

基本逻辑：

判断当前本地事务是否处于全局事务中（也就判断ConnectionContext中的xid是否为空）。
若是不处于全局事务中，则调用targetConnection对本地事务进行commit。
若是处于全局事务中，首先建立分支事务，再将ConnectionContext中的UndoLog写入到undo_log表中，而后调用targetConnection对本地事务进行commit，将UndoLog与业务SQL一块儿提交，最后上报分支事务的状态（成功 or 失败），并将ConnectionContext上下文重置。

综上所述，RM模块经过对JDBC数据源进行代理，干预业务SQL执行过程，加入了不少流程，好比业务SQL解析、业务SQL执行先后的数据快照查询并组织成UndoLog、全局锁检查、分支事务注册、UndoLog写入并随本地事务一块儿Commit、分支事务状态上报等。经过这种方式，Fescar真正作到了对业务代码无侵入，只须要经过简单的配置，业务方就能够轻松享受Fescar所带来的功能。Phase1总体流程引用Fescar官方图总结以下：

3、Phase2-分支事务提交或回滚

阶段2完成的是全局事物的最终提交或回滚，当全局事务中全部分支事务所有完成而且都执行成功，这时TM会发起全局事务提交,TC收到全全局事务提交消息后，会通知各分支事务进行提交；同理，当全局事务中全部分支事务所有完成而且某个分支事务失败了，TM会通知TC协调全局事务回滚，进而TC通知各分支事务进行回滚。

在业务应用启动过程当中，因为引入了Fescar客户端，RmRpcClient会随应用一块儿启动，该RmRpcClient采用Netty实现，能够接收TC消息和向TC发送消息，所以RmRpcClient是与TC收发消息的关键模块。

public class RMClientAT {

    public static void init(String applicationId, String transactionServiceGroup) {
        RmRpcClient rmRpcClient = RmRpcClient.getInstance(applicationId, transactionServiceGroup);
        AsyncWorker asyncWorker = new AsyncWorker();
        asyncWorker.init();
        DataSourceManager.init(asyncWorker);
        rmRpcClient.setResourceManager(DataSourceManager.get());
        rmRpcClient.setClientMessageListener(new RmMessageListener(new RMHandlerAT()));
        rmRpcClient.init();
    }
}

上述代码展现是的RmRpcClient初始化过程，有三个关键类RMHandlerAT、AsyncWorker和DataSourceManager。RMHandlerAT具备了分支提交和回滚两个方法，分支提交或回滚的逻辑能够从这里开始看；AsyncWorker是一个异步Worker,主要是完成分支事务异步提交的功能，具备失败重试功能；DataSourceManager对数据源管理和维护。

下面分红两部分来说：分支事务提交、分去事务回滚。

3.一、分支事务提交

在接收到TC发起的全局提交消息后，经RmRpcClient对通讯协议的处理，再交由RMHandlerAT来完成对分支事务的提交，分支事务提交从RMHandlerAT.doBranchCommit()开始，但最后由AsyncWorker异步Worker完成，直接看AsyncWorker中的代码实现：

分支事务提交关键逻辑在doBranchCommits方法中：

该方法主要是批量删除UndoLog日志，但并未使用ConnectionProxy去执行删除SQL，可能缘由是：一、彻底不必二、考虑效率优先

一样，对于分支事务提交也引用Fescar官方一张图来结尾：

3.二、分支事务回滚

一样，分支事务回滚是从RMHandlerAT.doBranchRollback开始的，而后到了dataSourceManager.branchRollback，最后完成分支事务回滚逻辑的是UndoLogManager.undo方法。

@Override
    protected void RMHandlerAT：doBranchRollback(BranchRollbackRequest request, BranchRollbackResponse response) throws TransactionException {
        String xid = request.getXid();
        long branchId = request.getBranchId();
        String resourceId = request.getResourceId();
        String applicationData = request.getApplicationData();
        LOGGER.info("AT Branch rolling back: " + xid + " " + branchId + " " + resourceId);
        BranchStatus status = dataSourceManager.branchRollback(xid, branchId, resourceId, applicationData);
        response.setBranchStatus(status);
        LOGGER.info("AT Branch rollback result: " + status);
    }
    
     @Override
    public BranchStatus DataSourceManager：branchRollback(String xid, long branchId, String resourceId, String applicationData) throws TransactionException {
        DataSourceProxy dataSourceProxy = get(resourceId);
        if (dataSourceProxy == null) {
            throw new ShouldNeverHappenException();
        }
        try {
            UndoLogManager.undo(dataSourceProxy, xid, branchId);
        } catch (TransactionException te) {
            if (te.getCode() == TransactionExceptionCode.BranchRollbackFailed_Unretriable) {
                return BranchStatus.PhaseTwo_RollbackFailed_Unretryable;
            } else {
                return BranchStatus.PhaseTwo_RollbackFailed_Retryable;
            }
        }
        return BranchStatus.PhaseTwo_Rollbacked;
    }

UndoLogManager.undo方法源码以下：

从上图能够看出，整个回滚到全局事务以前状态的代码逻辑集中在以下代码中：

AbstractUndoExecutor undoExecutor = UndoExecutorFactory.getUndoExecutor(dataSourceProxy.getDbType(), sqlUndoLog);
undoExecutor.executeOn(conn);

首先经过UndoExecutorFactory获取到对应的UndoExecutor，而后再执行UndoExecutor的executeOn方法完成回滚操做。目前三种类型的UndoExecutor结构以下：

undoExecutor.executeOn源码以下：

至此，整个分支事务回滚就结束了，分支事务回滚总体时序图以下：

引入Fescar官方对分支事务回滚原理介绍图做为结尾：

综合上述，Fescar在Phase2经过UndoLog自动完成分支事务提交与回滚，在这个过程当中不须要业务方作任何处理，业务方无感知，因些在该阶段对业务代码也是无侵入的。

4、总结

本文主要介绍了RM模块的相关代码，将RM模块按2PC模式分红Phase1和Phase2分别进行介绍，从Fescar源码上看，整个源码结构清晰，有利于研发人员快速学习Fescar的原理。在使用方面，只需进行简单的配置，就能够享受Fescar带来的便捷功能，对业务作到了无侵入；同时在性能方面，Fescar在分支事务提交过程当中采用异步模式，减小了全局锁的占用时间，进而提高了总体性能。后续，将继续学习Fescar的其它模块（TM、TC）与全局锁的实现逻辑，并作相关总结介绍。

参考

Fescar官方介绍

原文连接更多技术干货请关注阿里云云栖社区微信号：yunqiinsight