[Java知识库]基于注解和Aop实现多数据源动态切换

基于注解和Aop实现多数据源动态切换

1.前置说明

? 想要自定义动态数据源切换，得先了解一个类 AbstractRoutingDataSource：

AbstractRoutingDataSource 是在 Spring2.0.1 中引入的, 该类充当了 DataSource 的路由中介，它能够在运行时, 根据 key 值来动态切换到真正的 DataSource 上。

AbstractRoutingDataSource实现了InitializingBean接口，AbstractRoutingDataSource是一个抽象类，我们可以通过子类继承的方式重写determineCurrentLookupKey抽象方法返回key值，而这里的key值其实就是AbstractRoutingDataSource的属性resolvedDataSources这个map中的某个key,

比如 “master”,“salve” ；map的值就是 DataSource数据源对象。

? 大致的用法就是你提前准备好各种数据源，存入到一个 Map 中，Map 的 key 就是这个数据源的名字，Map 的 value 就是这个具体的数据源，然后再把这个 Map 配置到 AbstractRoutingDataSource 中，最后，每次执行数据库查询的时候，拿一个 key 出来，AbstractRoutingDataSource 会找到具体的数据源去执行这次数据库操作。

2.创建项目

首先我们创建一个 Spring Boot 项目，引入 Web、MyBatis 以及 MySQL 依赖，项目创建成功之后，再手动加入 Druid 和 AOP 依赖，如下：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>druid-spring-boot-starter</artifactId>
    <version>1.2.9</version>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>

3.配置文件

YAML 配置不像 properties 配置可以通过 @PropertySource 注解加载自定义的配置文件，YAML 配置没有类似的加载机制。这里利用Spring Boot 的 profile 机制来加载这个自定义的 application-druid.yaml 配置文件，具体做法就是在 application.yaml 中加一行配置，如下：

application.yml

spring:
  profiles:
    active: druid

application-druid.yml

# 数据源配置
spring:
  datasource:
    type: com.alibaba.druid.pool.DruidDataSource
    driverClassName: com.mysql.cj.jdbc.Driver
    ds:
      # 主库数据源，默认 master 不能变
      master:
        url: jdbc:mysql://127.0.0.1:3306/chat?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=Asia/Shanghai
        username: root
        password: root
      # 从库数据源
      slave:
        url: jdbc:mysql://127.0.0.1:3306/llp?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=Asia/Shanghai
        username: root
        password: root
    # 初始连接数
    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
    druid:
      webStatFilter:
        enabled: true
      statViewServlet:
        enabled: true
        # 设置白名单，不填则允许所有访问
        allow:
        url-pattern: /druid/*
        # 控制台管理用户名和密码
        login-username: admin
        login-password: admin
      filter:
        stat:
          enabled: true
          # 慢SQL记录
          log-slow-sql: true
          slow-sql-millis: 5000
          merge-sql: true
        wall:
          config:
            multi-statement-allow: true

#配置mybatis
mybatis:
  configuration:
    map-underscore-to-camel-case: true
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
  mapper-locations: classpath*:mapper/*.xml

接下来我们还需要提供一个配置类，将这个配置文件的内容加载到配置类中，如下：

/**
 * spring容器启动时将DruidProperties 作为bean对象注入到容器中
 *
 * @ConfigurationProperties 会读取application配置文件，set给DruidProperties属性
 */
@Data
@Component
@ConfigurationProperties(prefix = "spring.datasource")
public class DruidProperties {
    /**
     * 初始连接数
     */
    private int initialSize;

    /**
     * 最小连接池数量
     */
    private int minIdle;

    /**
     * 最大连接池数量
     */
    private int maxActive;

    /**
     * 配置获取连接等待超时的时间
     */
    private int maxWait;

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

    /**
     * 配置一个连接在池中最小生存的时间，单位是毫秒
     */
    private int minEvictableIdleTimeMillis;

    /**
     * 配置一个连接在池中最大生存的时间，单位是毫秒
     */
    private int maxEvictableIdleTimeMillis;

    /**
     * 配置检测连接是否有效
     */
    private String validationQuery;

    /**
     * 建议配置为true，不影响性能，并且保证安全性。申请连接的时候检测，如果空闲时间大于timeBetweenEvictionRunsMillis，执行validationQuery检测连接是否有效。
     */
    private boolean testWhileIdle;

    /**
     * 申请连接时执行validationQuery检测连接是否有效，做了这个配置会降低性能。
     */
    private boolean testOnBorrow;

    /**
     * 归还连接时执行validationQuery检测连接是否有效，做了这个配置会降低性能。
     */
    private boolean testOnReturn;

    /**
     * key:数据源名称
     * value:数据源配置信息
     */
    private Map<String, Map<String, String>> ds;

    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;
    }

}

配置的多个数据源将之读取到了一个名为 ds 的 Map 中，key:数据源名称（master、salve）value: 数据源配置信息， 将来就根据这个 Map 中的数据来构造数据源。

4.加载数据源

接下来我们要根据配置文件来加载数据源。加载方式如下：

定义DynamicDataSourceProvider接口

/**
 * 加载所有的数据源
 */
public interface DynamicDataSourceProvider {
    String DEFAULT_DATASOURCE = "master";
    /**
     * 加载所有的数据源
     * @return
     */
    Map<String, DataSource> loadDataSources();
}

YamlDynamicDataSourceProvider实现类

@Configuration
@EnableConfigurationProperties(DruidProperties.class)
public class YamlDynamicDataSourceProvider implements DynamicDataSourceProvider {
    @Autowired
    DruidProperties druidProperties;

    /**
     * 加载配置文件中的数据源
     * @return
     */
    @Override
    public Map<String, DataSource> loadDataSources() {
        //创建一个和配置文件数据源相同大小的map
        Map<String, DataSource> ds = new HashMap<>(druidProperties.getDs().size());
        try {
            //从配置类中获取数据源map, key：数据源名称(master,salve) value：数据源配置信息（url,name,password,driver）
            Map<String, Map<String, String>> map = druidProperties.getDs();
            //获取map中的所有的key(master,slave)
            Set<String> keySet = map.keySet();
            for (String s : keySet) {
                //根据数据源配置信息创建DruidDataSource对象
                DruidDataSource dataSource = (DruidDataSource) DruidDataSourceFactory.createDataSource(map.get(s));
                //将创建的dataSource数据源对象保存到map中， key:数据源名称、 value:dataSource数据源对象
                // druidProperties.dataSource(dataSource) 配置数据源信息，最大链接数、空闲时间等
                ds.put(s, druidProperties.dataSource(dataSource));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return ds;
    }
}

加载的核心工作在 YamlDynamicDataSourceProvider 类中完成的。该类中有一个 loadDataSources 方法表示读取所有的数据源对象。数据源的相关属性都在 druidProperties 对象中，我们先根据基本的数据库连接信息创建一个 DataSource 对象，然后再调用 druidProperties#dataSource 方法为这些数据源连接池配置其他的属性（最大连接数、最小空闲数等），最后，以 key-value 的形式将数据源存入一个 Map 集合中，每一个数据源的 key 就是你在 YAML 中配置的数据源名称。

5. 数据源切换

对于当前数据库操作使用哪个数据源？我们有很多种不同的设置方案，当然最为省事的办法是把当前使用的数据源信息存入到 ThreadLocal 中，ThreadLocal 的特点，简单说就是在哪个线程中存入的数据，在哪个线程才能取出来，换一个线程就取不出来了，这样可以确保多线程环境下的数据安全。

1.线程工具类

public class DynamicDataSourceContextHolder {
    public static final Logger log = LoggerFactory.getLogger(DynamicDataSourceContextHolder.class);

    /**
     * 使用ThreadLocal维护变量，ThreadLocal为每个使用该变量的线程提供独立的变量副本，
     * 所以每一个线程都可以独立地改变自己的副本，而不会影响其它线程所对应的副本。
     * 后续在aop中在执行目标方法之前，我们从注解获取需要执行哪一个数据源（dataType:slave、master）并保存到CONTEXT_HOLDER这个threadLocal对象中
     * dao层在获取连接时会执行AbstractRoutingDataSource类的determineTargetDataSource方法，而改方法调用其子类的
     * determineCurrentLookupKey方法来获取dataSoucrceType，最终决定时用哪一个数据源对象执行底层的jdbc对数据库的操作
     */
    private static final ThreadLocal<String> CONTEXT_HOLDER = new ThreadLocal<>();

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

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

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

2.定义一个标记数据源的注解

/**
 * @author llp
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD, ElementType.TYPE})
public @interface DataSource {

    String dataSourceName() default DynamicDataSourceProvider.DEFAULT_DATASOURCE;

    @AliasFor("dataSourceName")
    String value() default DynamicDataSourceProvider.DEFAULT_DATASOURCE;
}

3.AOP实现类

这个注解将来加在 Service 层的方法上，使用该注解的时候，需要指定一个数据源名称，不指定的话，默认就使用 master 作为数据源。

我们还需要通过 AOP 来解析当前的自定义注解，如下：

/**
 * order默认最低优先级，Integer.MAX_VALUE
 * 值越小优先级越高
 */
@Aspect
@Order(1)
@Component
public class DataSourceAspect {
    @Pointcut("@annotation(com.llp.dynamicdatasource.annotation.DataSource)"
            + "|| @within(com.llp.dynamicdatasource.annotation.DataSource)")
    public void dsPc() {

    }

    @Around("dsPc()")
    public Object around(ProceedingJoinPoint point) throws Throwable {
        //获取目标方法的@DataSource 数据源注解
        DataSource dataSource = getDataSource(point);
        //判断@DataSource注解对象是否为空，如果不为空则添加到threadLocal中（当前线程）
        if (Objects.nonNull(dataSource)) {
            DynamicDataSourceContextHolder.setDataSourceType(dataSource.dataSourceName());
        }

        try {
            //执行目标方法
            return point.proceed();
        } finally {
            // 这里使用try finally，不论目标方法是否执行成功，都需要销毁当前线程数据源 在执行方法之后
            DynamicDataSourceContextHolder.clearDataSourceType();
        }
    }

    /**
     * 获取需要切换的数据源
     */
    public DataSource getDataSource(ProceedingJoinPoint point) {
        //获取方法签名对象
        MethodSignature signature = (MethodSignature) point.getSignature();
        //AnnotationUtils.findAnnotation
        //简单理解就是先从目标方法上去获取@DataSource这个注解，没有找到在去从类上获取；因此如果方法和类上都含有这个注解则方法的优先级时高于类的
        DataSource dataSource = AnnotationUtils.findAnnotation(signature.getMethod(), DataSource.class);
        return dataSource;
    }
}

1. 首先，我们在 dsPc() 方法上定义了切点，我们拦截下所有带有 @DataSource 注解的方法，同时由于该注解也可以加在类上，如果该注解加在类上，就表示类中的所有方法都使用该数据源。
2. 接下来我们定义了一个环绕通知，首先根据当前的切点，调用 getDataSource 方法获取到 @DataSource 注解，这个注解可能来自方法上也可能来自类上，方法上的优先级高于类上的优先级。如果拿到的注解不为空，则我们在 DynamicDataSourceContextHolder 中设置当前的数据源名称，设置完成后进行方法的调用；如果拿到的注解为空，那么就直接进行方法的调用，不再设置数据源了（将来会自动使用默认的数据源）。最后记得方法调用完成后，从 ThreadLocal 中移除数据源。

6.定义动态数据源

/**
 * 定义动态数据源
 */
public class DynamicDataSource extends AbstractRoutingDataSource {

    DynamicDataSourceProvider dynamicDataSourceProvider;

    /**
     *
     * @param dynamicDataSourceProvider
     */
    public DynamicDataSource(DynamicDataSourceProvider dynamicDataSourceProvider) {
        this.dynamicDataSourceProvider = dynamicDataSourceProvider;
        //目标数据源
        Map<Object, Object> targetDataSources = new HashMap<>(dynamicDataSourceProvider.loadDataSources());
        //目标数据源
        super.setTargetDataSources(targetDataSources);
        //指定默认的数据源
        super.setDefaultTargetDataSource(dynamicDataSourceProvider.loadDataSources().get(DynamicDataSourceProvider.DEFAULT_DATASOURCE));
        //AbstractRoutingDataSource实现了InitializingBean即可，创建实例之后会执行afterPropertiesSet方法
//        super.afterPropertiesSet();
    }

    /**
     * 	protected DataSource determineTargetDataSource() {
     * 		Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
     * 		Object lookupKey = determineCurrentLookupKey();
     * 	    resolvedDataSources 这个map包含了所有数据源信息，根据key获取指定的数据源对象
     * 		DataSource dataSource = this.resolvedDataSources.get(lookupKey);
     * 		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;* 	}
     *
     *
     * @return
     */
    @Override
    protected Object determineCurrentLookupKey() {
        /**
         * 在aop中我们对被@DataSource注解修饰的目标方法/类 声明了数据源类型（master/salve）
         * 当目标方法的dao层去获取连接时，AbstractRoutingDataSource类的determineTargetDataSource 方法，而改方法调用了determineCurrentLookupKey 抽象方法
         * 我们对这个方法进行了重新，进而实现动态的获取数据源对象
         *
         *  如果dao层时使用mybatis则，时根据mapperProxy代理对象去执行目标方法，通过defaultSqlSession去执行crud,底层去获取connection连接时会调用 determineTargetDataSource方法
         */
        String dataSourceType = DynamicDataSourceContextHolder.getDataSourceType();
        return dataSourceType;
    }
}

这就是之前所说的 AbstractRoutingDataSource 了，该类有一个方法名为 determineCurrentLookupKey，当需要使用数据源的时候，系统会自动调用该方法，获取当前数据源的标记，如 master 或者 slave 或者其他，拿到标记之后，就可以据此获取到一个数据源了。

当我们配置 DynamicDataSource 的时候，需要配置两个关键的参数，一个是 setTargetDataSources，这个就是当前所有的数据源，把当前所有的数据源都告诉给 AbstractRoutingDataSource，这些数据源都是 key-value 的形式（将来根据 determineCurrentLookupKey 方法返回的 key 就可以获取到具体的数据源了）；另一个方法是 setDefaultTargetDataSource，这个就是默认的数据源，当我们执行一个数据库操作的时候，如果没有指定数据源（例如 Service 层的方法没有加 @DataSource 注解），那么默认就使用这个数据源。

将这个 bean 注册到 Spring 容器中：

@Configuration
public class DruidAutoConfiguration {

    @Autowired
    DynamicDataSourceProvider dynamicDataSourceProvider;

    /**
     * spring容器启动时将DynamicDataSource  bean对象注入到容器中
     * @return
     */
    @Bean
    DynamicDataSource dynamicDataSource() {
        return new DynamicDataSource(dynamicDataSourceProvider);
    }

    /**
     * 去除数据源监控页面的广告
     * 去除 Druid 监控页面的阿里广告
     * @param properties
     * @return
     */
    @Bean
    @ConditionalOnProperty(name = "spring.datasource.druid.statViewServlet.enabled", havingValue = "true")
    public FilterRegistrationBean removeDruidFilterRegistrationBean(DruidStatProperties properties) {
        // 获取web监控页面的参数
        DruidStatProperties.StatViewServlet config = properties.getStatViewServlet();
        // 提取common.js的配置路径
        String pattern = config.getUrlPattern() != null ? config.getUrlPattern() : "/druid/*";
        String commonJsPattern = pattern.replaceAll("\\*", "js/common.js");
        // 创建filter进行过滤
        Filter filter = new Filter() {
            @Override
            public void init(javax.servlet.FilterConfig filterConfig) throws ServletException {
            }

            @Override
            public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
                    throws IOException, ServletException {
                String text = Utils.readFromResource("support/http/resources/js/common.js");
                text = text.replace("this.buildFooter();", "");
                response.getWriter().write(text);
            }

            @Override
            public void destroy() {
            }
        };
        FilterRegistrationBean registrationBean = new FilterRegistrationBean();
        registrationBean.setFilter(filter);
        registrationBean.addUrlPatterns(commonJsPattern);
        return registrationBean;
    }
}

7.测试结果

创建海量数据的sql

#创建海量数据的sql脚本

CREATE TABLE dept( /*部门表*/
deptno MEDIUMINT   UNSIGNED  NOT NULL  DEFAULT 0,
dname VARCHAR(20)  NOT NULL  DEFAULT "",
loc VARCHAR(13) NOT NULL DEFAULT ""
) ;

#创建表EMP雇员
CREATE TABLE emp
(empno  MEDIUMINT UNSIGNED  NOT NULL  DEFAULT 0, /*编号*/
ename VARCHAR(20) NOT NULL DEFAULT "", /*名字*/
job VARCHAR(9) NOT NULL DEFAULT "",/*工作*/
mgr MEDIUMINT UNSIGNED NOT NULL DEFAULT 0,/*上级编号*/
hiredate DATE NOT NULL,/*入职时间*/
sal DECIMAL(7,2)  NOT NULL,/*薪水*/
comm DECIMAL(7,2) NOT NULL,/*红利*/
deptno MEDIUMINT UNSIGNED NOT NULL DEFAULT 0 /*部门编号*/
) ;

#工资级别表
CREATE TABLE salgrade
(
grade MEDIUMINT UNSIGNED NOT NULL DEFAULT 0,
losal DECIMAL(17,2)  NOT NULL,
hisal DECIMAL(17,2)  NOT NULL
);

#测试数据
INSERT INTO salgrade VALUES (1,700,1200);
INSERT INTO salgrade VALUES (2,1201,1400);
INSERT INTO salgrade VALUES (3,1401,2000);
INSERT INTO salgrade VALUES (4,2001,3000);
INSERT INTO salgrade VALUES (5,3001,9999);

delimiter $$

#创建一个函数，名字 rand_string，可以随机返回我指定的个数字符串
create function rand_string(n INT)
returns varchar(255) #该函数会返回一个字符串
begin
#定义了一个变量 chars_str， 类型  varchar(100)
#默认给 chars_str 初始值   'abcdefghijklmnopqrstuvwxyzABCDEFJHIJKLMNOPQRSTUVWXYZ'
 declare chars_str varchar(100) default
   'abcdefghijklmnopqrstuvwxyzABCDEFJHIJKLMNOPQRSTUVWXYZ'; 
 declare return_str varchar(255) default '';
 declare i int default 0; 
 while i < n do
    # concat 函数 : 连接函数mysql函数
   set return_str =concat(return_str,substring(chars_str,floor(1+rand()*52),1));
   set i = i + 1;
   end while;
  return return_str;
  end $$


 #这里我们又自定了一个函数,返回一个随机的部门号
create function rand_num( )
returns int(5)
begin
declare i int default 0;
set i = floor(10+rand()*500);
return i;
end $$

 #创建一个存储过程， 可以添加雇员
create procedure insert_emp(in start int(10),in max_num int(10))
begin
declare i int default 0;
#set autocommit =0 把autocommit设置成0
 #autocommit = 0 含义: 不要自动提交
 set autocommit = 0; #默认不提交sql语句
 repeat
 set i = i + 1;
 #通过前面写的函数随机产生字符串和部门编号，然后加入到emp表
 insert into emp values ((start+i) ,rand_string(6),'SALESMAN',0001,curdate(),2000,400,rand_num());
  until i = max_num
 end repeat;
 #commit整体提交所有sql语句，提高效率
   commit;
 end $$

 #添加8000000数据
# id从param1+1开始 创建param2条数据
call insert_emp(0,8000000)$$

#命令结束符，再重新设置为;
delimiter ;

创建实体类—emp.java

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Emp {
    private Integer empno;
    private String ename;
    private String job;
    private Integer mgr;
    private Date hiredate;
    private BigDecimal sal;
    private BigDecimal comm;
    private Integer deptno;

}

EmpService.java

@Service
public class EmpService {

    @Autowired
    EmpMapper empMapper;
	
    //主库
    @DataSource("master")
    public Integer master() {
        return empMapper.count();
    }
	
    //从库
    @DataSource("slave")
    public Integer slave() {
        return empMapper.count();
    }
}

EmpMapper.java

@Mapper
public interface EmpMapper {
    @Select("select count(*) from emp")
    Integer count();
}

修改启动类，添加包扫描

@MapperScan(basePackages = {"com.llp.dynamicdatasource.dao"})
@SpringBootApplication
public class DynamicDatasourceApplication {

    public static void main(String[] args) {
        SpringApplication.run(DynamicDatasourceApplication.class, args);
    }
}

测试类

@SpringBootTest
class DynamicDatasourceApplicationTests {

    @Autowired
    private EmpService empService;

    @Test
    void contextLoads() {
        Integer master = empService.master();
        System.out.println(master);
        Integer slave = empService.slave();
        System.out.println(slave);
    }
}