琼殿技术论坛»首页 › 后端架构/服务器运行环境 › MySQL论坛 › 1、SEATA分布式事务——XA模式
返回列表 发新帖
查看: 89|回复: 0

1、SEATA分布式事务——XA模式

[复制链接]
舒友

1

主题

0

回帖

0

积分

积极分子

金币
0
阅读权限
220
精华
0
威望
0
贡献
0
在线时间
0 小时
注册时间
2009-1-12
发表于 2026-3-28 09:18:00 | 显示全部楼层 |阅读模式

一、传统分布式XA事务的2PC

  2PC 即两阶段提交协议，是将整个事务流程分为两个阶段，准备阶段（Prepare phase）、提交阶段（commit phase），2 是指两个阶段，P 是指准备阶段，C 是指提交阶段。常见的关系型数据库如 Oracle、MySQL 都支持两阶段提交协议，如下图：

  • 成功情况

  • 失败情况

①、准备阶段（Prepare phase）：TM（事务管理器）给每个RM（资源管理器，也就是数据库）发送 Prepare 消息，每个RM在本地执行事务，并写本地的 Undo/Redo （Undo 日志是记录修改前的数据，用于数据库回滚，Redo 日志是记录修改后的数据，用于提交事务后写入数据文件）日志，此时事务没有提交。
②、提交阶段（commit phase）：如果TM（事务管理器）收到了RM（资源管理器，也就是数据库）的执行失败或者超时消息时，直接给每个RM发送回滚（Rollback）消息；否则，发送提交（Commit）消息；参与者根据事务管理器的指令执行提交（Commit）或者回滚（Rollback）操作，并释放事务处理过程中使用的锁资源。注意：必须在最后阶段释放锁资源。
  传统XA事务的详细细节和实现方式，请查看我的另一篇博客：mysql数据库事务的实现和XA事务

二、Seata的XA模式

  seata实现分布式事务的样例程序
  在 Seata 定义的分布式事务框架内，XA模式是利用RM（资源管理器，也就是数据库）对 XA 协议的支持，以 XA 协议的机制来管理分支事务的一种事务模式，如下所示：


从上面的图可以看到，seata XA 模式分为以下5个步骤：
①、TM（事务管理器） 开启全局事务；
②、RM 向 TC（事务协调者） 注册分支事务；
③、RM 向 TC 报告分支事务状态；
④、TC 向 RM （资源管理器，也就是数据库）发送commit/rollback 请求；
⑤、TM 结束全局事务Global Commit/Rollback
  负责RM 客户端的类是RmNettyRemotingClient.class，这个类的UML图如下所示：

RmNettyRemotingClient.class的父类中的内部类AbstractNettyRemotingClient.class::ClientHandler.class来处理 TC 发来的请求并再次委托给父类AbstractNettyRemoting.class::processMessage()函数来处理TC发来的请求

  • AbstractNettyRemotingClient.class::ClientHandler.class的源码如下：
...省略部分导包代码...
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
...省略部分导包代码...
public abstract class AbstractNettyRemotingClient extends AbstractNettyRemoting implements RemotingClient {
    ...省略部分代码...
    @Sharable
    class ClientHandler extends ChannelDuplexHandler {
        ClientHandler() {
        }
    
        public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof RpcMessage) {
                AbstractNettyRemotingClient.this.processMessage(ctx, (RpcMessage)msg);
            } else {
                AbstractNettyRemotingClient.LOGGER.error("rpcMessage type error");
            }
    
        }
    }
    ...省略部分代码...
}
  • AbstractNettyRemoting.class::processMessage()的源码如下：
public abstract class AbstractNettyRemoting implements Disposable {
    protected void processMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(String.format("%s msgId:%s, body:%s", this, rpcMessage.getId(), rpcMessage.getBody()));
        }
        Object body = rpcMessage.getBody();
        if (body instanceof MessageTypeAware) {
            MessageTypeAware messageTypeAware = (MessageTypeAware) body;
            final Pair<RemotingProcessor, ExecutorService> pair = this.processorTable.get((int) messageTypeAware.getTypeCode());
            if (pair != null) {
                if (pair.getSecond() != null) {
                    try {
                        pair.getSecond().execute(() -> {
                            try {
                                //最终调用的是RmBranchCommitProcessor.class的process()函数
                                pair.getFirst().process(ctx, rpcMessage);
                            } catch (Throwable th) {
                                LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
                            } finally {
                                MDC.clear();
                            }
                        });
                    } catch (RejectedExecutionException e) {
                        LOGGER.error(FrameworkErrorCode.ThreadPoolFull.getErrCode(),
                            "thread pool is full, current max pool size is " + messageExecutor.getActiveCount());
                        if (allowDumpStack) {
                            String name = ManagementFactory.getRuntimeMXBean().getName();
                            String pid = name.split("@")[0];
                            long idx = System.currentTimeMillis();
                            try {
                                String jstackFile = idx + ".log";
                                LOGGER.info("jstack command will dump to " + jstackFile);
                                Runtime.getRuntime().exec(String.format("jstack %s > %s", pid, jstackFile));
                            } catch (IOException exx) {
                                LOGGER.error(exx.getMessage());
                            }
                            allowDumpStack = false;
                        }
                    }
                } else {
                    try {
                        pair.getFirst().process(ctx, rpcMessage);
                    } catch (Throwable th) {
                        LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
                    }
                }
            } else {
                LOGGER.error("This message type [{}] has no processor.", messageTypeAware.getTypeCode());
            }
        } else {
            LOGGER.error("This rpcMessage body[{}] is not MessageTypeAware type.", body);
        }
    }
}
  • RmBranchCommitProcessor.class的源码如下：
package org.apache.seata.core.rpc.processor.client;

import io.netty.channel.ChannelHandlerContext;
import org.apache.seata.common.util.NetUtil;
import org.apache.seata.core.protocol.RpcMessage;
import org.apache.seata.core.protocol.transaction.BranchCommitRequest;
import org.apache.seata.core.protocol.transaction.BranchCommitResponse;
import org.apache.seata.core.rpc.RemotingClient;
import org.apache.seata.core.rpc.RpcContext;
import org.apache.seata.core.rpc.TransactionMessageHandler;
import org.apache.seata.core.rpc.processor.RemotingProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RmBranchCommitProcessor implements RemotingProcessor {
    private static final Logger LOGGER = LoggerFactory.getLogger(RmBranchCommitProcessor.class);
    private TransactionMessageHandler handler;
    private RemotingClient remotingClient;

    public RmBranchCommitProcessor(TransactionMessageHandler handler, RemotingClient remotingClient) {
        this.handler = handler;
        this.remotingClient = remotingClient;
    }

    public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        String remoteAddress = NetUtil.toStringAddress(ctx.channel().remoteAddress());
        Object msg = rpcMessage.getBody();
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("rm client handle branch commit process:" + msg);
        }

        this.handleBranchCommit(rpcMessage, remoteAddress, (BranchCommitRequest)msg);
    }

    private void handleBranchCommit(RpcMessage request, String serverAddress, BranchCommitRequest branchCommitRequest) {
        BranchCommitResponse resultMessage = (BranchCommitResponse)this.handler.onRequest(branchCommitRequest, (RpcContext)null);
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("branch commit result:" + resultMessage);
        }

        try {
            this.remotingClient.sendAsyncResponse(serverAddress, request, resultMessage);
        } catch (Throwable var6) {
            LOGGER.error("branch commit error: {}", var6.getMessage(), var6);
        }

    }
}

seata 的 xa 模式是两阶段提交：
①、第一阶段先执行 XA Start、执行SQL、XA End三个步骤，之后直接执行XA Prepare。
②、第二阶段执行 XA commit/rollback。
但是oracle数据库不支持，因为 oracle 实现的是标准的 xa 协议，即 xa end 后，TC（事务协调者）向RM （资源管理器，也就是数据库）统一发送 prepare，最后再发送 commit/rollback。这也导致了 seata 的 xa 模式对 oracle 数据库的支持不太好。

2.1、xa模式数据源代理与at模式数据源代理的区别

  seata 中的 XA 模式是使用数据源代理来实现的，需要手动配置数据源代理，代码如下：

import org.apache.seata.rm.datasource.xa.DataSourceProxyXA;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import javax.sql.DataSource;

@Bean
@ConfigurationProperties(prefix = "spring.datasource")
public DruidDataSource druidDataSource() {
    //也可以根据普通 DataSource 来创建 XAConnection，但是这种方式有兼容性问题（比如 oracle数据库不支持）
    return new DruidDataSource();
}

@Bean("dataSourceProxy")
public DataSource dataSource(DruidDataSource druidDataSource) {
    //所以 seata 使用了开发者自己配置 XADataSource
    //seata 提供的 XA 数据源代理，要求代码框架中必须使用 druid 连接池
    return new DataSourceProxyXA(druidDataSource);
}

①、在数据源代理根据普通数据源（DataSource）中获取的普通 JDBC 连接创建出相应的 Connection的数据源代理方式中，XA模式与 AT 模式的数据源代理机制的区别，如下所示：

DataSourceProxyXA、ConnectionProxyXA、StatementProxyXA的UML关系图，如下所示：

  • DataSourceProxyXA.class的部分源码
...省略部分导包代码...
import org.apache.seata.rm.datasource.util.XAUtils;
import javax.sql.DataSource;
import javax.sql.XAConnection;
...省略部分导包代码...
public class DataSourceProxyXA extends AbstractDataSourceProxyXA {
    ...省略部分代码...
    protected Connection getConnectionProxy(Connection connection) throws SQLException {
        return !RootContext.inGlobalTransaction() ? connection : this.getConnectionProxyXA(connection);
    }
    
    protected Connection getConnectionProxyXA() throws SQLException {
        Connection connection = this.dataSource.getConnection();
        return this.getConnectionProxyXA(connection);
    }
    //创建ConnectionProxyXA和XAConnection 
    private Connection getConnectionProxyXA(Connection connection) throws SQLException {
        Connection physicalConn = (Connection)connection.unwrap(Connection.class);
        XAConnection xaConnection = XAUtils.createXAConnection(physicalConn, this);
        ConnectionProxyXA connectionProxyXA = new ConnectionProxyXA(connection, xaConnection, this, RootContext.getXID());
        connectionProxyXA.init();
        return connectionProxyXA;
    }
    ...省略部分代码...
}
  • ConnectionProxyXA .class的部分源码和AbstractConnectionProxyXA.class的部分源码
...省略部分导包代码...
public class ConnectionProxyXA extends AbstractConnectionProxyXA implements Holdable {
    private static final Logger LOGGER = LoggerFactory.getLogger(ConnectionProxyXA.class);
    private static final int BRANCH_EXECUTION_TIMEOUT = ConfigurationFactory.getInstance().getInt("client.rm.branchExecutionTimeoutXA", 60000);
    private volatile boolean currentAutoCommitStatus = true;
    private volatile XAXid xaBranchXid;
    private volatile boolean xaActive = false;
    private volatile boolean xaEnded = false;
    private volatile boolean kept = false;
    private volatile boolean rollBacked = false;
    private volatile Long branchRegisterTime = null;
    private volatile Long prepareTime = null;
    private static final Integer TIMEOUT;
    private boolean shouldBeHeld = false;

    public ConnectionProxyXA(Connection originalConnection, XAConnection xaConnection, BaseDataSourceResource resource, String xid) {
        super(originalConnection, xaConnection, resource, xid);
        this.shouldBeHeld = resource.isShouldBeHeld();
    }
    
    ...省略部分代码...
}

package org.apache.seata.rm.datasource.xa;
...省略部分导包代码...
import javax.sql.XAConnection;
import java.sql.Connection;
...省略部分导包代码...
public abstract class AbstractConnectionProxyXA implements Connection {
     ...省略部分代码...
    protected Connection originalConnection;
    public AbstractConnectionProxyXA(Connection originalConnection, XAConnection xaConnection, BaseDataSourceResource resource, String xid) {
        this.originalConnection = originalConnection;
        this.xaConnection = xaConnection;
        this.resource = resource;
        this.xid = xid;
    }
    //用重载的方式创建不同的StatementProxyXA 
    @Override
    public Statement createStatement() throws SQLException {
        Statement targetStatement = originalConnection.createStatement();
        return new StatementProxyXA(this, targetStatement);
    }
    
    @Override
    public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
        Statement statement = originalConnection.createStatement(resultSetType, resultSetConcurrency);
        return new StatementProxyXA(this, statement);
    }
    
    @Override
    public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability)
            throws SQLException {
        Statement statement = originalConnection.createStatement(resultSetType, resultSetConcurrency,
                resultSetHoldability);
        return new StatementProxyXA(this, statement);
    }
     ...省略部分代码...
}
  • StatementProxyXA.class的部分源码
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLWarning;
import java.sql.Statement;
/**
 * Statement proxy for XA mode.
 *
 */
public class StatementProxyXA implements Statement {

    protected AbstractConnectionProxyXA connectionProxyXA;

    protected Statement targetStatement;

    public StatementProxyXA(AbstractConnectionProxyXA connectionProxyXA, Statement targetStatement) {
        this.connectionProxyXA = connectionProxyXA;
        this.targetStatement = targetStatement;
    }
    ...省略部分代码...
}

②、在数据源代理是指定的XA 数据源（XADataSource）进行代理方式中获取的普通 JDBC 连接创建出相应的 Connection的数据源代理方式中，XA模式与 AT模式的数据源代理机制的区别，如下所示：

作者在SEATA的2.3.0版本中没有找到XADataSourceProxy.class、XAConnectionProxy.class、StatementProxyXA.class

2.2、XA 第一阶段的部分源码

  XA第一阶段是指XA Start、执行SQL、XA End、XA Prepare，如下所示：

当 RM 收到 DML 请求后，seata 会使用 ExecuteTemplateXA.class中的静态函数execute()来执行，execute()中有一个地方很关键，就是把 autocommit 属性改为了 false，而 mysql 默认 autocommit 是 true。事务提交之后，还要把 autocommit 改回默认。如下所示：

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLWarning;
import java.sql.Statement;
/**
 * Statement proxy for XA mode.
 *
 */
public class StatementProxyXA implements Statement {
    protected AbstractConnectionProxyXA connectionProxyXA;
    
    protected Statement targetStatement;
    
    public StatementProxyXA(AbstractConnectionProxyXA connectionProxyXA, Statement targetStatement) {
        this.connectionProxyXA = connectionProxyXA;
        this.targetStatement = targetStatement;
    }
    
    @Override
    public int executeUpdate(String sql) throws SQLException {
        return ExecuteTemplateXA.execute(connectionProxyXA, (statement, args) -> statement.executeUpdate(
            (String)args[0]), targetStatement, sql);
    }
    ...省略部分代码...
}
2.2.1、XA Start环节
  • ExecuteTemplateXA .class的的部分源码
package org.apache.seata.rm.datasource.xa;
import org.apache.seata.rm.datasource.exec.StatementCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.SQLException;
import java.sql.Statement;
public class ExecuteTemplateXA {
    
    private static final Logger LOGGER = LoggerFactory.getLogger(ExecuteTemplateXA.class);

    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);
        }
        ...省略部分代码...
    }
     ...省略部分代码...
}
  • ConnectionProxyXA .class的部分源码——真正开启XA Start环节
package org.apache.seata.rm.datasource.xa;

import java.sql.Connection;
import java.sql.SQLException;
import javax.sql.PooledConnection;
import javax.sql.XAConnection;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import org.apache.seata.common.DefaultValues;
import org.apache.seata.common.util.StringUtils;
import org.apache.seata.config.ConfigurationFactory;
import org.apache.seata.core.exception.TransactionException;
import org.apache.seata.core.model.BranchStatus;
import org.apache.seata.core.model.BranchType;
import org.apache.seata.rm.BaseDataSourceResource;
import org.apache.seata.rm.DefaultResourceManager;
import org.apache.seata.rm.datasource.util.SeataXAResource;
import org.apache.seata.sqlparser.util.JdbcConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConnectionProxyXA extends AbstractConnectionProxyXA implements Holdable {
     ...省略部分代码...
    @Override
    public void setAutoCommit(boolean autoCommit) throws SQLException {
        if (currentAutoCommitStatus == autoCommit) {
            return;
        }
        if (isReadOnly()) {
            //If it is a read-only transaction, do nothing
            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;
            try {
                // 1. register branch to TC then get the branch message
                branchRegisterTime = System.currentTimeMillis();
                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);
            // Keep the Connection if necessary
            keepIfNecessary();
            try {
                start();//开启XA事务
            } 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;
    }
    //最终调用了start()函数才真正开启了XA Start环节
    private synchronized void start() throws XAException, SQLException {
        // 3. XA Start
        if (JdbcConstants.ORACLE.equals(resource.getDbType())) {
            xaResource.start(this.xaBranchXid, SeataXAResource.ORATRANSLOOSE);
        } else {
            xaResource.start(this.xaBranchXid, XAResource.TMNOFLAGS);
        }
    
        try {
            termination();
        } catch (SQLException e) {
            // the framework layer does not actively call ROLLBACK when setAutoCommit throws an SQL exception
            xaResource.end(this.xaBranchXid, XAResource.TMFAIL);
            xaRollback(xaBranchXid);
            // Branch Report to TC: Failed
            reportStatusToTC(BranchStatus.PhaseOne_Failed);
            throw  e;
        }
    }
     ...省略部分代码...
}
2.2.2、执行SQL环节
  • ExecuteTemplateXA .class的部分源码
package org.apache.seata.rm.datasource.xa;
import org.apache.seata.rm.datasource.exec.StatementCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.SQLException;
import java.sql.Statement;
/**
 * The type Execute template.
 *
 */
public class ExecuteTemplateXA {

    private static final Logger LOGGER = LoggerFactory.getLogger(ExecuteTemplateXA.class);

    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 {
                //执行SQL环节，最终调用的是StatementProxyXA.class中的匿名内部类执行了SQL
                // execute SQL
                res = statementCallback.execute(targetStatement, args);
                        ...省略部分代码...
    }
    ...省略部分代码...
}
  • StatementProxyXA .class的部分源代码，最终调用的是StatementProxyXA.class中的匿名内部类执行了SQL
public class StatementProxyXA implements Statement {
    protected AbstractConnectionProxyXA connectionProxyXA;
    
    protected Statement targetStatement;
    
    public StatementProxyXA(AbstractConnectionProxyXA connectionProxyXA, Statement targetStatement) {
        this.connectionProxyXA = connectionProxyXA;
        this.targetStatement = targetStatement;
    }
    
    @Override
    public int executeUpdate(String sql) throws SQLException {
        //最终调用了StatementProxyXA.class中的匿名内部类执行了SQL
        return ExecuteTemplateXA.execute(connectionProxyXA, (statement, args) -> statement.executeUpdate(
            (String)args[0]), targetStatement, sql);
    }
    ...省略部分代码...
}
2.2.3、XA End环节和XA Prepare环节
  • ExecuteTemplateXA .class的的部分源码
package org.apache.seata.rm.datasource.xa;
import org.apache.seata.rm.datasource.exec.StatementCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.SQLException;
import java.sql.Statement;
/**
 * The type Execute template.
 *
 */
public class ExecuteTemplateXA {

    private static final Logger LOGGER = LoggerFactory.getLogger(ExecuteTemplateXA.class);

    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
                    //XA End环节和XA Prepare环节
                    connectionProxyXA.commit();
                    ...省略部分代码...
    }
    ...省略部分代码...
}
  • ConnectionProxyXA .class的部分源码——真正执行XA End环节和XA Prepare环节
package org.apache.seata.rm.datasource.xa;

import java.sql.Connection;
import java.sql.SQLException;
import javax.sql.PooledConnection;
import javax.sql.XAConnection;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import org.apache.seata.common.DefaultValues;
import org.apache.seata.common.util.StringUtils;
import org.apache.seata.config.ConfigurationFactory;
import org.apache.seata.core.exception.TransactionException;
import org.apache.seata.core.model.BranchStatus;
import org.apache.seata.core.model.BranchType;
import org.apache.seata.rm.BaseDataSourceResource;
import org.apache.seata.rm.DefaultResourceManager;
import org.apache.seata.rm.datasource.util.SeataXAResource;
import org.apache.seata.sqlparser.util.JdbcConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConnectionProxyXA extends AbstractConnectionProxyXA implements Holdable {
     ...省略部分代码...
    @Override
    public synchronized void commit() throws SQLException {
        if (currentAutoCommitStatus || isReadOnly()) {
            // Ignore the committing on an autocommit session and read-only transaction.
            return;
        }
        if (!xaActive || this.xaBranchXid == null) {
            throw new SQLException("should NOT commit on an inactive session", SQLSTATE_XA_NOT_END);
        }
        try {
            // XA End: Success
            try {
                end(XAResource.TMSUCCESS);
            } catch (SQLException sqle) {
                // Rollback immediately before the XA Branch Context is deleted.
                String xaBranchXid = this.xaBranchXid.toString();
                rollback();
                throw new SQLException("Branch " + xaBranchXid + " was rollbacked on committing since " + sqle.getMessage(), SQLSTATE_XA_NOT_END, sqle);
            }
            long now = System.currentTimeMillis();
            checkTimeout(now);
            setPrepareTime(now);
            int prepare = xaResource.prepare(xaBranchXid);
            // Based on the four databases: MySQL (8), Oracle (12c), Postgres (16), and MSSQL Server (2022),
            // only Oracle has read-only optimization; the others do not provide read-only feedback.
            // Therefore, the database type check can be eliminated here.
            if (prepare == XAResource.XA_RDONLY) {
                // Branch Report to TC: RDONLY
                reportStatusToTC(BranchStatus.PhaseOne_RDONLY);
            }
        } catch (XAException xe) {
            // Branch Report to TC: Failed
            reportStatusToTC(BranchStatus.PhaseOne_Failed);
            throw new SQLException(
                "Failed to end(TMSUCCESS)/prepare xa branch on " + xid + "-" + xaBranchXid.getBranchId() + " since " + xe
                    .getMessage(), xe);
        } finally {
            cleanXABranchContext();
        }
    }
    private void xaEnd(XAXid xaXid, int flags) throws XAException {
        if (!xaEnded) {
            xaResource.end(xaXid, flags);
            xaEnded = true;
        }
    }
    
    private synchronized void end(int flags) throws XAException, SQLException {
        xaEnd(xaBranchXid, flags);
        termination();
    }
    
    private void termination() throws SQLException {
        termination(this.xaBranchXid.toString());
    }
    
    private void termination(String xaBranchXid) throws SQLException {
        // if it is not empty, the resource will hang and need to be terminated early
        BranchStatus branchStatus = BaseDataSourceResource.getBranchStatus(xaBranchXid);
        if (branchStatus != null) {
            releaseIfNecessary();
            throw new SQLException("failed xa branch " + xid
                    + " the global transaction has finish, branch status: " + branchStatus.getCode());
        }
    }
    //给TC汇报XA Commit状态和XA Rollback状态
    private void reportStatusToTC(BranchStatus status) {
        try {
            DefaultResourceManager.get().branchReport(BranchType.XA, xid, xaBranchXid.getBranchId(),
                    status, null);
        } catch (TransactionException te) {
            LOGGER.warn("Failed to report XA branch {} on {}-{} since {}:{}",
                    status, xid, xaBranchXid.getBranchId(), te.getCode(), te.getMessage());
        }
    }
     ...省略部分代码...
}
2.3、XA 第二阶段的部分源码

  XA第二阶段是指XA Commit或者XA Rollback，如下所示：

当XA模式的第一阶段执行完成后，便会根据第一阶段的执行结果来执行XA模式的第二阶段，如下图所示：

  • RmBranchCommitProcessor.class的部分源码
...省略导包代码...
public class RmBranchCommitProcessor implements RemotingProcessor {
     ...省略部分代码...
    @Override
    public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        String remoteAddress = NetUtil.toStringAddress(ctx.channel().remoteAddress());
        Object msg = rpcMessage.getBody();
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("rm client handle branch commit process:" + msg);
        }
        handleBranchCommit(rpcMessage, remoteAddress, (BranchCommitRequest) msg);
    }
    
    private void handleBranchCommit(RpcMessage request, String serverAddress, BranchCommitRequest branchCommitRequest) {
        BranchCommitResponse resultMessage;
        resultMessage = (BranchCommitResponse) handler.onRequest(branchCommitRequest, null);
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("branch commit result:" + resultMessage);
        }
        try {
            this.remotingClient.sendAsyncResponse(serverAddress, request, resultMessage);
        } catch (Throwable throwable) {
            LOGGER.error("branch commit error: {}", throwable.getMessage(), throwable);
        }
    }
     ...省略部分代码...
}
  • AbstractRMHandler.class的部分源码
...省略导包代码...
public abstract class AbstractRMHandler extends AbstractExceptionHandler
    implements RMInboundHandler, TransactionMessageHandler {
     ...省略部分代码...
    @Override
    public BranchCommitResponse handle(BranchCommitRequest request) {
        BranchCommitResponse response = new BranchCommitResponse();
        exceptionHandleTemplate(new AbstractCallback<BranchCommitRequest, BranchCommitResponse>() {
            @Override
            public void execute(BranchCommitRequest request, BranchCommitResponse response)
                throws TransactionException {
                doBranchCommit(request, response);
            }
        }, request, response);
        return response;
    }
    
    @Override
    public BranchRollbackResponse handle(BranchRollbackRequest request) {
        BranchRollbackResponse response = new BranchRollbackResponse();
        exceptionHandleTemplate(new AbstractCallback<BranchRollbackRequest, BranchRollbackResponse>() {
            @Override
            public void execute(BranchRollbackRequest request, BranchRollbackResponse response)
                throws TransactionException {
                doBranchRollback(request, response);
            }
        }, request, response);
        return response;
    }  
    
    protected void doBranchCommit(BranchCommitRequest request, BranchCommitResponse response)
        throws TransactionException {
        String xid = request.getXid();
        long branchId = request.getBranchId();
        String resourceId = request.getResourceId();
        String applicationData = request.getApplicationData();
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Branch committing: " + xid + " " + branchId + " " + resourceId + " " + applicationData);
        }
        BranchStatus status = getResourceManager().branchCommit(request.getBranchType(), xid, branchId, resourceId,
            applicationData);
        response.setXid(xid);
        response.setBranchId(branchId);
        response.setBranchStatus(status);
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Branch commit result: " + status);
        }
    
    }
    
    protected void doBranchRollback(BranchRollbackRequest request, BranchRollbackResponse response)
        throws TransactionException {
        String xid = request.getXid();
        long branchId = request.getBranchId();
        String resourceId = request.getResourceId();
        String applicationData = request.getApplicationData();
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Branch Rollbacking: " + xid + " " + branchId + " " + resourceId);
        }
        BranchStatus status = getResourceManager().branchRollback(request.getBranchType(), xid, branchId, resourceId,
            applicationData);
        response.setXid(xid);
        response.setBranchId(branchId);
        response.setBranchStatus(status);
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Branch Rollbacked result: " + status);
        }
    }
    
    @Override
    public AbstractResultMessage onRequest(AbstractMessage request, RpcContext context) {
        if (!(request instanceof AbstractTransactionRequestToRM)) {
            throw new IllegalArgumentException();
        }
        AbstractTransactionRequestToRM transactionRequest = (AbstractTransactionRequestToRM)request;
        transactionRequest.setRMInboundMessageHandler(this);
    
        return transactionRequest.handle(context);
    }
     ...省略部分代码...
}
  • BranchCommitRequest.class的部分源码
...省略导包代码...
public class BranchCommitRequest extends AbstractBranchEndRequest {

    @Override
    public short getTypeCode() {
        return MessageType.TYPE_BRANCH_COMMIT;
    }

    @Override
    public AbstractTransactionResponse handle(RpcContext rpcContext) {
        return handler.handle(this);
    }


}
  • ResourceManagerXA.class的部分源码
...省略导包代码...
public class ResourceManagerXA extends AbstractDataSourceCacheResourceManager {
     ...省略部分代码...
    @Override
    public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId,
                                     String applicationData) throws TransactionException {
        return finishBranch(true, branchType, xid, branchId, resourceId, applicationData);
    }
    
    @Override
    public BranchStatus branchRollback(BranchType branchType, String xid, long branchId, String resourceId,
                                       String applicationData) throws TransactionException {
        return finishBranch(false, branchType, xid, branchId, resourceId, applicationData);
    }
     ...省略部分代码...
}

ResourceManagerXA.class的UML关系图，如下所示：

上述类的调用关系，如下时序图所示：

ResourceManagerXA.class::finishBranch()函数最终调用了ConnectionProxyXA.class::xaCommit()函数或者ConnectionProxyXA.class::xaRollback()函数，如下所示：

...省略导包代码...
public class ConnectionProxyXA extends AbstractConnectionProxyXA implements Holdable {
     ...省略部分代码...
    public synchronized void xaCommit(String xid, long branchId, String applicationData) throws XAException {
        XAXid xaXid = XAXidBuilder.build(xid, branchId);
        xaResource.commit(xaXid, false);
        releaseIfNecessary();
    }
    
    
    public synchronized void xaRollback(String xid, long branchId, String applicationData) throws XAException {
        if (this.xaBranchXid != null) {
            xaRollback(xaBranchXid);
        } else {
            XAXid xaXid = XAXidBuilder.build(xid, branchId);
            xaRollback(xaXid);
        }
    }
     ...省略部分代码...
}


来源：https://www.cnblogs.com/Carey-ccl/p/19784564
回复

使用道具 举报

返回列表 发新帖
高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

Archiver|手机版|小黑屋|琼殿技术论坛 |nimba_sitemap:appname

相关侵权、举报、投诉及建议等，请发 E-mail：qiongdian@foxmail.com

Powered by Discuz! X5.0 © 2001-2026 Discuz! Team.

在本版发帖返回顶部