目录

1.前言：

2.背景介绍：

3.Dubbo：

4.动态数据源：

5.分布式事务：

6.运行及结果：

---

1.前言：

本文code的github地址：GitHub - wdquan1985/dubbo-dynamicDatasource-jtaAtomikos

分布式现在已经是web开发必须掌握的知识，虽然其有一定的难度，必须掌握的知识点包括以下两部分：

(1).分布式架构RPC框架：例如Duboo，spring cloud等。本文中使用Dubbo

(2).分布式事务：遇到分布式系统，就几乎不可避免的遇到分布式事务的问题，分布式事务包括四种模式 -- XA、TCC、AT、Saga。2PC/3PC协议是指两阶段提交和三阶段提交，四种分布式事务模式的实现，基本上都使用了2PC协议，因为3PC太难实现。可以参考这篇文章： 分布式事务的4种模式 - 知乎， 去了解分布式事务的全貌。

下面两种情况下，需要分布式事务：

数据库的水平拆分：单业务系统架构

业务数据库起初是单库单表，但随着业务数据规模的快速发展，数据量越来越大，对数据库表的增删改查将会越来越慢。单库单表逐渐成为瓶颈。所以我们对数据库进行了水平拆分，将原单库单表拆分成数据库分片。如下图所示，分库分表之后，原来在一个数据库上就能完成的写操作，可能就会跨多个数据库，这就产生了跨数据库事务问题。访问不同的数据库，要开启不同的事务

业务服务化拆分：多个应用服务

在业务发展初期，“一块大饼”的单业务系统架构，能满足基本的业务需求。但是随着业务的快速发展，系统的访问量和业务复杂程度都在快速增长，单系统架构逐渐成为业务发展瓶颈，代码可维护性差，容错率低，测试难度大，敏捷交付能力差等诸多问题，微服务应运而生。
如下图所示，按照面向服务架构（SOA）的设计原则，将单业务系统拆分成多个业务系统，降低了各系统之间的耦合度，使不同的业务系统专注于自身业务，更有利于业务的发展和系统容量的伸缩。

微服务的诞生一方面解决了上述问题，但是另一方面却引入新的问题，业务系统按照服务拆分之后，一个完整的业务往往需要调用多个服务，所以其中主要问题之一就是如何保证微服务间的业务数据一致性，这成为了一个难题，每个服务访问数据库（即使是相同的数据库）都要开启一个新的事务

(3).分布式事务的实现：

• 符合XA 模式的由TM, RM控制的两阶段提交，RM由关系型数据库厂商提供实现(例如mysql，oracle数据库，本身支持分布式事务)，java对XA规范的实现叫做JTA，JTA中TM具体实现包括Atomikos，Narayana，JOTM，BTM等。但是这种XA模式是有局限性的，只能够解决单个应用(单业务系统架构)中跨越多个数据源时（分布式数据库，数据库的水平拆分）数据操作的事务一致性问题，但是遇到分布式系统（例如Dubbo）中多个应用服务之间（ 分布式服务，业务服务化拆分）数据操作的事务一致性问题时，它就不起作用了，也就是说只适用于单业务系统架构，当然了，在分布式系统中，如果没有遇到跨微服务访问数据库的事务问题时，跟这个其实也是类似的。还有其它局限性，例如必须要拿到所有数据源，而且数据源还要支持XA协议；性能比较差，要把所有涉及到的数据都锁定，是强一致性的，会产生长事务。本文是初探分布式事务，所以选择使用Atomikos, 这对分布式事务有一个初步的了解很有帮助。会在以后的文章中选择其它的分布式事务实现，例如TCC的seata。总结：XA模式是分布式强一致性的解决方案，但性能低而使用较少，CAP理论中，很多互联网应用都是选择满足AP的。
• 如何解决 JTA/XA 的局限性呢？阿里的Seata 分别实现了 AT、TCC、SAGA 和 XA （XA尚未实现，请等待）事务模式，为用户打造一站式的分布式解决方案。以后会写文章研究阿里seata的使用。可以学习Seata入门理论介绍文章：分布式事务 Seata 及其三种模式详解
• Seata官网地址：Seata

2.背景介绍：

我们为什么要使用Dubbo?

随着Internet的发展，web应用的规模不断扩大，最终，我们发现传统的架构(单片应用，垂直架构等)已经无法满足需求。分布式服务架构和流计算架构是必须的，并且迫切需要一个治理架构来确保架构的有序发展。

整体架构:

在流量非常低的情况下，只有一个应用，所有的功能被部署在一起（即一个server上，例如一个tomcat），这样可以减少部署节点，降低成本。此时，数据访问框架(ORM)是简化CRUD工作负载的关键。

垂直架构:

当流量较大时，增加单片应用程序的实例（简单的集群，例如用nginx做负载均衡）并不能很好的加速访问，提高效率的一种方法是将单片应用拆分为离散的应用，此时，用于加速前端页面开发的Web框架(MVC)是关键。

分布式服务架构：

当垂直应用程序越来越多时，应用程序之间的交互是不可避免的，一些核心业务被提取出来并作为独立的服务来服务，从而逐渐形成一个稳定的服务中心，这样前端应用程序就可以更好地响应不断变化的市场需求。 很快。 此时，用于业务重用和集成的分布式服务框架（RPC）是关键。

流计算架构：
当服务越来越多时，容量评估变得困难，而且小规模的服务也经常造成资源浪费。 为解决这些问题，应添加调度中心，以根据流量管理集群容量并提高集群利用率。 目前，用于提高机器利用率的资源调度和治理中心（SOA）是关键。

3.Dubbo：

Dubbo官网：https://dubbo.apache.org/zh/

Dubbo架构:

指定节点角色：

• Provider：提供者提供并且暴露远程服务
• Consumer：调用Provider提供的远程服务
• Registry：注册中心负责服务注册和发现
• Monitor：监控中心计算服务的调用次数和耗时
• Container：远程服务在容器中启动，容器管理服务的生命周期

代码：

代码分为三部分：

1. dubbo-example-common: 通用API（接口定义），以及数据库表对应的model。
2. dubbo-example-consumer: 服务消费者，基于Springboot提供了Controller层，用户可以通过http请求访问。
3. dubbo-example-provider: 服务提供者，基于Springboot集成了Atomikos + mybatis-plus + mysql, 对多数据源提供了分布式事务管理。

父pom.xml中包含的主要依赖：其中包含了Zookeeper依赖，本实例使用Zookeeper做为注册中心。

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.dubbo</groupId>
                <artifactId>dubbo-dependencies-bom</artifactId>
                <version>${dubbo.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.dubbo</groupId>
                <artifactId>dubbo</artifactId>
                <version>${dubbo.version}</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.springframework</groupId>
                        <artifactId>spring</artifactId>
                    </exclusion>
                    <exclusion>
                        <groupId>javax.servlet</groupId>
                        <artifactId>servlet-api</artifactId>
                    </exclusion>
                    <exclusion>
                        <groupId>log4j</groupId>
                        <artifactId>log4j</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.apache.dubbo</groupId>
                <artifactId>dubbo-dependencies-zookeeper</artifactId>
                <version>${dubbo.version}</version>
                <type>pom</type>
                <exclusions>
                    <exclusion>
                        <groupId>org.slf4j</groupId>
                        <artifactId>slf4j-log4j12</artifactId>
                    </exclusion>
                    <exclusion>
                        <groupId>log4j</groupId>
                        <artifactId>log4j</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
        </dependencies>
    </dependencyManagement>

dubbo-example-common，dubbo-example-consumer和dubbo-example-provider要继承父pom。

    <parent>
        <artifactId>springboot-dubbo-atomikos-example</artifactId>
        <groupId>yiyi.example.dubbo</groupId>
        <version>0.0.1</version>
    </parent>

4.动态数据源：

Springboot 的动态数据源是把多个数据源存储在一个 Map 中，当需要使用某个数据源时，从 Map 中获取此数据源进行处理。在 Spring 中，已提供了抽象类 AbstractRoutingDataSource 来实现此功能。因此，我们在实现动态数据源的时候，只需要继承AbstractRoutingDataSource ，实现自己的获取数据源的逻辑即可。本实例使用Springboot的AOP，在service方法层这个级别实现动态数据源切换，没有在service方法内实现AOP动态数据源切换，如果想要在service层方法内实现动态数据源切换，就不要使用AOP的方式了，直接使用代码手动切换。

动态数据源流程如下所示：

动态数据源配置:

(1).数据源key的上下文：

需要在想要切换数据源的时候，立即就可以切换到想要的数据源，因此，需要有一个动态地设置和获取数据源 key 的地方（我们称为上下文），对于 web 应用，访问以线程为单位，所以选择使用 ThreadLocal ，代码如下：

public class DynamicDataSourceContextHolder {

    /**
     * 动态数据源名称上下文
     */
    private static final ThreadLocal<String> DATASOURCE_CONTEXT_KEY_HOLDER = new ThreadLocal<>();
    
    /**
     * 也就是所谓的数据库别名
     * 管理所有的数据源id;
     * 主要是为了判断数据源是否存在;
     */
    public static List<String> dataSourceIds = new ArrayList<String>();

    /**
     * 设置数据源
     * @param key
     */
    public static void setContextKey(String key){
        System.out.println("切换数据源"+key);
        DATASOURCE_CONTEXT_KEY_HOLDER.set(key);
    }

    /**
     * 获取数据源名称
     * @return
     */
    public static String getContextKey(){
        String key = DATASOURCE_CONTEXT_KEY_HOLDER.get();
        return key == null?DataSourceConstants.DS_KEY_MASTER:key;
    }

    /**
     * 删除当前数据源名称
     */
    public static void removeContextKey(){
        DATASOURCE_CONTEXT_KEY_HOLDER.remove();
    }
    
    /**
     * 是否包含指定的数据源名称
     */
    public static boolean containsDataSource(String dataSourceId){
        return dataSourceIds.contains(dataSourceId);
    }
}

(2).添加动态数据源类 , 继承AbstractRoutingDataSource。

public class DynamicDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return DynamicDataSourceContextHolder.getContextKey();;
    }
}

继承抽象类 AbstractRoutingDataSource ，需要实现方法 determineCurrentLookupKey，即路由策略， 根据返回的key值 ，从Map中获取数据源。本例中的key值，是从上面实现的数据源key的上下文中获取的
(3).设置动态数据源
在数据源配置文件 DynamicDataSourceConfig 中，就是将数据源添加到动态数据源的Map中，代码如下：

    @Bean(name = "dynamicDataSource")
    @Primary
    public DataSource dynamicDataSource(
            @Qualifier(DataSourceConstants.DS_KEY_MASTER) DataSource master,
            @Qualifier(DataSourceConstants.DS_KEY_SLAVE) DataSource slave
    		) {
        Map<Object, Object> dataSourceMap = new HashMap<>(2);
        dataSourceMap.put(DataSourceConstants.DS_KEY_MASTER, master);
        dataSourceMap.put(DataSourceConstants.DS_KEY_SLAVE, slave);
        //设置动态数据源
        DynamicDataSource dynamicDataSource = new DynamicDataSource();
        dynamicDataSource.setTargetDataSources(dataSourceMap);
        dynamicDataSource.setDefaultTargetDataSource(master);

        return dynamicDataSource;
    }

(4).配置AOP，动态切换数据源

进行到这一步，其实已经可以切换数据源了，看下面的例子：

    /**
     * 向多个数据源中插入user信息。
     * @return
     */
    @GetMapping("/insertUserInfoToMultiDatasource")
    public void insertUserInfoToMultiDatasource(@RequestParam(value="username", required=true) String username) {
    	//往主数据库中插入数据，切换到master数据源。
    	DynamicDataSourceContextHolder.setContextKey(DataSourceConstants.DS_KEY_MASTER);
    	multiDataSourceService.insertMasterUser(username);
        DynamicDataSourceContextHolder.removeContextKey();

        //往从数据库中插入数据， 切换到slave数据源。
        DynamicDataSourceContextHolder.setContextKey(DataSourceConstants.DS_KEY_SLAVE);
    	multiDataSourceService.insertSlaveUser(username);
        DynamicDataSourceContextHolder.removeContextKey();
    }

但是每次都通过DynamicDataSourceContextHolder.setContextKey切换数据源，如果有大量需要切换数据源的地方，就会产生很多重复，如何消除这些重复的代码，就需要用到动态代理。在 Spring 中，AOP 的实现是基于动态代理的，我们希望通过注解的方式指定service层方法需要的数据源，从而消除数据源切换时产品的模板代码。为什么注解不加在Dao层？因为如果加在Dao层，那么肯定不能切换数据源了，因为Dao层的代码对应于多个数据源，如果在其上加注解，那么就只是对应于一个特定的数据源了。

定义数据源注解:

在annotation包中，添加数据源注解 DS，此注解可以写在类中，也可以写在方法定义中，如下所示：
 

@Target({ElementType.METHOD,ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface DS {
    /**
     * 数据源名称
     */
    String value() default DataSourceConstants.DS_KEY_MASTER;
}

定义数据源切面:
 

@Component
public class DynamicDataSourceAspect {
    @Pointcut("@annotation(me.mason.demo.dynamicdatasource.annotation.DS)")
    public void dataSourcePointCut(){

    }

    @Around("dataSourcePointCut()") //使用环绕通知处理，使用上下文对使用注解DS的service层方法进行数据源切换，处理完后，恢复数据源
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        String dsKey = getDSAnnotation(joinPoint).value();
        DynamicDataSourceContextHolder.setContextKey(dsKey);
        try{
            return joinPoint.proceed();
        }finally {
            DynamicDataSourceContextHolder.removeContextKey();
        }
    }

    /**
     * 根据类或方法获取数据源注解
     */
    private DS getDSAnnotation(ProceedingJoinPoint joinPoint){
        Class<?> targetClass = joinPoint.getTarget().getClass();
        DS dsAnnotation = targetClass.getAnnotation(DS.class);
        // 先判断类的注解，再判断方法注解
        if(Objects.nonNull(dsAnnotation)){
            return dsAnnotation;
        }else{
            MethodSignature methodSignature = (MethodSignature)joinPoint.getSignature();
            return methodSignature.getMethod().getAnnotation(DS.class);
        }
    }
}

使用 AOP 进行数据源切换：

    //主数据库中插入User信息
    @Override
    @DS(value = DataSourceConstants.DS_KEY_MASTER)
    @Transactional
    public void insertMasterUser(String name) {
      TestUser testUser = new TestUser();
      testUser.setName(name);
      testUser.setPhone("13674890382");
      testUser.setRecordVersion(110L);
      testUser.setEmail("hellworld@163.com");
      testUser.setTitle("transcation_test");
      testUserMapper.insert(testUser);
   }

    //从数据库中插入User信息
    @Override
    @DS(value = DataSourceConstants.DS_KEY_SLAVE)
    @Transactional
    public void insertSlaveUser(String name) {
        TestUser testUser = new TestUser();
        testUser.setName(name);
        testUser.setPhone("13674890382");
        testUser.setRecordVersion(110L);
        testUser.setEmail("hellworld@163.com");
        testUser.setTitle("transcation_test");
        testUserMapper.insert(testUser);
   }

5.分布式事务：

XA规范主要定义了 (全局)事务管理器™ 和 (局部)资源管理器(RM) 之间的接口。RM由主流的数据库提供，主流的关系型数据库产品都是实现了XA接口的。TM主要是一些开源软件，例如Atomikos。JTA是java根据XA规范提供的事务处理标准，也是二阶段提交，JTA这种实现是强一致性的体现。按照互联网的实际场景都是最终一致性，柔性事务，其实，大多数情况下很多公司是使用消息队列的方式实现分布式事务。

本文将使用Atomikos做为TM，来实现分布式事务。

添加依赖：

<!--分布式事务-->
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-jta-atomikos</artifactId>
</dependency>

里面已经整合好了transactions-jms、transactions-jta、transactions-jdbc、javax.transaction-api

分布式事务的数据源相关配置：以slave数据源为例，master数据源类似。

    @Bean(DataSourceConstants.DS_KEY_SLAVE)
    @ConfigurationProperties(prefix = "spring.datasource.slave")
    public DataSource slaveDataSource() {
        //JTA标准(XA规范)中的两个角色中的    参与者(participants): 资源管理器(RM)
        MysqlXADataSource mysqlXaDataSource = new MysqlXADataSource();
        mysqlXaDataSource.setUrl(slaveDatabaseProp.getJdbcUrl());
        try {
			mysqlXaDataSource.setPinGlobalTxToPhysicalConnection(true);
		} catch (SQLException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
        mysqlXaDataSource.setPassword(slaveDatabaseProp.getPassword());
        mysqlXaDataSource.setUser(slaveDatabaseProp.getUsername());
        try {
			mysqlXaDataSource.setPinGlobalTxToPhysicalConnection(true);
		} catch (SQLException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}

        //JTA标准(XA规范)中的两个角色中的   协调者(Coordinater): 事务管理器(TM)
        AtomikosDataSourceBean xaDataSource = new AtomikosDataSourceBean();
        // 数据源唯一标识
        xaDataSource.setUniqueResourceName("slaveDataSourcejta");
        //Mysql分布式数据源(连接池)实现类 ->  XADataSource实现类，使用com.mysql.cj.jdbc.MysqlXADataSource
        xaDataSource.setXaDataSourceClassName(slaveDatabaseProp.getType());
        xaDataSource.setMinPoolSize(slaveDatabaseProp.getMinPoolSize());
        xaDataSource.setMaxPoolSize(slaveDatabaseProp.getMaxPoolSize());
        xaDataSource.setMaxLifetime(slaveDatabaseProp.getMaxLifetime());
        xaDataSource.setBorrowConnectionTimeout(slaveDatabaseProp.getBorrowConnectionTimeout());
        try {
			xaDataSource.setLoginTimeout(slaveDatabaseProp.getLoginTimeout());
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        xaDataSource.setMaintenanceInterval(slaveDatabaseProp.getMaintenanceInterval());
        xaDataSource.setMaxIdleTime(slaveDatabaseProp.getMaxIdleTime());
        // 返回连接前用于测试连接的SQL查询
        xaDataSource.setTestQuery(slaveDatabaseProp.getTestQuery());
        xaDataSource.setXaDataSource(mysqlXaDataSource);
        return xaDataSource;
    }

根据Spring官方介绍，要想自己控制事务，则必须实现Transaction接口，所以我们来创建动态数据源事务实现类DynamicDataSourceTransaction

/*
 * 根据Spring官方介绍，要想自己控制事务，则必须实现Transaction接口，所以我们来创建动态数据源事务实现类DynamicDataSourceTransaction
 */
/**
 * <P>多数据源切换，支持事务</P>
* <P>多数据源事务管理器是:根据数据源的不同类型，动态获取数据库连接，而不是从原来的缓存中(ThreadLocal ???)获取导致数据源没法切换</P>
* @author Bruce 2020/12/19
 */
@Slf4j
public class DynamicDataSourceTransaction implements Transaction{
    private final DataSource dataSource;
    private Connection defaultConnection;
    private String dataBaseName;
    private ConcurrentMap<String, Connection> dynamicConnectionMap;
    private boolean isConnectionTransactional;
    private boolean autoCommit;

    public DynamicDataSourceTransaction(DataSource dataSource) {
        Assert.notNull(dataSource, "No DataSource specified");
        this.dataSource = dataSource;
        this.dynamicConnectionMap = new ConcurrentHashMap<>();
        this.dataBaseName = DynamicDataSourceContextHolder.getContextKey();
    }
    
    /**
     * 开启事务处理方法, 事务获取数据源的Connection.
     */
    @Override
    public Connection getConnection() throws SQLException {
        String dataBase = DynamicDataSourceContextHolder.getContextKey();
        if (dataBase.equals(dataBaseName)) {
            if (defaultConnection != null) {
                return defaultConnection;
            }
            openMainConnection();
            dataBaseName = dataBase;
            return defaultConnection;
        } else {
            if (!dynamicConnectionMap.containsKey(dataBase)) {
                try {
                    Connection conn = dataSource.getConnection();
                    dynamicConnectionMap.put(dataBase, conn);
                } catch (SQLException ex) {
                    throw new CannotGetJdbcConnectionException("Could not get JDBC Connection", ex);
                }
            }
            return dynamicConnectionMap.get(dataBase);
        }
    }
    
    private void openMainConnection() throws SQLException {
        this.defaultConnection = DataSourceUtils.getConnection(this.dataSource);
        this.autoCommit = this.defaultConnection.getAutoCommit();
        this.isConnectionTransactional = DataSourceUtils.isConnectionTransactional(this.defaultConnection, this.dataSource);

        if (log.isDebugEnabled()) {
            log.debug(
                    "JDBC Connection ["
                            + this.defaultConnection
                            + "] will"
                            + (this.isConnectionTransactional ? " " : " not ")
                            + "be managed by Spring");
        }
    }

    /**
     * 提交处理方法
     */
    @Override
    public void commit() throws SQLException {
        if (this.defaultConnection != null && !this.isConnectionTransactional && !this.autoCommit) {
            if (log.isDebugEnabled()) {
                log.debug("Committing JDBC Connection [" + this.defaultConnection + "]");
            }
            this.defaultConnection.commit();
            for (Connection connection : dynamicConnectionMap.values()) {
                connection.commit();
            }
        }
    }
    
    /**
     * 回滚处理方法
     */
    @Override
    public void rollback() throws SQLException {
        if (this.defaultConnection != null && !this.isConnectionTransactional && !this.autoCommit) {
            if (log.isDebugEnabled()) {
                log.debug("Rolling back JDBC Connection [" + this.defaultConnection + "]");
            }
            this.defaultConnection.rollback();
            for (Connection connection : dynamicConnectionMap.values()) {
                connection.rollback();
            }
        }
    }

    /**
     * 关闭处理方法
     */
    @Override
    public void close() {
        DataSourceUtils.releaseConnection(this.defaultConnection, this.dataSource);
        for (Connection connection : dynamicConnectionMap.values()) {
            DataSourceUtils.releaseConnection(connection, this.dataSource);
        }
    }

    @Override
    public Integer getTimeout() {
        return null;
    }

}

将两个数据源(master和slave)加入到动态数据源中，并在配置Mybatis的sqlSessionFactory的时候使用动态数据源和自己创建的动态数据源事务实现类DynamicDataSourceTransaction。

    @Bean(name = "dynamicDataSource")
    @Primary
    public DataSource dynamicDataSource(
            @Qualifier(DataSourceConstants.DS_KEY_MASTER) DataSource master,
            @Qualifier(DataSourceConstants.DS_KEY_SLAVE) DataSource slave
    		) {
        Map<Object, Object> dataSourceMap = new HashMap<>(2);
        dataSourceMap.put(DataSourceConstants.DS_KEY_MASTER, master);
        dataSourceMap.put(DataSourceConstants.DS_KEY_SLAVE, slave);
        //设置动态数据源
        DynamicDataSource dynamicDataSource = new DynamicDataSource();
        dynamicDataSource.setTargetDataSources(dataSourceMap);
        dynamicDataSource.setDefaultTargetDataSource(master);

        return dynamicDataSource;
    }
    
    //这是设置mybatis 的 sqlSessionFactory, mybatis也使用session去访问数据库，跟hibernate应该是一样的。
    //session需要用到Datasource的connection， 所以配置里设置了上面的动态数据源 dynamicDataSource。
    @Bean(name = "sqlSessionFactory")
    public SqlSessionFactory sqlSessionFactory(
    		@Qualifier("dynamicDataSource") DataSource dataSource
    		) throws Exception {
        MybatisSqlSessionFactoryBean sqlSessionFactory = new MybatisSqlSessionFactoryBean();
        //设置动态数据源
        sqlSessionFactory.setDataSource(dataSource);
        //设置动态数据源事务管理
        sqlSessionFactory.setTransactionFactory(new DynamicDataSourceTransactionFactory());

        //配置mybatis-plus相关的配置
        //非常重要，由于我们在这里配置了Mybatis, 所以在application.propertis文件中的mybatis-plus***配置，失效
        MybatisConfiguration configuration = new MybatisConfiguration();
        configuration.setJdbcTypeForNull(JdbcType.NULL);
        //数据库字段中下划线，转化为model类中属性的驼峰命名,例如 user_order_num 转为 userOrderNum
        configuration.setMapUnderscoreToCamelCase(true);
        configuration.setCacheEnabled(false);
        //非常重要--可以自定义ID生成策略。
//        GlobalConfig globalConfig = new GlobalConfig();
//        globalConfig.setIdentifierGenerator(new IdentifierGenerator() {
//			
//			@Override
//			public Number nextId(Object entity) {
//				// TODO Auto-generated method stub
//				return null;
//			}
//		  });
//        configuration.setGlobalConfig(globalConfig);
        sqlSessionFactory.setConfiguration(configuration);
        //添加插件
        sqlSessionFactory.setPlugins(new Interceptor[]{
                paginationInterceptor()
        });
        return sqlSessionFactory.getObject();
    }

6.运行及结果：

从不同数据源中获取同一个表的信息，执行效果如下，能够拿到master和slave数据源中的test_user表中的数据。

其它的API怎样使用，在github的说明中已经给出，请读者自己执行。

如果文章对您有用，请帮忙点赞和收藏。