mongodb 多数据源动态切换

Posted 2023-04-05

参考技术A 根据上面场景，现在来说明mongodb的多数据源的影子库，生产库 动态切换怎么完成，mysql 已经有很多成熟的方案了，redis，mq这些有机会再研究.

这个应该说是相当简单了，在压测的请求加个header 标记, 过滤去里面判断到 header 标记是否存在，然后把标记放到 ThreadLocal 里面，后面的动态切换需要依赖这个 副本变量

SpringBoot+aop实现多数据源动态切换

            SpringBoot+aop实现多数据源动态切换

 

一、多数据源动态切换原理

项目中我们经常会遇到多数据源的问题，尤其是数据同步或定时任务等项目更是如此；又例如：读写分离数据库配置的系统。

1、多数据源设置：
1）静态数据源切换：
一般情况下，我们可以配置多个数据源，然后为每个数据源写一套对应的sessionFactory和dao层代码(以hibernate为例，mybatis同理)，——我们称之为静态数据源配置。

2）动态数据源切换：
可看出在Dao层代码中写死了两个SessionFactory，这样日后如果再多一个数据源，还要改代码添加一个SessionFactory，显然这并不符合开闭原则。比较好的做法是，配置多个数据源，只对应一套sessionFactory，数据源之间可以动态切换。

2、动态数据源切换时，如何保证数据库的事务：

目前事务最灵活的方式，是使用spring的声明式事务，本质是利用了spring的aop，在执行数据库操作前后，加上事务处理。

spring的事务管理，是基于数据源的，所以如果要实现动态数据源切换，而且在同一个数据源中保证事务是起作用的话，就需要注意二者的顺序问题，即：在事物起作用之前就要把数据源切换回来。

举一个例子：web开发常见是三层结构：controller、service、dao。一般事务都会在service层添加，如果使用spring的声明式事物管理，在调用service层代码之前，spring会通过aop的方式动态添加事务控制代码，所以如果要想保证事物是有效的，那么就必须在spring添加事务之前把数据源动态切换过来，也就是动态切换数据源的aop要至少在service上添加，而且要在spring声明式事物aop之前添加，根据上面分析:

• 最简单的方式是把动态切换数据源的aop加到controller层，这样在controller层里面就可以确定下来数据源了。不过，这样有一个缺点就是，每一个controller绑定了一个数据源，不灵活。对于这种：一个请求，需要使用两个以上数据源中的数据完成的业务时，就无法实现了。
• 针对上面的这种问题，可以考虑把动态切换数据源的aop放到service层，但要注意一定要在事务aop之前来完成。这样，对于一个需要多个数据源数据的请求，我们只需要在controller里面注入多个service实现即可。但这种做法的问题在于，controller层里面会涉及到一些不必要的业务代码，例如：合并两个数据源中的list…
• 此外，针对上面的问题，还可以再考虑一种方案，就是把事务控制到dao层，然后在service层里面动态切换数据源。

二、多数据源动态切换实现代码：

1、在application.yml中配置多数据源：

# 数据源配置
spring:
  datasource:
    type: com.alibaba.druid.pool.DruidDataSource
    driverClassName: oracle.jdbc.driver.OracleDriver
    druid:
      # 数据源1
      datasource1:
        url: jdbc:oracle:thin:@10.xx.xx.xx:1521:orcl
        username: xxx1
        password: xxx1
      # 数据源2
      datasource2:
        # 数据源开关/默认关闭
        #enabled: false
        url: jdbc:oracle:thin:@10.xx.xx.xx:1521:orcl
        username: xxx2
        password: xxx2
      # 初始连接数
      initialSize: 5
      # 最小连接池数量
      minIdle: 10
      # 最大连接池数量
      maxActive: 20
      # 配置获取连接等待超时的时间
      maxWait: 60000
      # 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
      timeBetweenEvictionRunsMillis: 60000
      # 配置一个连接在池中最小生存的时间，单位是毫秒
      minEvictableIdleTimeMillis: 300000
      # 配置一个连接在池中最大生存的时间，单位是毫秒
      maxEvictableIdleTimeMillis: 900000
      # 配置检测连接是否有效
      validationQuery: SELECT 1 FROM DUAL
      testWhileIdle: true
      testOnBorrow: false
      testOnReturn: false
      statViewServlet:
        enabled: true
        url-pattern: /monitor/druid/*
      filter:
        stat:
          # 慢SQL记录
          log-slow-sql: true
          slow-sql-millis: 1000
          merge-sql: true
        wall:
          config:
            multi-statement-allow: true

2、数据源配置文件封装类：

package com.ywx.framework.config.properties;

import com.alibaba.druid.pool.DruidDataSource;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

/**
 * 数据源配置文件封装类.
 *
 * @author ywx
 * @date 2019-06-25-10:17
 */
@Configuration
public class DruidProperties 
    @Value("$spring.datasource.druid.initialSize")
    private int initialSize;

    @Value("$spring.datasource.druid.minIdle")
    private int minIdle;

    @Value("$spring.datasource.druid.maxActive")
    private int maxActive;

    @Value("$spring.datasource.druid.maxWait")
    private int maxWait;

    @Value("$spring.datasource.druid.timeBetweenEvictionRunsMillis")
    private int timeBetweenEvictionRunsMillis;

    @Value("$spring.datasource.druid.minEvictableIdleTimeMillis")
    private int minEvictableIdleTimeMillis;

    @Value("$spring.datasource.druid.maxEvictableIdleTimeMillis")
    private int maxEvictableIdleTimeMillis;

    @Value("$spring.datasource.druid.validationQuery")
    private String validationQuery;

    @Value("$spring.datasource.druid.testWhileIdle")
    private boolean testWhileIdle;

    @Value("$spring.datasource.druid.testOnBorrow")
    private boolean testOnBorrow;

    @Value("$spring.datasource.druid.testOnReturn")
    private boolean testOnReturn;

    public DruidDataSource dataSource(DruidDataSource datasource) 
        /** 配置初始化大小、最小、最大 */
        datasource.setInitialSize(initialSize);
        datasource.setMaxActive(maxActive);
        datasource.setMinIdle(minIdle);

        /** 配置获取连接等待超时的时间 */
        datasource.setMaxWait(maxWait);

        /** 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒 */
        datasource.setTimeBetweenEvictionRunsMillis(timeBetweenEvictionRunsMillis);

        /** 配置一个连接在池中最小、最大生存的时间，单位是毫秒 */
        datasource.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis);
        datasource.setMaxEvictableIdleTimeMillis(maxEvictableIdleTimeMillis);

        /**
         * 用来检测连接是否有效的sql，要求是一个查询语句，常用select 'x'。如果validationQuery为null，testOnBorrow、testOnReturn、testWhileIdle都不会起作用。
         */
        datasource.setValidationQuery(validationQuery);
        /** 建议配置为true，不影响性能，并且保证安全性。申请连接的时候检测，如果空闲时间大于timeBetweenEvictionRunsMillis，执行validationQuery检测连接是否有效。 */
        datasource.setTestWhileIdle(testWhileIdle);
        /** 申请连接时执行validationQuery检测连接是否有效，做了这个配置会降低性能。 */
        datasource.setTestOnBorrow(testOnBorrow);
        /** 归还连接时执行validationQuery检测连接是否有效，做了这个配置会降低性能。 */
        datasource.setTestOnReturn(testOnReturn);
        return datasource;
    

3、数据源枚举：

package com.ywx.framework.common.enums;

/**
 * 多数据源配置.
 * @author ywx
 * @date 2019-06-25-10:21
 */
public enum DataSourceType 
    /**
     * 数据源一
     */
    FIRST,

    /**
     * 数据源二
     */
    SECOND,
    /**
     * 数据源三
     */
    THIRD

4、自定义多数据源配置类：

注：如果是读写分离的库，需在第二个数据源上加：@ConditionalOnProperty的配置

package com.ywx.framework.config;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceBuilder;
import com.ywx.framework.common.enums.DataSourceType;
import com.ywx.framework.config.properties.DruidProperties;
import com.ywx.framework.datasource.DynamicDataSource;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import javax.sql.DataSource;
import java.util.HashMap;
import java.util.Map;

/**
 * @author ywx
 * @date 2019-06-25-10:16
 */
@Configuration
public class DruidConfig

    @Bean
    @ConfigurationProperties("spring.datasource.druid.datasource1")
    public DataSource tmsDataSource(DruidProperties druidProperties)
    
        DruidDataSource dataSource = DruidDataSourceBuilder.create().build();
        return druidProperties.dataSource(dataSource);
    

    @Bean
    @ConfigurationProperties("spring.datasource.druid.datasource2")
    //@ConditionalOnProperty(prefix = "spring.datasource.druid.slave", name = "enabled", havingValue = "true")
    public DataSource pacsDataSource(DruidProperties druidProperties)
    
        DruidDataSource dataSource = DruidDataSourceBuilder.create().build();
        return druidProperties.dataSource(dataSource);
    

    @Bean
    @ConfigurationProperties("spring.datasource.druid.datasource3")
    public DataSource ecgDataSource(DruidProperties druidProperties)
    
        DruidDataSource dataSource = DruidDataSourceBuilder.create().build();
        return druidProperties.dataSource(dataSource);
    

    @Bean(name = "dynamicDataSource")
    @Primary
    public DynamicDataSource dataSource(DataSource tmsDataSource, DataSource pacsDataSource,DataSource ecgDataSource)
    
        Map<Object, Object> targetDataSources = new HashMap<>();
        targetDataSources.put(DataSourceType.FIRST.name(), tmsDataSource);
        targetDataSources.put(DataSourceType.SECOND.name(), pacsDataSource);
        targetDataSources.put(DataSourceType.THIRD.name(), ecgDataSource);
        return new DynamicDataSource(tmsDataSource, targetDataSources);
    

5、利用ThreadLocal解决线程安全问题：

package com.ywx.framework.datasource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 数据源线程池.
 * @author ywx
 * @date 2019-06-25-10:19
 */
public class DynamicDataSourceContextHolder 
    public static final Logger log = LoggerFactory.getLogger(DynamicDataSourceContextHolder.class);

    /**
     * 使用ThreadLocal维护变量，ThreadLocal为每个使用该变量的线程提供独立的变量副本，
     * 所以每一个线程都可以独立地改变自己的副本，而不会影响其它线程所对应的副本。
     */
    private static final ThreadLocal<String> CONTEXT_HOLDER = new ThreadLocal<>();

    /**
     * 设置数据源的变量
     */
    public static void setDateSoureType(String dsType) 
        log.info("切换到数据源", dsType);
        CONTEXT_HOLDER.set(dsType);
    

    /**
     * 获得数据源的变量
     */
    public static String getDateSoureType() 
        return CONTEXT_HOLDER.get();
    

    /**
     * 清空数据源变量
     */
    public static void clearDateSoureType() 
        CONTEXT_HOLDER.remove();
    

6、继承AbstractRoutingDataSource定义数据源标识：

package com.ywx.framework.datasource;

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

import javax.sql.DataSource;
import java.util.Map;

/**
 * 动态数据源.
 * @author ywx
 * @date 2019-06-25-10:19
 */
public class DynamicDataSource extends AbstractRoutingDataSource 
    public DynamicDataSource(DataSource defaultTargetDataSource, Map<Object, Object> targetDataSources) 
        super.setDefaultTargetDataSource(defaultTargetDataSource);
        super.setTargetDataSources(targetDataSources);
        super.afterPropertiesSet();
    

    @Override
    protected Object determineCurrentLookupKey() 
        return DynamicDataSourceContextHolder.getDateSoureType();
    

7、定义多数据源注解DataSource：

package com.ywx.framework.common.annotation;

import com.ywx.framework.common.enums.DataSourceType;

import java.lang.annotation.*;

/**
 * 自定义多数据源切换注解.
 * @author ywx
 * @date 2019-08-06-14:17
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface DataSource 
    /**
     * 切换数据源名称.
     */
    public DataSourceType value() default DataSourceType.FIRST;

8、定义DataSource注解切片：

注：在transaction interpter执行之前就把动态数据源配置好，所以在动态数据源的配置的AOP切片上加入Order(1),让其先执行即可。

package com.ywx.framework.aspectj;

import com.ywx.framework.common.annotation.DataSource;
import com.ywx.framework.datasource.DynamicDataSourceContextHolder;
import com.ywx.framework.utils.StringUtils;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;

/**
 * 多数据源切面处理.
 *
 * @author ywx
 * @date 2019-08-06-14:34
 */
@Aspect
@Order(1)
@Component
public class DataSourceAspect 
    protected Logger logger = LoggerFactory.getLogger(getClass());

    @Pointcut("@annotation(com.ywx.framework.common.annotation.DataSource)")
    public void dsPointCut() 

    

    @Around("dsPointCut()")
    public Object around(ProceedingJoinPoint point) throws Throwable 
        MethodSignature signature = (MethodSignature) point.getSignature();

        Method method = signature.getMethod();

        DataSource dataSource = method.getAnnotation(DataSource.class);

        if (StringUtils.isNotNull(dataSource)) 
            logger.info("切换数据源为", dataSource.value().name());
            DynamicDataSourceContextHolder.setDateSoureType(dataSource.value().name());
        

        try 
            return point.proceed();
         finally 
            // 销毁数据源 在执行方法之后
            logger.info("销毁数据源", dataSource.value().name());
            DynamicDataSourceContextHolder.clearDateSoureType();
        
    

以上，完成多数据源的动态切换基本代码。

三、测试-切点注解：

由于我们的动态数据源配置了默认库，所以如果方法是操作默认库的可以不需要注解，如果要操作非默认数据源，我们需要在方法上添加@DataSource("数据源名称")注解，这样就可以利用AOP实现动态切换了。

@Service
public class BizPacsServiceImpl implements IBizPacsService 
    protected Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private PacsMapper pacsMapper;

    // 切换到影像库数据源.
    @DataSource(value = DataSourceType.SECOND)
    public List<PacsApplication> selectPacsApplicationList(PacsApplication pacs) 
        return pacsMapper.selectPacsApplicationList(pacs);