Java 分布式事务（多数据源）

旧城等待， 2022-09-24 11:18 136阅读 0赞

最近这几天一直在整 怎么实现分布式事务。找了很多资料，不过大都相近类同。对Oracle、SQL Server、Mysql数已做过测试，其中Mysql5.0以上的才支持分布式事务。  
  
对于这些，主要是之前根本没有接触过分布式事务，还错找了一些分布式事数据库的资料，呵呵，结果不是我目前所需要的。   
测试过程中出现了很多错误，一直都通不过，以为是用户权限还有数据库服务的问题，但一切都配置良好的情况下还一直都通不过。结果发现，我导入的都是一些普通的JDBC连接包，于是狂搜实现XA事务的jar包。  
  
Mysql:            mysql-connector-java-5.1.6-bin.jar   
SQL Server:       sqljdbc.jar   
Oracle:           ojdbc14.jar   
用的是这些包才顺利通过运行。后面会附上这几个jar包。   
好了，把 [源代码][Link 1]也附上：

import com.microsoft.sqlserver.jdbc.SQLServerXADataSource;   
import com.mysql.jdbc.jdbc2.optional.MysqlXADataSource;   
import java.sql.\*;   
import java.util.logging.Level;   
import java.util.logging.Logger;   
import javax.sql.\*;   
import javax.transaction.xa.\*;   
import oracle.jdbc.xa.client.OracleXADataSource;

public class Mutil\_DataSource\_Test \{   
    public static void main(String\[\] args)\{   
        Mutil\_DataSource\_Test mdt = new Mutil\_DataSource\_Test();   
        try \{   
            mdt.test1();   
        \} catch (Exception ex) \{   
            System.out.println("除SQLException、XAException之外的异常: \\n");   
            Logger.getLogger(Mutil\_DataSource\_Test.class.getName()).log(Level.SEVERE, null, ex);  
  
        \}   
    \}

class MyXid implements Xid\{   
        int formatId;   
        byte globalTransactionId\[\];   
        byte branchQualifier\[\];   
        public MyXid()\{

\}   
        public MyXid(int formatId,byte\[\] globalTransactionId,byte\[\] branchQualifier)\{  
  
            this.formatId = formatId;   
            this.globalTransactionId = globalTransactionId;   
            this.branchQualifier = branchQualifier;   
        \}

public int getFormatId() \{   
            return this.formatId;   
        \}   
        public void setFormatId(int formatId)\{   
            this.formatId = formatId;   
        \}   
        public byte\[\] getGlobalTransactionId() \{   
            return this.globalTransactionId;   
        \}   
        public void setGlobalTransactionId(byte\[\] globalTransactionId)\{   
            this.globalTransactionId = globalTransactionId;   
        \}   
        public byte\[\] getBranchQualifier() \{   
            return this.branchQualifier;   
        \}   
        public void setBranchQualifier(byte\[\] branchQualifier)\{   
            this.branchQualifier = branchQualifier;   
        \}   
    \}

//多数据库测试   
    public void test1() \{   
        //定义所需用到的变量   
         Connection mysqlCn = null;   
        Connection sqlCn = null;   
        Connection mysqlCn2 = null;   
        Connection oraCn = null;

MysqlXADataSource mysqlDs = null;   
        SQLServerXADataSource sqlDs = null;   
        MysqlXADataSource mysqlDs2 = null;   
        OracleXADataSource oraDs = null;

XAConnection xamysqlCn = null;   
        XAConnection xasqlCn = null;   
        XAConnection xamysqlCn2 = null;   
        XAConnection xaoraCn = null;

XAResource xamysqlRes = null;   
        XAResource xasqlRes = null;   
        XAResource xamysqlRes2 = null;   
        XAResource xaoraRes = null;

Xid mysqlXid = null;   
        Xid sqlXid = null;   
        Xid mysqlXid2 = null;   
        Xid oraXid = null;

Statement mysqlpst = null;   
        Statement sqlpst = null;   
        Statement mysqlpst2 = null;   
        Statement orapst = null;   
    try\{   
        //获得数据源   
        mysqlDs = new MysqlXADataSource();   
        mysqlDs.setURL("jdbc:mysql://localhost:3306/test");   
        mysqlDs2 = new MysqlXADataSource();   
        mysqlDs2.setURL("jdbc:mysql://10.10.10.119:3306/test");   
        sqlDs = new SQLServerXADataSource();   
        sqlDs.setURL("jdbc:sqlserver://10.10.10.119:1433;DatabaseName=RTC;loginTimeout=20;user=sa;password=chgpwd122105");  
  
//        sqlDs.setUser("sa");   
//        sqlDs.setPassword("chgpwd122105");   
//        sqlDs.setServerName("10.10.10.119");   
//        sqlDs.setPortNumber(1433);   
//        sqlDs.setDatabaseName("RTC");   
        oraDs = new OracleXADataSource();   
        oraDs.setURL("jdbc:oracle:thin:@10.10.10.119:1521:WMS");   
        //获得连接   
        xamysqlCn = mysqlDs.getXAConnection("root", "9999");   
System.out.println("xamysqlCn: "+xamysqlCn);   
        xasqlCn = sqlDs.getXAConnection();   
System.out.println("xasqlCn: "+xasqlCn);   
        xamysqlCn2 = mysqlDs2.getXAConnection("root", "9999");   
System.out.println("xamysqlCn2: "+xamysqlCn2);   
        xaoraCn = oraDs.getXAConnection("tiger", "tiger");   
System.out.println("xaoraCn: "+xaoraCn);

mysqlCn = xamysqlCn.getConnection();   
        sqlCn = xasqlCn.getConnection();   
        mysqlCn2 = xamysqlCn2.getConnection();   
        oraCn = xaoraCn.getConnection();

mysqlpst = mysqlCn.createStatement();   
        sqlpst = sqlCn.createStatement();   
        mysqlpst2 = mysqlCn2.createStatement();   
        orapst = oraCn.createStatement();   
        //定义XAResource   
        xamysqlRes = xamysqlCn.getXAResource();   
        xasqlRes = xasqlCn.getXAResource();   
        xamysqlRes2 = xamysqlCn2.getXAResource();   
        xaoraRes = xaoraCn.getXAResource();   
        //定义Xid   
        mysqlXid = new MyXid(0, new byte\[\]\{0x01\}, new byte\[\]\{0x02\});   
        sqlXid = new MyXid(0, new byte\[\]\{0x01\}, new byte\[\]\{0x03\});   
        mysqlXid2 = new MyXid(0, new byte\[\]\{0x01\}, new byte\[\]\{0x04\});   
        oraXid = new MyXid(0, new byte\[\]\{0x01\}, new byte\[\]\{0x05\});   
        //执行Mysql   
        xamysqlRes.start(mysqlXid, XAResource.TMNOFLAGS);          
        mysqlpst.executeUpdate("insert into test values(4,'XA','F','Class4')");   
        xamysqlRes.end(mysqlXid, XAResource.TMSUCCESS);   
        //执行SQLServer   
        xasqlRes.start(sqlXid, XAResource.TMNOFLAGS);   
        sqlpst.executeUpdate("insert into test values('444')");   
        xasqlRes.end(sqlXid, XAResource.TMSUCCESS);   
        //执行Mysql   
        xamysqlRes2.start(mysqlXid2, XAResource.TMNOFLAGS);   
        mysqlpst2.executeUpdate("insert into test values(4,'XA','F','Class4')");   
        xamysqlRes2.end(mysqlXid2, XAResource.TMSUCCESS);   
        //执行Oracle   
        System.out.println("xaoraRes: "+xaoraRes);   
        xaoraRes.start(oraXid, XAResource.TMNOFLAGS);   
        orapst.executeUpdate("insert into test123 values('4','44','444')");   
        System.out.println("oraXid: "+oraXid);   
        xaoraRes.end(oraXid, XAResource.TMSUCCESS);   
        //准备   
        int mysqlRea = xamysqlRes.prepare(mysqlXid);   
        int sqlRea = xasqlRes.prepare(sqlXid);   
        int mysqlRea2 = xamysqlRes2.prepare(mysqlXid2);   
        int oraRea = xaoraRes.prepare(oraXid);   
        //判断准备就绪与否  提交或回滚   
        if(mysqlRea == xamysqlRes.XA\_OK && mysqlRea2 == xamysqlRes.XA\_OK && oraRea == xaoraRes.XA\_OK && sqlRea == xasqlRes.XA\_OK)\{  
  
//        if(mysqlRea == xamysqlRes.XA\_OK && mysqlRea2 == xamysqlRes.XA\_OK && oraRea == xaoraRes.XA\_OK)\{  
  
//        if(mysqlRea == xamysqlRes.XA\_OK && sqlRea == xasqlRes.XA\_OK && mysqlRea2 == xamysqlRes.XA\_OK)\{  
  
//        if(mysqlRea == xamysqlRes.XA\_OK && mysqlRea2 == xamysqlRes.XA\_OK)\{   
            xamysqlRes.commit(mysqlXid, false);   
System.out.println("Mysql 事务提交成功！");   
            xasqlRes.commit(sqlXid, false);   
System.out.println("SQLServer 事务提交成功！");   
            xamysqlRes2.commit(mysqlXid2, false);   
System.out.println("Mysql2 事务提交成功！");   
            xaoraRes.commit(oraXid, false);   
System.out.println("Oracle 事务提交成功！");   
        \}else\{   
            xamysqlRes.rollback(mysqlXid);   
            xasqlRes.rollback(sqlXid);   
            xamysqlRes2.rollback(mysqlXid2);   
            xaoraRes.rollback(oraXid);   
System.out.println("事务回滚成功！");   
        \}   
    \}catch(SQLException ex)\{   
        Logger.getLogger(Mutil\_DataSource\_Test.class.getName()).log(Level.SEVERE, null, ex);  
  
        try\{   
            xamysqlRes.rollback(mysqlXid);   
            xasqlRes.rollback(sqlXid);   
            xamysqlRes2.rollback(mysqlXid2);   
            xaoraRes.rollback(oraXid);   
        \}catch(XAException e)\{   
            System.out.println("回滚也出错咯！~");   
            e.printStackTrace();   
        \}   
    \}catch(XAException ex)\{   
        Logger.getLogger(Mutil\_DataSource\_Test.class.getName()).log(Level.SEVERE, null, ex);  
  
    \}finally\{   
        try \{   
        //关闭   
        mysqlpst.close();   
        mysqlCn.close();   
        xamysqlCn.close();   
        sqlpst.close();   
        sqlCn.close();   
        xasqlCn.close();   
        mysqlpst2.close();   
        mysqlCn2.close();   
        xamysqlCn2.close();   
        orapst.close();   
        oraCn.close();   
        xaoraCn.close();   
        \} catch (SQLException ex) \{   
            Logger.getLogger(Mutil\_DataSource\_Test.class.getName()).log(Level.SEVERE, null, ex);  
  
        \}   
    \}   
    \}   
\}

分布式事务分为两个阶段，第一个阶段相当于是一个预提交，第二阶段才是正真的提交。   
首先要实现的是Xid接口，formatId可以理解为一个全局事务的ID，不过我上面的[代码][Link 1]没有去做一些异常的处理，还有正确的链接的关闭，只是自己做的一个小小的测试，以后项目中再去处理这些。第一阶段如果出错或是不顺利则不会提交，需要做一些回滚处理。如果顺利则准备提交，进入第二阶段，在第二阶段可能会出现问题，如果第一个分支事务提交成功了，而后有一个分支事务提交失败，这样则会造成数据不准确，目前还不知道有没有方法可以解决些问题，好像XAResource.recover()可以处理，但具体怎么解决，在项目中自己还是可以想到办法解决。  
  
对XA分布式事务  我是初学者，而且也没有谁指点，只是在网上找些资料。我上面的理解可能也还有些不足或是错误。希望高手看到后能指点指点。

*  [XA\_Jar包.rar][XA_Jar_.rar] (2.6 MB)
 *  下载次数: 106

[Link 1]: http://www.xuebuyuan.com/
[XA_Jar_.rar]: http://dl.iteye.com/topics/download/5370bd9a-87c0-3f23-8615-c0643434b307