分布式事务系列（一）XA

怼烎@ 2022-09-15 12:41 298阅读 0赞

## 1、简介 ##

### 什么是分布式事务？ ###

两个不同的服务器上，对两个不同的库的写操作，要求执行的原子性（要么都执行，要么都不执行）。

### XA协议 ###

X/Open组织定义的资源管理器（Resource Manager）和事务管理器（Transaction Manager）的接口标准。  
采用**两阶段提交2PC**方式来管理分布式事务。

*  资源管理器（RM）：管理系统资源，还具有管理事务提交或回滚的能力。数据库就是一种资源管理器。
 *  事务管理器（TM）：分布式事务的核心管理者，与RM进行通信，协调并完成事务的处理。  
    ![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAeXlzeA_size_20_color_FFFFFF_t_70_g_se_x_16]

### 2PC ###

存在问题：

1.  阻塞/性能问题
2.  单点故障

## 2、代码-模拟XA ##

假设：  
两个数据库实例，分别为RM1、RM2。

### Step 1——获取RM1、RM2 ###

Connection conn1= null;
      Connection conn2= null;
      try { 
            //获取资源管理器操作接口实例 RM1
            conn1 = DriverManager.getConnection(url,name,password);
            XAConnection xaConn1=new MysqlXAConnection(conn1,true);
            XAResource rm1=xaConn1.getXAResource();
    
            //获取资源管理器操作接口实例 RM2
            conn2 = DriverManager.getConnection(url,name,password);
            XAConnection xaConn2=new MysqlXAConnection(conn2,true);
            XAResource rm2=xaConn2.getXAResource();
      
       } catch (SQLException | XAException e) { 
            e.printStackTrace();
       }

### Step 2——Application请求TM ###

正式开始分布式事务部分，其中：  
全局事务id：global\_tx\_id\_123  
分支事务1id：branch\_tx\_id\_001  
分支事务2id：branch\_tx\_id\_002

//模拟全局事务id
    			byte[] grid="global_tx_id_123".getBytes();
                int formatId=1;
    
                //分别执行RM1-分支事务1和RM2-分支事务2
                //RM1的分支事务
                byte[] bqual1="branch_tx_id_001".getBytes();
                Xid xid1=new MysqlXid(grid,bqual1,formatId);
                rm1.start(xid1,XAResource.TMNOFLAGS);
                PreparedStatement p1=conn1.prepareStatement("insert into a_test values(1,'a')");
                p1.execute();
                rm1.end(xid1,XAResource.TMSUCCESS);
    
                //RM2的分支事务
                byte[] bqual2="branch_tx_id_002".getBytes();
                Xid xid2=new MysqlXid(grid,bqual2,formatId);
                rm2.start(xid2,XAResource.TMNOFLAGS);
                PreparedStatement p2=conn2.prepareStatement("insert into a_test values(2,'b')");
                p2.execute();
                rm2.end(xid2,XAResource.TMSUCCESS);

至此，两个分支事务已经分别执行，但并未提交。只是在各自的本地执行，undo log做了记录。

### Step 3——两阶段提交 ###

> 注意：  
> 第一阶段只有当两个RM都准备完毕，才会进行提交操作。否则，就要执行事务的回滚操作。

//两阶段提交
                //第一阶段-Prepare
                int rm1Prepare=rm1.prepare(xid1);
                int rm2Prepare=rm2.prepare(xid2);
                //第二阶段-Commit Or CallBack
                if(rm1Prepare==XAResource.XA_OK && rm2Prepare==XAResource.XA_OK){ 
                    rm1.commit(xid1,false);
                    rm2.commit(xid2,false);
                }else{ 
                    rm1.rollback(xid1);
                    rm2.rollback(xid2);
                }

## 3、源码关键类 ##

主要分为两部分：  
（1）Java对XA的支持  
主要在rt.jar的包javax下。

（2）Mysql对XA的支持。  
在mysql-connector-java.jar的com.mysql.jdbc。

### XADataSource ###

public interface XADataSource extends CommonDataSource { 
      XAConnection getXAConnection() throws SQLException;
      XAConnection getXAConnection(String user, String password)
        throws SQLException;
    }

### XAConnection ###

package javax.sql;
    import java.sql.*;
    public interface XAConnection extends PooledConnection { 
    
    
      /** * Retrieves an <code>XAResource</code> object that * the transaction manager will use * to manage this <code>XAConnection</code> object's participation in a * distributed transaction. * * @return the <code>XAResource</code> object * @exception SQLException if a database access error occurs * @exception SQLFeatureNotSupportedException if the JDBC driver does not support * this method * @since 1.4 */
      javax.transaction.xa.XAResource getXAResource() throws SQLException;
    
     }

### XAResource ###

该类在rt.jar的transaction包下，提供了事务操作的基本接口。如start、commit、rollback、prepare等。

package javax.transaction.xa;
    
    public interface XAResource { 
        int TMENDRSCAN = 8388608;
        int TMFAIL = 536870912;
        int TMJOIN = 2097152;
        int TMNOFLAGS = 0;
        int TMONEPHASE = 1073741824;
        int TMRESUME = 134217728;
        int TMSTARTRSCAN = 16777216;
        int TMSUCCESS = 67108864;
        int TMSUSPEND = 33554432;
        int XA_RDONLY = 3;
        int XA_OK = 0;
    
        void commit(Xid var1, boolean var2) throws XAException;
    
        void end(Xid var1, int var2) throws XAException;
    
        void forget(Xid var1) throws XAException;
    
        int getTransactionTimeout() throws XAException;
    
        boolean isSameRM(XAResource var1) throws XAException;
    
        int prepare(Xid var1) throws XAException;
    
        Xid[] recover(int var1) throws XAException;
    
        void rollback(Xid var1) throws XAException;
    
        boolean setTransactionTimeout(int var1) throws XAException;
    
        void start(Xid var1, int var2) throws XAException;
    }

### Xid和MysqlXid ###

事务id

public interface Xid { 
        int MAXGTRIDSIZE = 64;
        int MAXBQUALSIZE = 64;
    
        int getFormatId();
    
        byte[] getGlobalTransactionId();
    
        byte[] getBranchQualifier();
    }
    
    public class MysqlXid implements Xid { 
    	//分支限定符
    	byte[] myBqual;
    	//用于标记格式，默认为1
        int myFormatId;
    	//全局事务id
        byte[] myGtrid;
    
        public MysqlXid(byte[] gtrid, byte[] bqual, int formatId) { 
            this.myGtrid = gtrid;
            this.myBqual = bqual;
            this.myFormatId = formatId;
        }
         public byte[] getBranchQualifier() { 
            return this.myBqual;
        }
    
        public int getFormatId() { 
            return this.myFormatId;
        };
    
        public byte[] getGlobalTransactionId() { 
            return this.myGtrid;
        }
    }

### MysqlXAConnection ###

Mysql支持XA的主要实现类，在mysql-connector-java.jar的com.mysql.jdbc包下。

XA语法：

<table> 
 <thead> 
  <tr> 
   <th>语法</th> 
   <th>含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>XA {START/BEGIN} xid [JOIN/RESUME]</td> 
   <td>开启XA事务</td> 
  </tr> 
  <tr> 
   <td>XA PREPARE xid</td> 
   <td>2PC的prepare阶段</td> 
  </tr> 
  <tr> 
   <td>XA COMMIT xid [ONE PHASE]</td> 
   <td>2PC的commit阶段</td> 
  </tr> 
  <tr> 
   <td>XA ROLLBACK xid</td> 
   <td>回滚</td> 
  </tr> 
  <tr> 
   <td>XA RECOVER [CONVERT XID]</td> 
   <td>所有处理prepared状态的事务</td> 
  </tr> 
  <tr> 
   <td>XA END xid [SUSPEND [FOR MIGRATE]]</td> 
   <td>结束分支事务</td> 
  </tr> 
 </tbody> 
</table>

以start()为例，主要逻辑就是在拼接XA的开启事务语法、并执行。

public void start(Xid xid, int flags) throws XAException { 
            StringBuilder commandBuf = new StringBuilder(MAX_COMMAND_LENGTH);
            commandBuf.append("XA START ");
            appendXid(commandBuf, xid);
    
            switch (flags) { 
                case TMJOIN:
                    commandBuf.append(" JOIN");
                    break;
                case TMRESUME:
                    commandBuf.append(" RESUME");
                    break;
                case TMNOFLAGS:
                    // no-op
                    break;
                default:
                    throw new XAException(XAException.XAER_INVAL);
            }
    
            dispatchCommand(commandBuf.toString());
    		//设置连接为全局事务
            this.underlyingConnection.setInGlobalTx(true);
        }
        
        //执行命令，返回结果集
        private ResultSet dispatchCommand(String command) throws XAException { 
            Statement stmt = null;
    
            try { 
                if (this.logXaCommands) { 
                    this.log.logDebug("Executing XA statement: " + command);
                }
    
                // TODO: Cache this for lifetime of XAConnection
                stmt = this.underlyingConnection.createStatement();
    
                stmt.execute(command);
    
                ResultSet rs = stmt.getResultSet();
    
                return rs;
            } catch (SQLException sqlEx) { 
                throw mapXAExceptionFromSQLException(sqlEx);
            } finally { 
                if (stmt != null) { 
                    try { 
                        stmt.close();
                    } catch (SQLException sqlEx) { 
                    }
                }
            }
        }

其余方法类似。

public int prepare(Xid xid) throws XAException { 
            StringBuilder commandBuf = new StringBuilder(MAX_COMMAND_LENGTH);
            commandBuf.append("XA PREPARE ");
            appendXid(commandBuf, xid);
    
            dispatchCommand(commandBuf.toString());
    
            return XA_OK; // TODO: Check for read-only
        }
    
     public void commit(Xid xid, boolean onePhase) throws XAException { 
            StringBuilder commandBuf = new StringBuilder(MAX_COMMAND_LENGTH);
            commandBuf.append("XA COMMIT ");
            appendXid(commandBuf, xid);
    
            if (onePhase) { 
                commandBuf.append(" ONE PHASE");
            }
    
            try { 
                dispatchCommand(commandBuf.toString());
            } finally { 
                this.underlyingConnection.setInGlobalTx(false);
            }
        }

## 4、小结 ##

XA协议也有一些问题：  
（1）长期锁定数据  
数据在整个事务处理过程结束前，一直被锁定。  
（2）协议阻塞  
在第一阶段prepare后，分支事务进行阻塞阶段。只有收到coomit或者rollback指令，才会结束阻塞状态。  
（3）性能较差  
“事务协调”花费了大量的时间，并发事务数据的锁冲突。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAeXlzeA_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/637cbdfb932d4406b3ea255cc4a2ed13.png