文档章节

分布式事务系列(3.2)jotm分布式事务源码分析

乒乓狂魔
 乒乓狂魔
发布于 2015/05/28 07:12
字数 4016
阅读 3569
收藏 98

#1 系列目录

#2 了解xapool

我们在前一篇文章中了解到jotm配合xapool共同完成了分布式事务。jotm主要提供了事务管理器TransactionManager的功能。而xapool则通过使用非XA数据库驱动实现了XA数据库驱动的效果。深入了解xapool之前,我们需要认识下XA数据库驱动到底是什么

##2.1 XA数据库驱动

###2.1.1 DataSource和ConnectionPoolDataSource

先来看下javax.sql.DataSource接口内容:

public interface DataSource  extends CommonDataSource,Wrapper {
  Connection getConnection() throws SQLException;
  Connection getConnection(String username, String password)
    throws SQLException;
}

DataSource就是提供Connection的。再来看下ConnectionPoolDataSource接口内容

public interface ConnectionPoolDataSource  extends CommonDataSource {
  PooledConnection getPooledConnection() throws SQLException;
  PooledConnection getPooledConnection(String user, String password)
    throws SQLException;
 }

ConnectionPoolDataSource就是提供PooledConnection,它代表了一个与数据库的物理连接,接口如下:

    public interface PooledConnection {
	   Connection getConnection() throws SQLException;
      }

这个物理连接能够产生Connection。Connection作为PooledConnection的一个handle,即PooledConnection可以产生多个Connection来供使用。总结如下:

ConnectionPoolDataSource-》PooledConnection-》Connection

DataSource-》Connection

更多详细内容见JDBC分布式事务浅析

总之需要带着怀疑的眼光去看待问题,以上看法不一定对。如果想深入探究,就需要各位去仔细去斟酌。

###2.1.2 XADataSource

再来看下javax.sql.XADataSource的接口内容:

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

XADataSource就是提供XAConnection的,什么是XAConnection呢?

    public interface XAConnection extends PooledConnection {
        javax.transaction.xa.XAResource getXAResource();
        }

可以看到XAConnection不仅能够获取Connection(PooledConnection的功能),还能获取与数据库通信的一个代表XAResource(上上一篇文章已说明,见AP、TM、RM三大对象

有了这个通信代表,我们就可以与数据库进行交互,实现两阶段提交协议。

###2.1.3 什么是XA数据库驱动

即实现了XADataSource接口的数据库驱动,它能够为我们创建XAConnection,有了XAConnection我们既能获取普通常见的Connection,又能获取XAResource,实现与数据库的交互,进而可以实现两阶段提交协议。

所以XA数据库驱动的最重要核心就是:有了XAResource的实现,能与数据库进行双向交互

目前大部分数据库都是支持XA驱动的,如

  • 对mysql来说,我们所用的mysql-connector-java jar包,就是支持的,它所提供的XADataSource实现是: com.mysql.jdbc.jdbc2.optional.MysqlXADataSource
  • 对oracle来说,它所提供的XADataSource实现是: oracle.jdbc.xa.client.OracleXADataSource

##2.2 xapool的内容

XA数据库驱动按道理上讲,应该是数据库来负责提供的。xapool就是在数据库驱动不是XA驱动的前提下,模拟了XA驱动。这种模拟情况下,与数据库的交互仍然是单向的。

###2.2.1 xapool的整体功能

  • Generic Pool:一个pool的功能,用来存储Object
  • StandardDataSource和StandardXADataSource,分别用于产生Connection和XAConnection
  • StandardPoolDataSource和StandardXAPoolDataSource 对上述那一对dataSource都加上了Generic Pool的功能,使用pool来维护和管理Connection(实际上是PooledConnection,后文会说到)和XAConnection

###2.2.2 认识GenericPool

它实现了一个pool的功能,和其他的pool的大体功能都差不多,它可以存储和管理任何对象。只要你实现了相应的接口,这里即PoolHelper接口,看下GenericPool的构造函数:

public GenericPool(PoolHelper helper) {
	this(
		helper,
		DEFAULT_MINSIZE,
		DEFAULT_MAXSIZE,
		DEFAULT_EXPIRATION,
		DEFAULT_SLEEPTIME);
}

GenericPool需要一个PoolHelper和一些参数设置(pool中object个数的最大值、最小值等)。来看下PoolHelper:

public interface PoolHelper {
	public void expire(Object o); // object specific work to kill the object
	public boolean checkThisObject(Object o);
	// check if the object is still valid
	public boolean testThisObject(Object o); // check if the object is closed
	public GenerationObject create() throws SQLException;
	public GenerationObject create(String _user, String _password)
		throws SQLException;
	public String toString();
}

可以看到GenericPool通过PoolHelper对外暴漏出一些方法,通过PoolHelper的create方法将创建的object存放至pool中,通过PoolHelper的expire方法将pool中的object真正意义上的删除掉(而不仅仅是从pool中移除)。

###2.2.3 StandardDataSource和StandardXADataSource

StandardDataSource需要实现DataSource的功能,即能够获取Connection。这个实现还是比较简单的,就是利用jdbc原始方式DriverManager来获取,如下:

Connection conn=DriverManager.getConnection(url, prop)

StandardXADataSource需要实现XADataSource的功能,即能够获取XAConnection,同时通过XAConnection能获取XAResource。

这里的XAResource留到和jotm的事务管理器一起来说。

###2.2.4 StandardPoolDataSource和StandardXAPoolDataSource

StandardPoolDataSource:内部拥有一个GenericPool,它自己实现了PoolHelper,我们来看看它把什么对象交给pool来管理了,即看看它怎么实现PoolHelper的create方法的:

StandardPoolDataSource向pool中存放的对象

它内部拥有一个ConnectionPoolDataSource对象,即图片中cpds对象,它能够产生PooledConnection(上文已说过,可以回去查看)。

可以看到向pool中存放的对象,并不是Connection,而是PooledConnection,上文说过,它是与数据库的一个物理连接,它能够产生Connection。

总结一下就是:StandardPoolDataSource利用内部ConnectionPoolDataSource对象来创建PooledConnection,然后把创建的PooledConnection交给内部的GenericPool来维护和管理。

StandardXAPoolDataSource:内部拥有一个GenericPool,它自己实现了PoolHelper,我们来看看它把什么对象交给pool来管理了,即看看它怎么实现PoolHelper的create方法的:

StandardXAPoolDataSource向pool中存放的对象

它内部拥有一个XADataSource对象,即图片中的xads,它能够产生XAConnection(上文已说过,可以回去查看)。

可以看到向pool中存放的对象是XAConnection。

总结一下就是:StandardXAPoolDataSource利用内部的XADataSource对象来创建XAConnection,然后把创建的XAConnection交给内部的GenericPool来维护和管理。

#3 jotm的事务管理器

##3.1 jotm要实现的JTA接口

jotm需要实现的JTA定义的接口有:

  • javax.transaction.UserTransaction:面向开发者的使用接口(jotm的实现是org.objectweb.jotm.Current)
  • javax.transaction.TransactionManager:事务管理器接口(jotm的实现是org.objectweb.jotm.Current)
  • javax.transaction.Transaction:事务接口(jotm的实现是org.objectweb.jotm.TransactionImpl)
  • javax.transaction.xa.Xid:全局事务的唯一标示(jotm的实现是org.objectweb.jotm.XidImpl)

##3.2 UserTransaction开启一个事务

即org.objectweb.jotm.Current实现begin方法的源码内容:

  • 第一步:创建一个全局的唯一标示Xid

    XidImpl otid = new XidImpl();
    
  • 第二步:根据唯一标示和超时设置创建一个事务

    TransactionImpl tx = new TransactionImpl(otid, transactionTimeout)
    
  • 第三步:对上述事务设定一个定时器

    tx.setTimer(timermgr.addTimer(tx, transactionTimeout, null, false));
    
  • 第四步:把上述事务和Xid关系绑定到当前线程

    threadTx.set(tx);
    txXids.put(xid, tx);
    

    org.objectweb.jotm.Current有这三个线程绑定数据:

    private static transient ThreadLocal<TransactionImpl> threadTx = new ThreadLocal<TransactionImpl>();
    private static transient ThreadLocal<Integer> threadTimeout = new ThreadLocal<Integer>();
    private transient static Map<Xid, TransactionImpl> txXids = new HashMap<Xid, TransactionImpl>();
    

##3.3 TransactionImpl与XAResource的交互

在我们创建完成事务后,开始使用业务逻辑操作时即上一篇的工程项目中使用JdbcTemplate时,就会执行TransactionImpl与XAResource的交互,即javax.transaction.Transaction接口方法enlistResource(XAResource xaRes):即把与资源管理器的通信代表XAResource加入到当前事务中来,在看TransactionImpl是如何来实现的之前,先了解下TransactionImpl中SubCoordinator的内部结构

###3.3.1 TransactionImpl中SubCoordinator的内部结构

TransactionImpl内部有一个SubCoordinator,这就是一个两阶段提交中的协调者:

SubCoordinator内部含有2个集合:

private Vector resourceList = new Vector();
private Vector javaxxidList = new Vector();

一个是存放XAResource的,另一个是存放对应XAResource的xid。这样的做法有点难以接受,他们仅仅靠位置来对应XAResource和Xid的关系。这里说明下,每个事务都会有一个xid进行标示,每个XAResource都会有另一个xid来标示。

SubCoordinator实现了一个接口org.objectweb.jotm.Resource,接口定义如下:

public interface Resource extends Remote {

    /** 
     * phase 1 of the 2PC.
     *
     * [@return](http://my.oschina.net/u/556800) int vote commit, rollback, or readonly.
     */
    public int prepare() throws RemoteException;
    public final static int VOTE_COMMIT = 0;
    public final static int VOTE_ROLLBACK = 1;
    public final static int VOTE_READONLY = 2;

    /** 
     * rollback transaction
     */
    public void rollback() throws RemoteException;

    /** 
     * phase 2 of the 2PC.
     */
    public void commit() throws RemoteException;

    /** 
     * commit 1 phase.
     */
    public void commit_one_phase() throws RemoteException;

    /** 
     * forget heuristics about this transaction.
     */
    public void forget() throws RemoteException;
}

可以清楚的看到这个org.objectweb.jotm.Resource接口就是针对两阶段提交来定义的。先有prepare方法进行投票,根据返回值结果(上述三个结果VOTE_COMMIT、VOTE_ROLLBACK、VOTE_READONLY)来选择下一步是执行commit还是rollback还是forget(当返回结果为VOTE_READONLY只读时,执行forget,即忽略这个事务)。

来看下SubCoordinator是如何来实现这个接口的:

接口方法int prepare():

  • 第一步:先将所有的XAResource结束事务边界,即调用XAResource的end方法,然后从事务TransactionImpl的enlistedXARes列表中移至delistedXARes

    事务TransactionImpl也有如下三个数据:

    private List<XAResource> enlistedXARes = Collections.synchronizedList(new ArrayList<XAResource>());
    
    private List<XAResource> delistedXARes = null;
    
    private List<javax.transaction.xa.Xid> enlistedJavaxXid =
    Collections.synchronizedList(new ArrayList<javax.transaction.xa.Xid>());
    

    也就是说事务中保存了一份XAResource,事务中的SubCoordinator也保存了一份

    整个过程代码如下:

    //先将enlistedXARes全部移至delistedXARes
    delistedXARes = new ArrayList<XAResource>(enlistedXARes);
    for (XAResource xar : delistedXARes) {
        delistResource(xar, flag);
    }
    
    public boolean delistResource(XAResource xares, int flag){
    	//结束事务边界
    	xares.end (myjavaxxid, flag);
    
    	//从enlistedXARes、enlistedJavaxXid列表中移除
    	enlistedXARes.remove(xares);
    	enlistedJavaxXid.remove(javaxxid);
    	//略
    }
    
  • 第二步:开始准备预提交,遍历SubCoordinator中所有的XAResource,依次进行预提交即执行每个XAResource的prepare方法(简化了部分内容)

    ret=VOTE_READONLY;
    int errors = 0;
    for (int i = 0; i < resourceList.size(); i++) {
        XAResource res = (XAResource) resourceList.elementAt(i);
    	javax.transaction.xa.Xid myjavaxxid = (javax.transaction.xa.Xid) javaxxidList.elementAt(i);
    	if (errors > 0) {
    		res.rollback(myjavaxxid);
    	}else{
    		try {
    			switch (res.prepare(myjavaxxid)) {
                        case XAResource.XA_OK :
                            ret = VOTE_COMMIT;
                            break;
                        case XAResource.XA_RDONLY :
                            break;
                        }
            } catch (XAException e) {
                 ret = VOTE_ROLLBACK;
    			 errors++
            }
    	}
    }
    return ret;
    

这里还是同步执行,效率肯定很慢。

接口方法void rollback():

  • 第一步:同上第一步,先将所有的XAResource结束事务边界,然后从事务TransactionImpl的enlistedXARes列表中移至delistedXARes
  • 第二步:遍历SubCoordinator中所有的XAResource,依次执行每个XAResource的rollback方法。

接口方法void commit():也是类似,将XAResource依次执行commit操作。

###3.3.2 TransactionImpl的enlistResource操作

即把与资源管理器的通信代表XAResource加入到当前事务中来,enlistResource(XAResource xares)含有如下操作:

  • 第一步:添加到TransactionImpl中的SubCoordinator中

    subcoord.addResource(xares);
    
  • 第二步:创建该XAResource对应的xid

    Xid resXid = new XidImpl( getXid(),subcoord.getXaresIndex(xares) );
    javax.transaction.xa.Xid javaxxid = new JavaXidImpl(resXid);
    
  • 第三步:把该xid加入进SubCoordinator中

    subcoord.addJavaxXid(javaxxid);
    
  • 第四步:开始XAResource的事务边界

    xares.start (javaxxid, flag);
    
  • 第五步:把上述XAResource和xid也存放到事务TransactionImpl中

    enlistedXARes.add(xares);
    enlistedJavaxXid.add(javaxxid);
    

TransactionImpl的delistResource操作就不再说明了。详情自己去看源码。

##3.4 UserTransaction提交或回滚一个事务

这个就不再详细说明了。

  • 第一步:首先从当前线程中获取绑定的事务即TransactionImpl,依托该事务进行回滚或者提交,TransactionImpl会依托内部的协调者SubCoordinator来进行提交或者回滚
  • 第二步:清空当前线程绑定的事务

#4 jotm和xapool的交互

##4.1 在业务逻辑操作中XAResource加入进事务

在执行业务逻辑的时候即如下:

jdbcTemplate.update("insert into user(name,age) values(?,?)",user.getName(),user.getAge());

再对照着配置文件来看:

<bean id="dataSourceA" class="org.enhydra.jdbc.pool.StandardXAPoolDataSource"  destroy-method="shutdown">
 	<property name="dataSource">  
        <bean class="org.enhydra.jdbc.standard.StandardXADataSource" destroy-method="shutdown">  
            <property name="transactionManager" ref="jotm" />  
            <property name="driverName" value="com.mysql.jdbc.Driver" />  
            <property name="url" value="jdbc:mysql://localhost:3306/test?useUnicode=true&amp;characterEncoding=utf-8" />  
        </bean>  
    </property>     
    <property name="user" value="root" />  
    <property name="password" value="ligang" />  
</bean>

这时候jdbcTemplate会从dataSource中获取一个Connection来执行sql,即从StandardXAPoolDataSource中获取Connection。具体步骤如下:

  • 第一步:StandardXAPoolDataSource发现它内部的pool还没有初始化,启动初始化。初始化会创建指定数量的的对象,StandardXAPoolDataSource在pool中存放和维护的对象是StandardXAConnection。而StandardXAConnection的创建委托给了由内部的StandardXADataSource来进行创建。初始化过后,pool中就存在了指定数量的StandardXAConnection对象

  • 第二步:获取Connection,会先从pool中获取一个StandardXAConnection,该对象再进行包装,封装成一个StandardXAConnectionHandle,返回给用户,所以用户使用的Connection是StandardXAConnectionHandle类型的

  • 第三步:使用StandardXAConnectionHandle来执行sql操作前,会进行事务的判断,如下:

    StandardXAConnectionHandle的事务逻辑

    先获取和当前线程绑定的事务,它是通过上述配置文件中配置的transactionManager来获取的,如下:

    Transaction ntx = transactionManager.getTransaction();
    

    如果存在事务,则设置当前自动提交为false。因为StandardXAConnection底层还是使用的是普通的Connection来完成的,所以此操作就是设置一个普通的Connection的自动提交为false。

    将XAResource加入到当前事务中。这部分内容上文已经说了。

这个XAResource是什么类型呢?就是StandardXAConnection,它实现了XAResource:

public class StandardXAConnection extends StandardPooledConnection implements XAConnection, XAResource

也就是说事务TransactionImpl的提交和回滚和prepare操作依赖于TransactionImpl内部的SubCoordinator,而SubCoordinator又会依次调用每个XAResource(这里即StandardXAConnection)的提交和回滚和prepare操作。而StandardXAConnection内部只有一个普通的Connection,所以StandardXAConnection要利用普通的Connection来模拟XAResource的操作。具体怎么模拟的,这里就不再介绍了,详情去看源代码。

##4.2 案例总结

针对上一篇文章介绍的工程项目,简单总结下:

@Transactional
public void save(User user){
	userDao.save(user);
	logDao.save(user);
	throw new RuntimeException();
}
  • 第一步:有了@Transactional注解,会把创建出代理对象,加入事务拦截器,在执行save方法之前,会先使用UserTransaction开启一个事务即TransactionImpl,并创建一个xid,进行标示,然后把该事务绑定到当前线程。

  • 第二步:使用userDao执行业务逻辑时,即使用JdbcTemplate操作时,会从StandardXAPoolDataSource中获取一个Connection。

    首选会初始化StandardXAPoolDataSource中的pool,创建出指定数量的StandardXAConnection

    然后再从pool中获取一个StandardXAConnection,StandardXAConnection又进行再次包装成StandardXAConnectionHandle,返回给用户作为Connection。

  • 第三步:使用Connection(这里即StandardXAConnectionHandle)执行sql前,会根据事务管理器获取当前线程绑定的事务,如果有,则设置StandardXAConnectionHandle的自动提交为false,最终是设置到普通的Connection上了。并且把XAResource加入到当前事务中,即把StandardXAConnection加入到TransactionImpl中,同时开启XAResource的事务边界,即调用start方法

  • 第四步:一旦执行过程发生异常,spring的PlatformTransactionManager这里即JtaTransactionManager,会获取UserTransaction(这里即org.objectweb.jotm.Current)进行回滚操作,它会获取当前此线程绑定的事务TransactionImpl进行回滚,TransactionImpl会委托到内部的协调者SubCoordinator,SubCoordinator会调用每个加入进来的XAResource(这里即StandardXAConnection)执行回滚操作,StandardXAConnection则会依托内部的普通Connection进行回滚操作。

##4.3 遗留的问题

可以看到上述过程,事务TransactionImpl内部的协调者SubCoordinator虽然实现了两阶段提交过程的代码,但是在上述案例中并没有体现出来,也就是没有去调用过prepare过程。

上述过程SubCoordinator仅仅起到了一个收集Connection的作用,首选把所有的Connection的自动提交设置为false,执行业务操作,一旦发现异常,则执行每个Connection的rollback操作。

所以到底怎么去使用两阶段提交模式呢?还需要去仔细去研究研究这两方面的内容:

  • javax.resource.spi.XATerminator接口和org.objectweb.jotm.XATerminatorImpl实现,该接口也是javax针对两阶段提交协议定义的接口,和jotm中定义的org.objectweb.jotm.Resource差不多。实现其实还是基于SubCoordinator来执行的,这在什么情况下使用呢?

  • SubCoordinator的源代码,它还继承了一个远程调用的PortableRemoteObject。有兴趣的可以去研究下

    /**
     * This object is the local coordinator. It may be registered as
     * sub-coordinator in case of distributed transaction, so it must
     * be callable remotely and implement Resource
     */
    public class SubCoordinator extends PortableRemoteObject implements Resource
    

#5 结束语 本篇文章主要介绍了使用jotm和xapool实现分布式事务的原理。下一篇就开始介绍atomikos对于分布式事务的支持。同样先给出例子:

  • atomikos使用非XA数据库驱动实现分布式事务的例子
  • atomikos使用XA数据库驱动实现分布式事务的例子

© 著作权归作者所有

共有 人打赏支持
乒乓狂魔
粉丝 1011
博文 105
码字总数 271356
作品 0
长宁
程序员
私信 提问
加载中

评论(2)

乒乓狂魔
乒乓狂魔

引用来自“hello_study”的评论

用spring的AbstractRoutingDataSource动态数据源进行切换时发现加了事务后不能切换数据库连接了,不知道有没有好的解决办法
你具体是怎么做的呢?动态数据源的切换和事务都是通过注解来实现AOP代理的吗?是不是事务代理执行了,而动态数据源代理就没有执行?他们的执行顺序是不是动态数据源代理先执行?具体的可以参考这篇文章 http://linhongyu.blog.51cto.com/6373370/1615895
h
hello_study
用spring的AbstractRoutingDataSource动态数据源进行切换时发现加了事务后不能切换数据库连接了,不知道有没有好的解决办法
分布式事务系列(开篇)提出疑问和研究过程

1 前言 系列目录 - 分布式事务系列(开篇)提出疑问和研究过程- 分布式事务系列(1.1)Spring事务管理器PlatformTransactionManager源码分析- 分布式事务系列(1.2)Spring事务体系- 分布式事...

乒乓狂魔
2015/05/12
0
6
分布式事务系列(4.1)Atomikos的分布式案例

1 系列目录 - 分布式事务系列(开篇)提出疑问和研究过程- 分布式事务系列(1.1)Spring事务管理器PlatformTransactionManager源码分析- 分布式事务系列(1.2)Spring事务体系- 分布式事务系...

乒乓狂魔
2015/06/01
0
3
分布式事务系列(3.1)jotm的分布式案例

1 系列目录 - 分布式事务系列(开篇)提出疑问和研究过程- 分布式事务系列(1.1)Spring事务管理器PlatformTransactionManager源码分析- 分布式事务系列(1.2)Spring事务体系- 分布式事务系...

乒乓狂魔
2015/05/25
0
5
【分布式事务系列】提出疑问和研究过程

分布式事务 【分布式事务系列】提出疑问和研究过程 对于我们这种初学者,可能会使用Spring带给我们的@Transactional,可能了解JTA,可能会使用jotm、atomikos,又会遇到一些名词XA,支持XA的...

陶邦仁
2015/12/09
418
0
分布式事务系列(2.1)分布式事务的概念

1 系列目录 - 分布式事务系列(开篇)提出疑问和研究过程- 分布式事务系列(1.1)Spring事务管理器PlatformTransactionManager源码分析- 分布式事务系列(1.2)Spring事务体系- 分布式事务系...

乒乓狂魔
2015/05/21
0
7

没有更多内容

加载失败,请刷新页面

加载更多

[LintCode] Serialize and Deserialize Binary Tree(二叉树的序列化和反序列化)

描述 设计一个算法,并编写代码来序列化和反序列化二叉树。将树写入一个文件被称为“序列化”,读取文件后重建同样的二叉树被称为“反序列化”。 如何反序列化或序列化二叉树是没有限制的,你...

honeymose
今天
5
0
java框架学习日志-7(静态代理和JDK代理)

静态代理 我们平时去餐厅吃饭,不是直接告诉厨师做什么菜的,而是先告诉服务员点什么菜,然后由服务员传到给厨师,相当于服务员是厨师的代理,我们通过代理让厨师炒菜,这就是代理模式。代理...

白话
今天
23
0
Flink Window

1.Flink窗口 Window Assigner分配器。 窗口可以是时间驱动的(Time Window,例如:每30秒钟),也可以是数据驱动的(Count Window,例如:每一百个元素)。 一种经典的窗口分类可以分成: 翻...

满小茂
今天
18
0
my.ini

1

architect刘源源
今天
16
0
docker dns

There is a opensource application that solves this issue, it's called DNS Proxy Server It's a DNS server that solves containers hostnames, if could not found a hostname that mat......

kut
今天
17
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部