SSM使用AbstractRoutingDataSource后究竟如何解决跨库事务

Setting:

　　绑定三个数据源(XA规范),将三个实例绑定到AbStractoutingDataSource的实例MultiDataSource(自定义的)对象中,mybatis SqlSessionFactory数据源设定为MultiDataSource,DataSourceTransactionManager数据源绑定MultiDataSource,自定义注解,切面,就某个字段被crud操作涉及时,进入切面,计算目标数据库,动态切换数据库.不涉及事务时实现了动态切换,加了事务后..........

Question:

　　1. AbstractRoutingDataSource是什么? 为什么它能动态切换数据库?

　　2. 加上事务后还能动态切换数据库吗? Spring的事务是怎么开启的?

　　3. 结合Mybatis 如何实现动态切换的同时开启分布式事务?

My opinion:

　　For question 1:

　　我们用一个类来继承AbstractRoutingDataSource,只需实现一个方法,那就是

　　@Override

    protected Object determineCurrentLookupKey() {

        log.info("调用一次{}", MultiDataSourceHolder.getDataSourceKey());

        String dataSourceKey = MultiDataSourceHolder.getDataSourceKey();

        //初始化AbstractRoutingDataSource实现类时,dsRoutingSetProperties保存了多数据源名称

        return dataSourceKey == null ? dsRoutingSetProperties.getDataSourceKeysMapping().get(0) : dataSourceKey;

    }

　　我们来看一下具体方法如何作用的

　　　　//类实现了DataSource接口 getConnection() 调用此方法

protected DataSource determineTargetDataSource() {

        Object lookupKey = this.determineCurrentLookupKey();

　　　　 //这里的作用的resolvedDataSources也是一个map,而我们实例化多数据源时设置的是名叫targetDataSource的map

        DataSource dataSource = (DataSource)this.resolvedDataSources.get(lookupKey);

        if (dataSource == null && (this.lenientFallback || lookupKey == null)) {

　　　　 dataSource = this.resolvedDefaultDataSource; } }

      //我们来看另一个方法afterPropertiesSet,这个方法是实现了接口InitializingBean,在ioc实例化bean的时候,还未生成代理对象前会调用的方法,也就是还为注入到单例缓存池时

public void afterPropertiesSet() {

    　　...

        this.resolvedDataSources = new HashMap(this.targetDataSources.size());

　　　　 //java8添加的迭代方式

        this.targetDataSources.forEach((key, value) -> {

            Object lookupKey = this.resolveSpecifiedLookupKey(key);

　　　　　　　//实际上只有这里做了处理,如果map的值是String类型,用jndi的方式获取DataSource(我的技能树没点到这里,看不懂)

            DataSource dataSource = this.resolveSpecifiedDataSource(value);

            this.resolvedDataSources.put(lookupKey, dataSource);

        });

        if (this.defaultTargetDataSource != null) {

            //这里的处理同上

            this.resolvedDefaultDataSource = this.resolveSpecifiedDataSource(this.defaultTargetDataSource);

        }

　　看到这里,我们大概知道了AbstractRoutingDataSource其实就是对DataSource的包装,之前baidu搜索spring的设计模式时,大部分的博文都会说到Spring动态切换数据库就

用的是装饰模式,每每看到这里就是一脸懵(为了整明白才在个人Demo上加上动态数据库切换),做完之后那么问题来了,加了事务后的结果是怎么样呢?

　　For question 2:

　　看一下分别加了没加@Transactional 和加了注解的两个结果 null和dataSourcexx 是切面时存入线程变量的值

分别调用三次和调用一次,这代表着一共产生了三个连接和一个连接.为什么加了事务之后只用了一次连接 ? 切面往线程变量里存了映射,但是没起作用?

花了三天才解决这个问题,这里我们从头撸起找原因

先看下事务的三个关键性接口

PlatformTransactionManager    getTransaction/commit/rollback

TransactionStatus   事务运行状态   isNewTransaction()/hasSavepoint()     是否为新事务/是否有保存点

TransactionDefinition   事务属性 定义了大量关于隔离级别/传播行为的常量

我们知道@Transactional声明式事务是基于动态代理,事务也是以aop参与责任链形式调用

从头看 @EnableTransactionManagement,点开可以看到 @Import({TransactionManagementConfigurationSelector.class}),我们点进去瞅瞅

//ImportSelector接口需要实现的方法,作用是只要返回类的全限定名即可将Bean注入到容器内

protected String[] selectImports(AdviceMode adviceMode) {

        switch(adviceMode) {

　　　　　//打个断点进去 发现走的是Proxy,默认走的就是Proxy,这里注册了两个bean,进去ProxyTransactionManagementConfiguration里看看,发现就是一个配置类,

   //注册了三个bean,我们看一下TransactionInterceptor(拦截器,aop作用的实质)这个类

　　进入TransactionInterceptor类后,感觉有用的就只有一个invoke方法,方法内回调父类的 invokeWithinTransaction()方法,我们跟进去看

　   //方法很长,直接打一个端点在第一行,随便一个方法加上@Transactional,顺利进入

    protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass, TransactionAspectSupport.InvocationCallback invocation) throws Throwable {

　　　　 Object result;

        if (...) {

          ...

        } else {

　　　　　　　 //方法直接进到这里了

            TransactionAspectSupport.TransactionInfo txInfo = this.createTransactionIfNecessary(tm, txAttr, joinpointIdentification);

            try {

　　　　　　　　　　//执行下一层拦截器链 直到执行目标方法

                result = invocation.proceedWithInvocation();

protected TransactionAspectSupport.TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm, @Nullable TransactionAttribute txAttr, final String joinpointIdentification) {

      ....

                status = tm.getTransaction((TransactionDefinition)txAttr);

      ...

　　　　 //把TransactionStatus封装进TransactionInfo对象中,同时绑定到线程变量

        return this.prepareTransactionInfo(tm, (TransactionAttribute)txAttr, joinpointIdentification, status);

public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException {

 　　　　 Object transaction = this.doGetTransaction();

　　　　 //如果ConnnectionHolder不为null transactionActive为true 进入这里 ,因为已经确定存在事务了,进去就事务传播行为做判断,是否建立保存点等

        if (this.isExistingTransaction(transaction)) {

            return this.handleExistingTransaction((TransactionDefinition)definition, transaction, debugEnabled);

        }

　　　　　　.....

　　　　　　else {

　　　　　　　//第一次 进入这里   不需要处理其他

            AbstractPlatformTransactionManager.SuspendedResourcesHolder suspendedResources = this.suspend((Object)null);

protected Object doGetTransaction() {

        DataSourceTransactionManager.DataSourceTransactionObject txObject = new DataSourceTransactionManager.DataSourceTransactionObject();

        txObject.setSavepointAllowed(this.isNestedTransactionAllowed());

　　　　 //这里第一次(如果事务多次访问数据库)返回的为null,下文会看到具体过程

        ConnectionHolder conHolder = (ConnectionHolder)TransactionSynchronizationManager.getResource(this.obtainDataSource());

        txObject.setConnectionHolder(conHolder, false);

　　//看到这里可以发现 transaction 对象就是DataSourceTransactionObject,继承自JdbcTransactionObjectSupport,有一个属性ConnectionHolder,用来干嘛的??

        return txObject;

    }

public static Object getResource(Object key) {

        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);

　　　　 //这里的key就是 我们在声明TransactionManager注入的数据源 ,我这里就是MultiDataSource

        Object value = doGetResource(actualKey);

private static Object doGetResource(Object actualKey) {

　　　　　//记住这个 resources  它是一个ThreadLocal<Map<Object, Object>>对象 ,第一进来肯定直接return null了

        Map<Object, Object> map = (Map)resources.get();

        if (map == null) {

            return null;

        } else {

　　　　　　　//这里试想 如果map不为null 通过绑定的数据源 我们能拿到什么 ?

            Object value = map.get(actualKey);

            if (value instanceof ResourceHolder && ((ResourceHolder)value).isVoid()) {

                map.remove(actualKey);

                if (map.isEmpty()) {

                    resources.remove();

                }

                value = null;

            }

            return value;

        }

    }

至此,函数回调 ,执行doBegin()方法

protected void doBegin(Object transaction, TransactionDefinition definition) {

        DataSourceTransactionManager.DataSourceTransactionObject txObject = (DataSourceTransactionManager.DataSourceTransactionObject)transaction;

        Connection con = null;

        try {

　　　　　　　　//前文发现,自身的ConnnectionHolder对象为null,这里就直接进入了

            if (!txObject.hasConnectionHolder() || txObject.getConnectionHolder().isSynchronizedWithTransaction()) {

                Connection newCon = this.obtainDataSource().getConnection();

                if (this.logger.isDebugEnabled()) {

                    this.logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");

                }

　　　　　　　　　　//创建连接,重新设置ConnectionHolder对象

                txObject.setConnectionHolder(new ConnectionHolder(newCon), true);

            }

            txObject.getConnectionHolder().setSynchronizedWithTransaction(true);

            con = txObject.getConnectionHolder().getConnection();

            Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);

            txObject.setPreviousIsolationLevel(previousIsolationLevel);

            if (con.getAutoCommit()) {

                txObject.setMustRestoreAutoCommit(true);

                if (this.logger.isDebugEnabled()) {

                    this.logger.debug("Switching JDBC Connection [" + con + "] to manual commit");

                }

　　　　　　　　　　//取消自动提交事务

                con.setAutoCommit(false);

            }

         　　// 这里影响后面 第二次进入 是否还会进入本方法

            txObject.getConnectionHolder().setTransactionActive(true);

            if (txObject.isNewConnectionHolder()) {

　　　　　　　　　　//方法进入这里  bindResource方法是不是有点熟悉 ?前文有个方法叫doGetResource()

                TransactionSynchronizationManager.bindResource(this.obtainDataSource(), txObject.getConnectionHolder());

            }

public static void bindResource(Object key, Object value) throws IllegalStateException {

        Map<Object, Object> map = (Map)resources.get();

　　　　//之前的map为空,这下不为空了

        if (map == null) {

            map = new HashMap();

            resources.set(map);

        }

　　　　 //之前介绍的key 就是数据源实例对象,value就是刚传进来的ConnnectionHolder

        Object oldValue = ((Map)map).put(actualKey, value);

看到这里大概清楚了为什么AbstractRoutingSource的方法为什么没有调用.大致总结下, 希望不会带偏读者

Spring创建事务时,每次都会

　　1.创建一个TransactionAspectSupport.TransactionInfo对象,该对象中封装了PlatformTransactionManager,joinpointIdentification("被调用方法名"),TransactionStatus(事务状态,封装一个DataSourceTransactionObject对象,也是就是transaction对象,里面再包一层ConnectionHolder,保存数据库连接),同时关联一个本类对象OldTransactionInfo,

　　2.获取transaction对象,先尝试从线程变量中获取ConnectionHolder对象,判定ConnectionHolder是否为null,如果是空值,则定义此次事务为新事务(newTransaction),从事务管理器中绑定的数据源中获取连接,绑定到ConnectionHolder对象中,并将holder存入线程变量.

　 3,将OldTransaction对象设为自身,并存入线程变量,执行下一层aop或目标方法

　　4.事务内再度调用带事务切面的代理对象时,步骤2从线程变量获取到Holder,进入handleExistingTransaction()方法,就事务传播行为决定是否挂起事务,设立保存点等,将事务属性newTransaction设为false

　　....

在发现这个问题时尝试了各种诡异举措来试图解决它:

　　1.编程式事务TransactionTemplate 可以分为两个事务,但是这并没有什么意义..能不能使两个事务一起提交?

　　2.自定义MyTemplate继承TransactionTemplate ,commit时开启新线程,当两个运行都观测到目标结果(如redis)才提交,..可想而知,新的线程,存在原线程变量里的啥都成了孤岛, 卒..

　　3.自定义事务管理器 / 自定义SqlSessionTemplate,将三个数据源都绑定到SqlSessionTemplate中,获取线程变量中的数据库名,返回对应的数据源 .. 然而不是不会用就是根本就没被调用.. 这方面的知识还是太浅薄了..

For question 3:

　　查看网上很多的案例,发现有用JdbcTemplate控制的,有分多个SqlSessionFactory 用JTA控制的,让Dao层分开被SqlSessionFactory管理,确实可以实现跨库事务,但这一开始就决定了要落入哪个库.和我这个应用场景不一致..能否借鉴折中一下呢?

@Configuration

@MapperScan(value = "com.zuan.cinema.mapper",sqlSessionFactoryRef = "sqlSessionFactory00")

@MapperScan(value = "com.zuan.cinema.mapper.m",sqlSessionFactoryRef = "sqlSessionFactory01")

public class DataSourceConfiguration {

　　 @Bean

    @Primary

    public SqlSessionFactory sqlSessionFactory00(@Qualifier("dataSource00") DataSource dataSource) throws Exception {

        SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();

        sqlSessionFactoryBean.setDataSource(dataSource);

        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:mybatis/mapper/*.xml"));

        Interceptor interceptor = new PageInterceptor();

        sqlSessionFactoryBean.setPlugins(new Interceptor[]{interceptor});

        return sqlSessionFactoryBean.getObject();

    }

    @Bean

    public SqlSessionFactory sqlSessionFactory01(@Qualifier("multiDataSource") DataSource dataSource) throws Exception {

        SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();

        sqlSessionFactoryBean.setDataSource(dataSource);

        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:mybatis/mapper/m/*.xml"));

        Interceptor interceptor = new PageInterceptor();

        sqlSessionFactoryBean.setPlugins(new Interceptor[]{interceptor});

        return sqlSessionFactoryBean.getObject();

    }

}

　　因为crud操作时,只会受一个字段影响决定使用哪个数据库的连接,那么就可以定义两个sqlSessionFatory,一个管理普通mapper,引用普通数据源,另一个管理直接关联字段的mapper,应用MultiDataSource,这样进入事务后必然通过MultiDataSource动态选择数据库.这里只有用JTA事务管理才能实现切换,它和DataSourceTransactionManager实现机制不一样,建立的连接也不样.更多的我看不懂...

Epilog:

　　结果只在个人Demo场景下测试成功,没有经历过生产环境的洗礼.并且上面的推断也出自未经过生产环境的人之口,请自行判断...

巴特西

SSM使用AbstractRoutingDataSource后究竟如何解决跨库事务

最新文章

热门文章