AbstractRoutingDataSource实现动态数据源切换 专题

Posted 2022-12-19

 

需求：系统中要实现切换数据库（业务数据库和his数据库）

网上很多资料上有提到AbstractRoutingDataSource，大致是这么说的

在Spring 2.0.1中引入了AbstractRoutingDataSource, 该类充当了DataSource的路由中介, 能有在运行时, 根据某种key值来动态切换到真正的DataSource上。

     Spring动态配置多数据源，即在大型应用中对数据进行切分，并且采用多个数据库实例进行管理，这样可以有效提高系统的水平伸缩性。而这样的方案就会不同于常见的单一数据实例的方案，这就要程序在运行时根据当时的请求及系统状态来动态的决定将数据存储在哪个数据库实例中，以及从哪个数据库提取数据。

Spring对于多数据源，以数据库表为参照，大体上可以分成两大类情况： 
一是，表级上的跨数据库。即，对于不同的数据库却有相同的表（表名和表结构完全相同）。 
二是，非表级上的跨数据库。即，多个数据源不存在相同的表。 
Spring2.x的版本中采用Proxy模式，就是我们在方案中实现一个虚拟的数据源，并且用它来封装数据源选择逻辑，这样就可以有效地将数据源选择逻辑从Client中分离出来。Client提供选择所需的上下文（因为这是Client所知道的），由虚拟的DataSource根据Client提供的上下文来实现数据源的选择。 
具体的实现就是，虚拟的DataSource仅需继承AbstractRoutingDataSource实现determineCurrentLookupKey（）在其中封装数据源的选择逻辑

一、原理

首先看下AbstractRoutingDataSource类结构，继承了AbstractDataSource：

public abstract class AbstractRoutingDataSource extends org.springframework.jdbc.datasource.AbstractDataSource implements org.springframework.beans.factory.InitializingBean

既然是AbstractDataSource，当然就是javax.sql.DataSource的子类，于是我们自然地回去看它的getConnection方法：

public Connection getConnection() throws SQLException 
        return determineTargetDataSource().getConnection();
    

    public Connection getConnection(String username, String password) throws SQLException 
        return determineTargetDataSource().getConnection(username, password);
    

原来关键就在determineTargetDataSource()里：

protected DataSource determineTargetDataSource() 
        Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
        Object lookupKey = determineCurrentLookupKey();//业务代码能更改这个值，就可使用指定的DB
        DataSource dataSource = this.resolvedDataSources.get(lookupKey);//此处来获取指定的DB
        if (dataSource == null && (this.lenientFallback || lookupKey == null)) 
            dataSource = this.resolvedDefaultDataSource;
        
        if (dataSource == null) 
            throw new IllegalStateException("Cannot determine target DataSource for lookup key [" + lookupKey + "]");
        
        return dataSource;
    

 

这里用到了我们需要进行实现的抽象方法determineCurrentLookupKey()，该方法返回需要使用的DataSource的key值，然后根据这个key从resolvedDataSources这个map里取出对应的DataSource，如果找不到，则用默认的resolvedDefaultDataSource。

回过头看AbstractDataSource的afterPropertiesSet方法：

1 public void afterPropertiesSet() 
 2         if (this.targetDataSources == null) 
 3             throw new IllegalArgumentException("Property targetDataSources is required");
 4         
 5         this.resolvedDataSources = new HashMap<Object, DataSource>(this.targetDataSources.size());
 6         for (Map.Entry entry : this.targetDataSources.entrySet())    
 7             Object lookupKey = resolveSpecifiedLookupKey(entry.getKey());
 8             DataSource dataSource = resolveSpecifiedDataSource(entry.getValue());
 9             this.resolvedDataSources.put(lookupKey, dataSource);//这个值就是所有DataSource的集合
10         
11         if (this.defaultTargetDataSource != null) 
12             this.resolvedDefaultDataSource = resolveSpecifiedDataSource(this.defaultTargetDataSource);
13         
14     

 

扩展：afterPropertiesSet方法调用时机：javaconfig配置场景，打断点时显示，在@Bean所在方法执行完成后，会调用此方法

package org.springframework.beans.factory;

/**
 * Interface to be implemented by beans that need to react once all their
 * properties have been set by a BeanFactory: for example, to perform custom
 * initialization, or merely to check that all mandatory properties have been set.
 *
 * <p>An alternative to implementing InitializingBean is specifying a custom
 * init-method, for example in an XML bean definition.
 * For a list of all bean lifecycle methods, see the BeanFactory javadocs.
 *
 * @author Rod Johnson
 * @see BeanNameAware
 * @see BeanFactoryAware
 * @see BeanFactory
 * @see org.springframework.beans.factory.support.RootBeanDefinition#getInitMethodName
 * @see org.springframework.context.ApplicationContextAware
 */
public interface InitializingBean 

    /**
     * Invoked by a BeanFactory after it has set all bean properties supplied
     * (and satisfied BeanFactoryAware and ApplicationContextAware).
     * <p>This method allows the bean instance to perform initialization only
     * possible when all bean properties have been set and to throw an
     * exception in the event of misconfiguration.
     * @throws Exception in the event of misconfiguration (such
     * as failure to set an essential property) or if initialization fails.
     */
    void afterPropertiesSet() throws Exception;

 

 

配置数据源实例：

<bean id="onlineDynamicDataSource" class="com.xx.stat.base.dynamic.DynamicDataSource">
       <property name="targetDataSources">   
          <map key-type="java.lang.String">   
             <entry key="xx" value-ref="dataSourceXX"/>   
             <entry key="yy" value-ref="dataSourceYY"/>   
          </map>   
       </property>   
       <property name="defaultTargetDataSource" ref="dataSource"/>  
    </bean>

 

观察上面的配置文件，发现我们配置的是targetDataSources和defaultTargetDataSource

二、Spring配置多数据源的方式和具体使用过程

1、数据源的名称常量类

public enum DatabaseTypeEnum 
    DB_DLHMC("dlhmc", "dlhmc数据库,默认的数据库"),DB_HIS("his", "HIS数据库");
    private String value;
    private String desc;

    private DatabaseTypeEnum(String value, String description) 
        this.value = value;
        this.desc = description;
    

    public String getValue() 
        return value;
    

    public String getDesc() 
        return desc;
    

    @Override
    public String toString() 

        return "" + value + ":" + desc + "";
    

    public static DatabaseTypeEnum from(String value) 
        for (DatabaseTypeEnum item : values()) 
            if (item.getValue() == value) 
                return item;
            
        
        throw new IllegalArgumentException(String.format(
                "非法的输入参数 %s ! 必须是%s中的其中一个。", value, Arrays.asList(values())
                        .toString()));
    

 

2、建立一个获得和设置上下文环境的类，主要负责改变上下文数据源的名称

1 public class DatabaseContextHolder 
 2     private static ThreadLocal<String> contextHolder=new ThreadLocal<String>();
 3     public static void setDbType(String dbType)
 4         contextHolder.set(dbType);
 5     
 6     public static String getDbType()
 7         return contextHolder.get();
 8     
 9     
10     public static void clearDbType()
11         contextHolder.remove();
12     
13 
14 

 

3、建立动态数据源类，注意，这个类必须继承AbstractRoutingDataSource，且实现方法 determineCurrentLookupKey，该方法返回一个Object，一般是返回字符串

import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;

public class DynamicDataSource extends AbstractRoutingDataSource
    @Override
    protected Object determineCurrentLookupKey() 
        return DatabaseContextHolder.getDbType();//可以决定使用那个db
    

 

4、编写spring的配置文件配置多个数据源
Java Config
一个示例

@Bean
    public AbstractRoutingDataSource routingDataSource(@Qualifier("dataSourceDLHMC") DataSource dataSourceDLHMC,
                                                       @Qualifier("dataSourceHIS") DataSource dataSourceHIS) 
        Map<Object, Object> targetDataSources = new HashMap<>();
        targetDataSources.put(DatabaseTypeEnum.DB_DLHMC, dataSourceDLHMC);
        targetDataSources.put(DatabaseTypeEnum.DB_HIS, dataSourceHIS);

        AbstractRoutingDataSource routingDataSource = new DynamicDataSource();
        routingDataSource.setTargetDataSources(targetDataSources);
        routingDataSource.setDefaultTargetDataSource(dataSourceDLHMC);
        return routingDataSource;
    

或 

xml

<!-- 数据源配置 -->
    <bean id="defaultDS" class="com.alibaba.druid.pool.DruidDataSource" p:driverClassName="$jdbc.driver" p:url="$jdbc.jdbcUrl" p:username="$jdbc.username" p:password="$jdbc.password"
        p:initialSize="$jdbc.initialSize" p:maxActive="$jdbc.maxActive" p:testOnBorrow="$jdbc.testOnBorrow:false" destroy-method="close">
    </bean>
    <bean id="hisDS" class="com.alibaba.druid.pool.DruidDataSource" p:driverClassName="$his.jdbc.driver" p:url="$his.jdbc.jdbcUrl" p:username="$his.jdbc.username" p:password="$his.jdbc.password"
        p:initialSize="$his.jdbc.initialSize" p:maxActive="$his.jdbc.maxActive" p:testOnBorrow="$his.jdbc.testOnBorrow:false" destroy-method="close">
    </bean>
    <bean id="dataSource" class="com.supconit.util.datasource.DynamicDataSource">
        <property name="targetDataSources">
            <map key-type="java.lang.String">
                <entry key="dlhmc" value-ref="defaultDS" />
                <entry key="his" value-ref="hisDS" />
                <!-- entry key="2" value-ref="ds2" / -->
            </map>
        </property>
        <property name="defaultTargetDataSource" ref="defaultDS" />
    </bean>

 

5、使用

@Override
public List<VBedPatientNew> selectNursinglevel() 
        DatabaseContextHolder.setDbType(DatabaseTypeEnum.DB_HIS.getValue());//指定一个数据源
        List<VBedPatientNew> result=selectList("selectNursinglevel");//业务代码
        DatabaseContextHolder.clearDbType();//清除当前设置。防内存洗发露
        return result;

也可以使用AOP来实现：

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

@Target(ElementType.METHOD, ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface HISDB 

 

import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Aspect
@Component
public class DriverDBAspect 

    public static final Logger LOGGER = LoggerFactory.getLogger(DriverDBAspect.class);

    @Around("@annotation(HISDB)")
    public Object proceed(ProceedingJoinPoint proceedingJoinPoint) throws Throwable 
        Object result = null;
        try 
            LOGGER.info("=============begin set db to driver=============");
            DBContextHolder.setDatabase(DatabaseTypeEnum.DB_HIS);
            result = proceedingJoinPoint.proceed();
            LOGGER.info("=============end set db to driver=============");
         finally 
            DBContextHolder.clear();
        
        return result;
    

@Override
@HISDB
public List<VBedPatientNew> selectNursinglevel() 
     List<VBedPatientNew> result=selectList("selectNursinglevel");
     return result;

 

 aop使用场景的另一种用法：

import com.example.springboot.multidatasource.annotation.DataSourceTypeAnno;
import com.example.springboot.multidatasource.common.DataSourceContextHolder;
import com.example.springboot.multidatasource.common.DataSourceEnum;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;

@Component
@Aspect
public class DataSourceAspect 
    @Pointcut("execution(* com.example.springboot.multidatasource.dao..*(..)) " +
            "&& @annotation(com.example.springboot.multidatasource.annotation.DataSourceTypeAnno)")
    public void dataSourcePointcut() 
    

    @Around("dataSourcePointcut()")
    public Object doAround(ProceedingJoinPoint pjp) 
        MethodSignature methodSignature = (MethodSignature) pjp.getSignature();
        Method method = methodSignature.getMethod();
        DataSourceTypeAnno typeAnno = method.getAnnotation(DataSourceTypeAnno.class);
        DataSourceEnum sourceEnum = typeAnno.value();

        if (sourceEnum == DataSourceEnum.master) 
            DataSourceContextHolder.setDataSourceType(DataSourceEnum.master);
         else if (sourceEnum == DataSourceEnum.slaver) 
            DataSourceContextHolder.setDataSourceType(DataSourceEnum.slaver);
        

        Object result = null;
        try 
            result = pjp.proceed();
         catch (Throwable throwable) 
            throwable.printStackTrace();
         finally 
            DataSourceContextHolder.resetDataSourceType();
        

        return result;
    

 

 

 

 

springboot多数据源读写分离和主库数据源service层事务控制

读写分离如果撇开框架无非就是实现多个数据源，主库用写的数据源，从库用读的数据源。

因为想研究​​数据库​​读写分离和分库分表的设计，所以就自己搭建了一套springboot+druid+mybatis+aop 实现一主多从的设计。 
第一步：首先需要自定义数据源的配置项，springboot默认解析的是带前缀​​spring​​.datasource.下面的配置项，为了不冲突，就直接定义datasource.当成我们的前缀， 
@ConfigurationProperties(prefix = “datasource.write”)可以用来加载指定前缀的配置项，非常方便 
因为使用druid，所以需要生成datasource的时候需要指定类型。

DataSourceBuilder.create().type(dataSourceType).build()

readSize是用来定义从库的大小，有多少从库就要配置多少个从库datasource 
第二步：从库的负载均衡，主要是MyAbstractRoutingDataSource这个类 
第三步，从写springboot-mybatis架包的MybatisAutoConfiguration类的创建SqlSessionFactory方法，将里面的数据源换成我们自定义的AbstractRoutingDataSource 
第四步骤。自定义事务MyDataSourceTransactionManagerAutoConfiguration

完整代码和单元​​测试​​： 
github:​​​https://github.com/ggj2010/javabase.git​​

主要的架包

<!-- jdbc driver begin-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <scope>runtime</scope>
        </dependency>
        <!--mybatis springboot-->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
        </dependency>
        <!-- jdbc driver end-->
        <dependency>
            <groupId>org.aspectj</groupId>
            <artifactId>aspectjweaver</artifactId>
        </dependency>

自定义数据源配置项：

#多数据源 1主2从
datasource:
 #从库数量
  readSize: 2
   # 使用druid数据源
  type: com.alibaba.druid.pool.DruidDataSource
#主库
  write:
    url: jdbc:mysql://localhost:3306/master?useUnicode=true&characterEncoding=utf-8
    username: root
    password: root
    driver-class-name: com.mysql.jdbc.Driver
    filters: stat
    maxActive: 20
    initialSize: 1
    maxWait: 60000
    minIdle: 1
    timeBetweenEvictionRunsMillis: 60000
    minEvictableIdleTimeMillis: 300000
    validationQueryTimeout: 900000
    validationQuery: SELECT SYSDATE() from dual
    testWhileIdle: true
    testOnBorrow: false
    testOnReturn: false
    poolPreparedStatements: true
    maxOpenPreparedStatements: 20
  read1:
    url: jdbc:mysql://localhost:3306/slave1?useUnicode=true&characterEncoding=utf-8
    username: root
    password: root
    driver-class-name: com.mysql.jdbc.Driver
    filters: stat
    maxActive: 20
    initialSize: 1
    maxWait: 60000
    minIdle: 1
    timeBetweenEvictionRunsMillis: 60000
    minEvictableIdleTimeMillis: 300000
    validationQueryTimeout: 900000
    validationQuery: SELECT SYSDATE() from dual
    testWhileIdle: true
    testOnBorrow: false
    testOnReturn: false
    poolPreparedStatements: true
    maxOpenPreparedStatements: 20
  read2:
    url: jdbc:mysql://localhost:3306/slave2?useUnicode=true&characterEncoding=utf-8
    username: root
    password: root
    driver-class-name: com.mysql.jdbc.Driver
    filters: stat
    maxActive: 20
    initialSize: 1
    maxWait: 60000
    minIdle: 1
    timeBetweenEvictionRunsMillis: 60000
    minEvictableIdleTimeMillis: 300000
    validationQueryTimeout: 900000
    validationQuery: SELECT SYSDATE() from dual
    testWhileIdle: true
    testOnBorrow: false
    testOnReturn: false
    poolPreparedStatements: true
    maxOpenPreparedStatements: 20

解析配置项：

@Configuration
@Slf4j
public class DataSourceConfiguration 

    @Value("$datasource.type")
    private Class<? extends DataSource> dataSourceType;

    @Bean(name = "writeDataSource")
    @Primary
    @ConfigurationProperties(prefix = "datasource.write")
    public DataSource writeDataSource() 
        log.info("-------------------- writeDataSource init ---------------------");
        return DataSourceBuilder.create().type(dataSourceType).build();
    

    /**
     * 有多少个从库就要配置多少个
     * @return
     */
    @Bean(name = "readDataSource1")
    @ConfigurationProperties(prefix = "datasource.read1")
    public DataSource readDataSourceOne() 
        log.info("-------------------- readDataSourceOne init ---------------------");
        return DataSourceBuilder.create().type(dataSourceType).build();
    

    @Bean(name = "readDataSource2")
    @ConfigurationProperties(prefix = "datasource.read2")
    public DataSource readDataSourceTwo() 
        log.info("-------------------- readDataSourceTwo init ---------------------");
        return DataSourceBuilder.create().type(dataSourceType).build();
    

 

重写SqlSessionFactory

@Configuration
@AutoConfigureAfter( DataSourceConfiguration.class )
@Slf4j
public class MybatisConfiguration extends MybatisAutoConfiguration 
    @Value("$datasource.readSize")
    private String dataSourceSize;

    @Bean
    public SqlSessionFactory sqlSessionFactorys() throws Exception 
        log.info("-------------------- 重载父类 sqlSessionFactory init ---------------------");
        return super.sqlSessionFactory(roundRobinDataSouceProxy());
    

    /**
     * 有多少个数据源就要配置多少个bean
     * @return
     */
    @Bean
    public AbstractRoutingDataSource roundRobinDataSouceProxy() 
        int size = Integer.parseInt(dataSourceSize);
        MyAbstractRoutingDataSource proxy = new MyAbstractRoutingDataSource(size);
        Map<Object, Object> targetDataSources = new HashMap<Object, Object>();
        DataSource writeDataSource = SpringContextHolder.getBean("writeDataSource");
        // 写
        targetDataSources.put(DataSourceType.write.getType(), SpringContextHolder.getBean("writeDataSource"));

        for (int i = 0; i < size; i++) 
            targetDataSources.put(i, SpringContextHolder.getBean("readDataSource" + (i + 1)));
        
        proxy.setDefaultTargetDataSource(writeDataSource);
        proxy.setTargetDataSources(targetDataSources);
        return proxy;
    

 

本地线程全局变量

public class DataSourceContextHolder 
    private static final ThreadLocal<String> local = new ThreadLocal<String>();

    public static ThreadLocal<String> getLocal() 
        return local;
    

    /**
     * 读可能是多个库
     */
    public static void read() 
        local.set(DataSourceType.read.getType());
    

    /**
     * 写只有一个库
     */
    public static void write() 
        local.set(DataSourceType.write.getType());
    

    public static String getJdbcType() 
        return local.get();
    

 

多数据源切换

public class MyAbstractRoutingDataSource extends AbstractRoutingDataSource 
    private final int dataSourceNumber;
    private AtomicInteger count = new AtomicInteger(0);

    public MyAbstractRoutingDataSource(int dataSourceNumber) 
        this.dataSourceNumber = dataSourceNumber;
    

    @Override
    protected Object determineCurrentLookupKey() 
        String typeKey = DataSourceContextHolder.getJdbcType();
        if (typeKey.equals(DataSourceType.write.getType()))
            return DataSourceType.write.getType();
        // 读 简单负载均衡
        int number = count.getAndAdd(1);
        int lookupKey = number % dataSourceNumber;
        return new Integer(lookupKey);
    

 

enum类型

public enum DataSourceType 
    read("read", "从库"), write("write", "主库");
    @Getter
    private String type;
    @Getter
    private String name;

    DataSourceType(String type, String name) 
        this.type = type;
        this.name = name;
    

 

aop拦截设置本地线程变量

@Aspect
@Component
@Slf4j
public class DataSourceAop 

    @Before("execution(* com.ggj.encrypt.modules.*.dao..*.find*(..)) or execution(* com.ggj.encrypt.modules.*.dao..*.get*(..))")
    public void setReadDataSourceType() 
        DataSourceContextHolder.read();
        log.info("dataSource切换到：Read");
    

    @Before("execution(* com.ggj.encrypt.modules.*.dao..*.insert*(..)) or execution(* com.ggj.encrypt.modules.*.dao..*.update*(..))")
    public void setWriteDataSourceType() 
        DataSourceContextHolder.write();
        log.info("dataSource切换到：write");
    

 

自定义事务

@Configuration
@EnableTransactionManagement
@Slf4j
public class MyDataSourceTransactionManagerAutoConfiguration extends DataSourceTransactionManagerAutoConfiguration 
    /**
     * 自定义事务
     * MyBatis自动参与到spring事务管理中，无需额外配置，
     *只要org.mybatis.spring.SqlSessionFactoryBean引用的数据源与DataSourceTransactionManager引用的数据源一致即可，
     *否则事务管理会不起作用。
     * @return
     */
    @Bean(name = "transactionManager")
    public DataSourceTransactionManager transactionManagers() 
        log.info("-------------------- transactionManager init ---------------------");
        return new DataSourceTransactionManager(SpringContextHolder.getBean("roundRobinDataSouceProxy"));
    

import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.After;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.springframework.stereotype.Component;

import com.dao.annotation.TargetDataSource;
import com.dao.route.DataSourceSwitch;

import lombok.extern.slf4j.Slf4j;

/**
 * 切换数据源AOP
 *
 */
@Aspect
@Component
@Slf4j
public class TargetDataSourceAspect 
    
    @Before(" @annotation(ds)")
    public void changeDataSource(JoinPoint point, TargetDataSource ds)
        String dsId = ds.value();
        if (!DataSourceSwitch.containsDataSource(dsId)) 
            log.error("数据源[]不存在，使用default:", ds.value(), point.getSignature());
         else 
            log.debug("Use DataSource :  > ", ds.value(), point.getSignature());
            DataSourceSwitch.setDataSource(ds.value());
        
    

    @After(" @annotation(ds)")
    public void restoreDataSource(JoinPoint point, TargetDataSource ds) 
        log.debug("Revert DataSource :  > ", ds.value(), point.getSignature());
        DataSourceSwitch.clearDataSource();