Seata-XA模式 原理

1 XA模式示例

示例参考：github上seata-sample
业务代码层面和xa完全相同，仅数据库代理层面替换成DataSourceProxyXA即可，更多内容，参考示例。

2 架构

注：此图来自seata官网。

3 源码分析

3.1 TM开启全局事务

此过程和AT模式一样，使用@GlobalTransactional即可。

3.2 RM执行分支事务

因为DataSource使用了代理，所以所有DB操作均交个DataSourceProxyXA完成，当执行db操作时，请求将会由ExecuteTemplateXA执行。

3.2.1 执行sql：ExecuteTemplateXA#execute

在以下代码中我们可以看到，其主要逻辑如下：

• 记录下事务提交方式，自动提交 or 非自动提交
• 事务提交方式改为非自动提交，见：connectionProxyXA.setAutoCommit方法。这一步将向TC注册分支事务。
• 执行DB操作。
• 如果执行失败了，那么rollback，见：connectionProxyXA.rollback()
• 如果开始时事务为自动提交，那么commit，见：connectionProxyXA.commit()

public static <T, S extends Statement> T execute(AbstractConnectionProxyXA connectionProxyXA, StatementCallback<T, S> statementCallback, S targetStatement, Object... args) throws SQLException {
    boolean autoCommitStatus = connectionProxyXA.getAutoCommit();
    if (autoCommitStatus) {
        // XA Start
        connectionProxyXA.setAutoCommit(false);
    }
    try {
        T res = null;
        try {
            // execute SQL
            res = statementCallback.execute(targetStatement, args);

        } catch (Throwable ex) {
            if (autoCommitStatus) {
                // XA End & Rollback
                try {
                    connectionProxyXA.rollback();
                } catch (SQLException sqle) {
                    // log and ignore the rollback failure.
                    LOGGER.warn(
                        "Failed to rollback xa branch of " + connectionProxyXA.xid +
                            "(caused by SQL execution failure(" + ex.getMessage() + ") since " + sqle.getMessage(),
                        sqle);
                }
            }

            if (ex instanceof SQLException) {
                throw ex;
            } else {
                throw new SQLException(ex);
            }

        }
        if (autoCommitStatus) {
            try {
                // XA End & Prepare
                connectionProxyXA.commit();
            } catch (Throwable ex) {
                LOGGER.warn(
                    "Failed to commit xa branch of " + connectionProxyXA.xid + ") since " + ex.getMessage(),
                    ex);
                // XA End & Rollback
                if (!(ex instanceof SQLException) || !AbstractConnectionProxyXA.SQLSTATE_XA_NOT_END.equalsIgnoreCase(((SQLException) ex).getSQLState())) {
                    try {
                        connectionProxyXA.rollback();
                    } catch (SQLException sqle) {
                        // log and ignore the rollback failure.
                        LOGGER.warn(
                            "Failed to rollback xa branch of " + connectionProxyXA.xid +
                                "(caused by commit failure(" + ex.getMessage() + ") since " + sqle.getMessage(),
                            sqle);
                    }
                }

                if (ex instanceof SQLException) {
                    throw ex;
                } else {
                    throw new SQLException(ex);
                }

            }
        }
        return res;
    } finally {
        if (autoCommitStatus) {
            connectionProxyXA.setAutoCommit(true);
        }

    }
}
复制代码

3.2.2 设置自动提交方式：ConnectionProxyXA#setAutoCommit

• 向TC注册分支事务。
• 将当前分支事务id和xaResource绑定起来，即：xaResource.start

public void setAutoCommit(boolean autoCommit) throws SQLException {
    if (currentAutoCommitStatus == autoCommit) {
        return;
    }
    if (autoCommit) {
        // According to JDBC spec:
        // If this method is called during a transaction and the
        // auto-commit mode is changed, the transaction is committed.
        if (xaActive) {
            commit();
        }
    } else {
        if (xaActive) {
            throw new SQLException("should NEVER happen: setAutoCommit from true to false while xa branch is active");
        }
        // Start a XA branch
        long branchId = 0L;
        try {
            // 1. register branch to TC then get the branchId
            branchId = DefaultResourceManager.get().branchRegister(BranchType.XA, resource.getResourceId(), null, xid, null,
                null);
        } catch (TransactionException te) {
            cleanXABranchContext();
            throw new SQLException("failed to register xa branch " + xid + " since " + te.getCode() + ":" + te.getMessage(), te);
        }
        // 2. build XA-Xid with xid and branchId
        this.xaBranchXid = XAXidBuilder.build(xid, branchId);
        try {
            // 3. XA Start
            xaResource.start(this.xaBranchXid, XAResource.TMNOFLAGS);
        } catch (XAException e) {
            cleanXABranchContext();
            throw new SQLException("failed to start xa branch " + xid + " since " + e.getMessage(), e);
        }
        // 4. XA is active
        this.xaActive = true;

    }

    currentAutoCommitStatus = autoCommit;
}
复制代码

3.2.3 设置自动提交方式：ConnectionProxyXA#commit

• 终止xaBranchXid分支所执行的工作，即：xaResource.end
• 请求资源管理器准备好xaBranchXid事务的提交工作，即：xaResource.prepare

public void commit() throws SQLException {
    if (currentAutoCommitStatus) {
        // Ignore the committing on an autocommit session.
        return;
    }
    if (!xaActive || this.xaBranchXid == null) {
        throw new SQLException("should NOT commit on an inactive session", SQLSTATE_XA_NOT_END);
    }
    try {
        // XA End: Success
        xaResource.end(xaBranchXid, XAResource.TMSUCCESS);
        // XA Prepare
        xaResource.prepare(xaBranchXid);
        // Keep the Connection if necessary
        keepIfNecessary();
    } catch (XAException xe) {
        try {
            // Branch Report to TC: Failed
            DefaultResourceManager.get().branchReport(BranchType.XA, xid, xaBranchXid.getBranchId(),
                BranchStatus.PhaseOne_Failed, null);
        } catch (TransactionException te) {
            LOGGER.warn("Failed to report XA branch commit-failure on " + xid + "-" + xaBranchXid.getBranchId()
                + " since " + te.getCode() + ":" + te.getMessage() + " and XAException:" + xe.getMessage());

        }
        throw new SQLException(
            "Failed to end(TMSUCCESS)/prepare xa branch on " + xid + "-" + xaBranchXid.getBranchId() + " since " + xe
                .getMessage(), xe);
    } finally {
        cleanXABranchContext();
    }
}
复制代码

3.2.4 设置自动提交方式：ConnectionProxyXA#rollback

• 向TC汇报事务执行失败。
• 终止xaBranchXid分支所执行的工作，即：xaResource.end

public void rollback() throws SQLException {
    if (currentAutoCommitStatus) {
        // Ignore the committing on an autocommit session.
        return;
    }
    if (!xaActive || this.xaBranchXid == null) {
        throw new SQLException("should NOT rollback on an inactive session");
    }
    try {
        // Branch Report to TC
        DefaultResourceManager.get().branchReport(BranchType.XA, xid, xaBranchXid.getBranchId(), BranchStatus.PhaseOne_Failed, null);

    } catch (TransactionException te) {
        // log and ignore the report failure
        LOGGER.warn("Failed to report XA branch rollback on " + xid + "-" + xaBranchXid.getBranchId()
            + " since " + te.getCode() + ":" + te.getMessage());
    }
    try {
        // XA End: Fail
        xaResource.end(xaBranchXid, XAResource.TMFAIL);
    } catch (XAException xe) {
        throw new SQLException(
            "Failed to end(TMFAIL) xa branch on " + xid + "-" + xaBranchXid.getBranchId() + " since " + xe
                .getMessage(), xe);
    } finally {
        cleanXABranchContext();
    }
}
复制代码

3.3 TM向TC发送请求，通知进行Global Commit/Rollback

这个过程和AT模式完全相同。

3.4 TC向各个分支发送Branch Commit/Rollback请求

这个过程和AT模式完全相同。

3.5 RM进行Branch Commit/Rollback

AbstractRMHandler收到请求后，最终交给ResourceManagerXA处理，具体的Commit和Rollback交个xaCommit、xaRollback完成，他们直接触发xaResource的commit、rollback方法。

public void xaCommit(String xid, long branchId, String applicationData) throws XAException {
    XAXid xaXid = XAXidBuilder.build(xid, branchId);
    xaResource.commit(xaXid, false);
    releaseIfNecessary();
}

public void xaRollback(String xid, long branchId, String applicationData) throws XAException {
    XAXid xaXid = XAXidBuilder.build(xid, branchId);
    xaResource.rollback(xaXid);
    releaseIfNecessary();
}
复制代码

3.6 异常补偿

异常补偿流程和AT模式完全相同。

4 XAResource

从上面源码层面可以看到，Seata的XA模式是基于jdk的XAResource接口实现的。以下是jdk文档对此接口的描述。

4.1 XAResource简介

XAResource 接口是基于X/Open CAE规范（分布式事务处理：XA 规范）的工业标准XA接口的Java映射。

在分布式事务处理 (DTP) 环境中，XA接口定义资源管理器和事务管理器之间的协定。JDBC驱动程序或JMS提供者实现此接口，以支持全局事务与数据库或消息服务连接之间的关联。

可由应用程序在外部事务管理器控制事务的环境中使用的任何事务资源均可支持XAResource接口。数据库管理系统就属于此类资源。应用程序可以通过多个数据库连接访问数据。通过事务管理器将每个数据库连接作为事务资源添加到列表中。事务管理器为参与全局事务的每个连接获取XAResource。事务管理器使用start方法建立全局事务与资源之间的关联，而使用end方法取消事务与资源之间的关联。资源管理器负责将全局事务关联到在start与end方法调用之间对其数据执行的所有工作。

在事务提交时，事务管理器通知资源管理器根据二阶段提交协议准备、提交或回滚事务。

4.2 方法摘要

4.3 字段摘要

5 参考文档

Seata XA 模式

Seata 原理

Seata-AT模式

Seata-TCC模式

Seata-Saga模式

Seata-XA模式

TCC-Transaction原理