动态切换数据源及事务

chanchaw大约 11 分钟languagejava

概述

网筒项目仓库地址 https://gitee.com/chanchaw/barrel.gitopen in new window 分支 dev 中通过自定义注解的方式制作了多数据源动态切换的功能，但是在 service 方法中使用事务就会报错，看提示是没有成功切换数据源。该方法还有个问题，没有实现动态添加数据源，有新数据源添加到项目需要手动修改源码再重新部署。本文制作的项目通过 AOP 切换数据源解决上面两个问题：1. 动态添加数据源 2. 使用事务。源码在分支 dynamicMult

实现

3个基础配置类

在包 com\xdfsoft\barrel\dynamicds 下的3个文件 TenantDbConfig、PrimaryDbConfig、DatasourceKey 依次是租户数据库配置，对应配置文件的 tenant.datasource，动态添加数据源时根据 request.header.barreluser 查询 userDbMap 中对应的数据库名称，调用 DynamicDataSource # addAndSwitchDataSource 新建数据源并切换主数据库配置，对应配置文件的 primarydb.datasource，该数据库是账套数据库，保存租户对应的数据库名称映射关系，如果多租户项目同时带有微信功能，则该数据库同时为微信后台数据库数据库 key 常量配置类

动态切换

类 com.xdfsoft.barrel.dynamicds.DynamicDataSource 继承 AbstractRoutingDataSource 实现动态切换功能的逻辑

多数据源配置类

com.xdfsoft.barrel.dynamicds.DataSourceConfig 是多数据源配置类，项目启动时创建 primarydb 并获取所有用户和数据源的映射关系（方法 initUserDbMap 获取映射关系并生成名称 userDbMap 的 spring bean ），之后的 AOP 中会通过 HttpServletRequest 获取 barreluser 后从该 Map 中获取用户对应的数据库进行切换。为实现用户切换数据源的功能还制作了刷新该映射 Map 的 API - user/refreshUserDbMap

切换数据源AOP

com.xdfsoft.barrel.aspect.DynamicDataSourceAspect 实现动态切换，在访问租户 controller ，租户 service ，Primary Controller , Primary Service 都会切换数据源，方法 getDataSourceLookupKey 通过读取“用户数据源映射关系”Map 返回数据库名称供切换。看源码注意，之后有扩展新功能，租户账套的 API 都要放在 com.xdfsoft.barrel.controller 下，同理租户的 service 都要放在 com.xdfsoft.barrel.service 下账套（微信）的API 放在 com.xdfsoft.barrel.controllerCloud 同理 service 放在 com.xdfsoft.barrel.serviceCloud 下

源码

TenantDbConfig.java

package com.xdfsoft.barrel.dynamicds;

import com.zaxxer.hikari.HikariDataSource;
import lombok.Builder;
import lombok.Data;
import lombok.experimental.Tolerate;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

import java.text.MessageFormat;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.Collectors;

@Data
@Builder
@Component
@ConfigurationProperties(prefix = "tenant.datasource")
public class TenantDbConfig {
    private String name;
    private String port;
    private String host;
    private String driverClassName;
    private String username;
    private String password;
    private String url;
    private int initializationFailTimeout;
    private int loginTimeout;
    private long idleTimeout;
    private int minimumIdle;
    private long connectionTimeout;
    private long maxLifeTime;
    private int maximumPoolSize;
    private int initialSize;
    private String validationQuery;
    private boolean testOnBorrow;
    private long timeBetweenEvictionRunsMillis;
    private int minEvictableIdleTimeMillis;
    private Map<String, String> jdbcUrlParam;

    @Tolerate
    public TenantDbConfig() {
    }

    /**
     * 获取连接
     *
     * @param database
     * @return
     */
    public String getJdbcUrl(String database) {
        Map<String, String> defaultParam = new HashMap<>();
        defaultParam.put("useUnicode", "true");
        defaultParam.put("serverTimezone", "Asia/Shanghai");
        defaultParam.put("characterEncoding", "UTF-8");
        defaultParam.put("allowMultiQueries", "true");
        String urlTemplate = "jdbc:mysql://{0}:{1}/{2}?{3}";
        if (jdbcUrlParam != null && jdbcUrlParam.size() > 0) {
            defaultParam.putAll(jdbcUrlParam);
        }
        return MessageFormat.format(urlTemplate, this.host, this.port, database, createLinkStringByMap(defaultParam));
    }

    private String createLinkStringByMap(Map<String, String> map) {
        return map.keySet().stream().map(item -> item + "=" + map.get(item)).collect(Collectors.joining("&"));
    }

    /**
     * 创建数据源
     *
     * @param dbConfig
     * @return
     */
    public static HikariDataSource getDataSource(TenantDbConfig dbConfig) {
        synchronized (TenantDbConfig.class) {
            HikariDataSource ds = new HikariDataSource();
            ds.setDriverClassName(dbConfig.getDriverClassName());
            ds.setJdbcUrl(dbConfig.getUrl());
            ds.setUsername(dbConfig.getUsername());
            ds.setPassword(dbConfig.getPassword());
            ds.setValidationTimeout(dbConfig.getConnectionTimeout());
            ds.setConnectionTimeout(dbConfig.getConnectionTimeout());
            ds.setMaximumPoolSize(dbConfig.getMaximumPoolSize());
            ds.setMaxLifetime(dbConfig.getMaxLifeTime());
            ds.setMinimumIdle(dbConfig.getMinimumIdle());
            ds.setConnectionTestQuery(dbConfig.getValidationQuery());
            ds.setIdleTimeout(dbConfig.getIdleTimeout());
            ds.setInitializationFailTimeout(dbConfig.getInitializationFailTimeout());
            return ds;
        }
    }
}

PrimaryDbConfig.java

package com.xdfsoft.barrel.dynamicds;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

@Data
@Component
@ConfigurationProperties(prefix = "primarydb.datasource")
public class PrimaryDbConfig {
    private String driverClassName;
    private String username;
    private String password;
    private String url;
    private int initializationFailTimeout;
    private int minimumIdle;
    private int loginTimeout;
    private long idleTimeout;
    private long connectionTimeout;
    private long maxLifeTime;
    private int maximumPoolSize;
    private int initialSize;
    private String validationQuery;
    private boolean testOnBorrow;
    private long timeBetweenEvictionRunsMillis;
    private int minEvictableIdleTimeMillis;
}

DatasourceKey.java

package com.xdfsoft.barrel.dynamicds;

public class DatasourceKey {
    public static final String PRIMARY_KEY = "primary";
    public static final String CURRENT_DATASOURCE_KEY = "CURRENT_DATASOURCE_KEY";
}

application.yml

server:
  port: 7075
  servlet:
    context-path: /barrelbe

logging:
  level:
    com.xdf.barrel.dao.business: debug
# mybatis generator 的配置
mybatis:
  config-location: classpath:mybatis-config.xml
  mapper-locations: classpath:mappers/*.xml
  type-aliases-package: com.xdfsoft.barrel.model


spring:
  main:
    allow-circular-references: true
  aop:
    auto: true
  mvc:
    pathmatch:
      matching-strategy: ant_path_matcher
  datasource:
    type: com.alibaba.druid.pool.DruidDataSource
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: Asia/Shanghai
#  细码单标题
detail-title: 成品发货细码单

barrel:
  datasource:
    druid:
      # 数据源基本配置
      type: com.alibaba.druid.pool.DruidDataSource
      driverClassName: com.mysql.cj.jdbc.Driver
      url: jdbc:mysql://localhost:3306/barrel?userUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&serverTimezone=Asia/Shanghai
      username: root
      password: 'chanchaw'

      # 数据源其他配置
      # 连接池初始化连接数
      initialSize: 5
      # 连接池最大活跃连接数量
      maxActive: 20
      # 连接池最小活跃数量
      minIdle: 5
      # 获取连接最大等待时间
      maxWait: 60000
      timeBetweenEvictionRunsMillis: 60000
      minEvictableIdleTimeMillis: 300000
      # 打开PSCache,并且制定每个连接上的PSCache 的大小
      poolPreparedStatements: true
      maxPoolPreparedStatementPerConnectionSize: 20
      validationQuery: SELECT 1 FROM DUAL
      validationQueryTimeout: 30000
      # 是否在获得连接后检测期可用性
      testOnBorrow: false
      # 是否在连接放回连接池后检测期可用性
      testOnReturn: false
      # 是否在连接空闲一段时间后检测其可用性
      testWhileIdle: true
      # 配置监控统计拦截的 Filter，去掉后监控界面SQL无法统计，wall 用于防火墙
      filters: stat,wall
      # 通过 connection-properties 属性打开 mergeSql功能；慢SQL记录
      connectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000
      useGlobalDataSourceStat: true

log:
  datasource:
    druid:
      type: com.alibaba.druid.pool.DruidDataSource
      driverClassName: com.mysql.cj.jdbc.Driver
      jdbcUrl: jdbc:mysql://localhost:3306/xms_account?userUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&serverTimezone=Asia/Shanghai
      username: root
      password: 'chanchaw'

      # 下面开始配置 Druid
      # 连接池初始化连接数
      initialSize: 5
      # 连接池最大活跃连接数量
      maxActive: 20
      # 连接池最小活跃数量
      minIdle: 5
      # 获取连接最大等待时间
      maxWait: 60000
      timeBetweenEvictionRunsMillis: 60000
      minEvictableIdleTimeMillis: 300000
      # 打开PSCache,并且制定每个连接上的PSCache 的大小
      poolPreparedStatements: true
      maxPoolPreparedStatementPerConnectionSize: 20
      validationQuery: SELECT 1 FROM DUAL
      validationQueryTimeout: 30000
      # 是否在获得连接后检测期可用性
      testOnBorrow: false
      # 是否在连接放回连接池后检测期可用性
      testOnReturn: false
      # 是否在连接空闲一段时间后检测其可用性
      testWhileIdle: true
      # 配置监控统计拦截的 Filter，去掉后监控界面SQL无法统计，wall 用于防火墙
      filters: stat,wall
      # 通过 connection-properties 属性打开 mergeSql功能；慢SQL记录
      connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
      useGlobalDataSourceStat: true
wechat:
  datasource:
    druid:
      type: com.alibaba.druid.pool.DruidDataSource
      driverClassName: com.mysql.cj.jdbc.Driver
      url: jdbc:mysql://125.91.108.8:3306/wechatbarrel?userUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&serverTimezone=Asia/Shanghai
      username: root
      password: '!BARRELserver20221220@'

      # 下面开始配置 Druid
      # 连接池初始化连接数
      initialSize: 5
      # 连接池最大活跃连接数量
      maxActive: 20
      # 连接池最小活跃数量
      minIdle: 5
      # 获取连接最大等待时间
      maxWait: 60000
      timeBetweenEvictionRunsMillis: 60000
      minEvictableIdleTimeMillis: 300000
      # 打开PSCache,并且制定每个连接上的PSCache 的大小
      poolPreparedStatements: true
      maxPoolPreparedStatementPerConnectionSize: 20
      validationQuery: SELECT 1 FROM DUAL
      validationQueryTimeout: 30000
      # 是否在获得连接后检测期可用性
      testOnBorrow: false
      # 是否在连接放回连接池后检测期可用性
      testOnReturn: false
      # 是否在连接空闲一段时间后检测其可用性
      testWhileIdle: true
      # 配置监控统计拦截的 Filter，去掉后监控界面SQL无法统计，wall 用于防火墙
      filters: stat,wall
      # 通过 connection-properties 属性打开 mergeSql功能；慢SQL记录
      connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
      useGlobalDataSourceStat: true
barrel001:
  datasource:
    druid:
      type: com.alibaba.druid.pool.DruidDataSource
      driverClassName: com.mysql.cj.jdbc.Driver
      url: jdbc:mysql://localhost:3306/barrel001?userUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&serverTimezone=Asia/Shanghai
      username: root
      password: 'chanchaw'
      # 下面开始配置 Druid
      # 连接池初始化连接数
      initialSize: 5
      # 连接池最大活跃连接数量
      maxActive: 20
      # 连接池最小活跃数量
      minIdle: 5
      # 获取连接最大等待时间
      maxWait: 60000
      timeBetweenEvictionRunsMillis: 60000
      minEvictableIdleTimeMillis: 300000
      # 打开PSCache,并且制定每个连接上的PSCache 的大小
      poolPreparedStatements: true
      maxPoolPreparedStatementPerConnectionSize: 20
      validationQuery: SELECT 1 FROM DUAL
      validationQueryTimeout: 30000
      # 是否在获得连接后检测期可用性
      testOnBorrow: false
      # 是否在连接放回连接池后检测期可用性
      testOnReturn: false
      # 是否在连接空闲一段时间后检测其可用性
      testWhileIdle: true
      # 配置监控统计拦截的 Filter，去掉后监控界面SQL无法统计，wall 用于防火墙
      filters: stat,wall
      # 通过 connection-properties 属性打开 mergeSql功能；慢SQL记录
      connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
      useGlobalDataSourceStat: true

barrel002:
  datasource:
    druid:
      type: com.alibaba.druid.pool.DruidDataSource
      driverClassName: com.mysql.cj.jdbc.Driver
      url: jdbc:mysql://localhost:3306/barrel002?userUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&serverTimezone=Asia/Shanghai
      username: root
      password: 'chanchaw'
      # 下面开始配置 Druid
      # 连接池初始化连接数
      initialSize: 5
      # 连接池最大活跃连接数量
      maxActive: 20
      # 连接池最小活跃数量
      minIdle: 5
      # 获取连接最大等待时间
      maxWait: 60000
      timeBetweenEvictionRunsMillis: 60000
      minEvictableIdleTimeMillis: 300000
      # 打开PSCache,并且制定每个连接上的PSCache 的大小
      poolPreparedStatements: true
      maxPoolPreparedStatementPerConnectionSize: 20
      validationQuery: SELECT 1 FROM DUAL
      validationQueryTimeout: 30000
      # 是否在获得连接后检测期可用性
      testOnBorrow: false
      # 是否在连接放回连接池后检测期可用性
      testOnReturn: false
      # 是否在连接空闲一段时间后检测其可用性
      testWhileIdle: true
      # 配置监控统计拦截的 Filter，去掉后监控界面SQL无法统计，wall 用于防火墙
      filters: stat,wall
      # 通过 connection-properties 属性打开 mergeSql功能；慢SQL记录
      connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
      useGlobalDataSourceStat: true

barrel009:
  datasource:
    druid:
      type: com.alibaba.druid.pool.DruidDataSource
      driverClassName: com.mysql.cj.jdbc.Driver
      url: jdbc:mysql://localhost:3306/barrel009?userUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&serverTimezone=Asia/Shanghai
      username: root
      password: 'chanchaw'
      # 下面开始配置 Druid
      # 连接池初始化连接数
      initialSize: 5
      # 连接池最大活跃连接数量
      maxActive: 20
      # 连接池最小活跃数量
      minIdle: 5
      # 获取连接最大等待时间
      maxWait: 60000
      timeBetweenEvictionRunsMillis: 60000
      minEvictableIdleTimeMillis: 300000
      # 打开PSCache,并且制定每个连接上的PSCache 的大小
      poolPreparedStatements: true
      maxPoolPreparedStatementPerConnectionSize: 20
      validationQuery: SELECT 1 FROM DUAL
      validationQueryTimeout: 30000
      # 是否在获得连接后检测期可用性
      testOnBorrow: false
      # 是否在连接放回连接池后检测期可用性
      testOnReturn: false
      # 是否在连接空闲一段时间后检测其可用性
      testWhileIdle: true
      # 配置监控统计拦截的 Filter，去掉后监控界面SQL无法统计，wall 用于防火墙
      filters: stat,wall
      # 通过 connection-properties 属性打开 mergeSql功能；慢SQL记录
      connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
      useGlobalDataSourceStat: true


# ************************** 以下是制作动态数据源时追加的配置 *****************************
tenant:
 datasource:
  driver-class-name: com.mysql.cj.jdbc.Driver
  port: 3306
  host: 127.0.0.1
  username: root
  password: 'chanchaw'
  idle-timeout: 1200000
  minimum-idle: 15
  connection-timeout: 30000
  login-timeout: 100
  max-life-time: 1800000
  maximum-pool-size: 100
  initial-size: 15
  initialization-fail-timeout: 60000
  validation-query: select 1
  test-on-borrow: true
  time-between-eviction-runs-millis: 600000
  min-evictable-idle-time-millis: 1800000

primarydb:
 datasource:
  driver-class-name: com.mysql.cj.jdbc.Driver
  url: jdbc:mysql://125.91.108.8:3306/wechatbarrel?useUnicode=true&serverTimezone=Asia/Shanghai&characterEncoding=UTF-8&allowMultiQueries=false
  username: root
  password: '!BARRELserver20221220@'
  idle-timeout: 1200000
  minimum-idle: 15
  connection-timeout: 30000
  login-timeout: 100
  max-life-time: 1800000
  maximum-pool-size: 100
  initial-size: 15
  validation-query: select 1
  initialization-fail-timeout: 60000
  test-on-borrow: true
  time-between-eviction-runs-millis: 600000
  min-evictable-idle-time-millis: 1800000
# ************************** 上面是制作动态数据源时追加的配置 *****************************

DynamicDataSource.java

package com.xdfsoft.barrel.dynamicds;

import com.zaxxer.hikari.HikariDataSource;
import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class DynamicDataSource extends AbstractRoutingDataSource {
    private static DynamicDataSource instance;
    private static ConcurrentHashMap<Object, Object> dynamicTargetDataSources;
    private static final ThreadLocal<String> CURRENT_DATASOURCE = ThreadLocal.withInitial(() -> DatasourceKey.PRIMARY_KEY);
    private static final ThreadLocal<String> STORAGE_DATASOURCE = ThreadLocal.withInitial(() -> DatasourceKey.PRIMARY_KEY);

    public static DynamicDataSource getInstance() {
        if (instance == null) {
            synchronized (DynamicDataSource.class) {
                if (instance == null) {
                    instance = new DynamicDataSource();
                }
            }
        }
        return instance;
    }

    @Override
    public void setTargetDataSources(Map<Object, Object> targetDataSources) {
        dynamicTargetDataSources = new ConcurrentHashMap<>(targetDataSources);
        super.setTargetDataSources(dynamicTargetDataSources);
    }

    /**
     * 清除DS
     */
    public synchronized void clearDataSource() {
        CURRENT_DATASOURCE.remove();
    }

    /**
     * 设置当前线程用户所属数据库
     *
     * @param key
     */
    public synchronized void setStorageDatasource(String key) {
        STORAGE_DATASOURCE.set(key);
    }

    /***
     * 获取当前保存的数据源
     * @return
     */
    public synchronized String getStorageDatasource() {
        return STORAGE_DATASOURCE.get();
    }

    /**
     * 切换数据源
     *
     * @param key
     * @return
     */
    public synchronized boolean switchDataSource(String key) {
        if (dynamicTargetDataSources.containsKey(key)) {
            CURRENT_DATASOURCE.set(key);
            return true;
        }
        return false;
    }

    /**
     * 获取当前数据源字符
     *
     * @return
     */
    public synchronized String getCurrentDataSourceName() {
        return String.valueOf(this.determineCurrentLookupKey());
    }

    /**
     * 设置默认数据源
     *
     * @return
     */
    public synchronized boolean switchDefault() {
        CURRENT_DATASOURCE.set(DatasourceKey.PRIMARY_KEY);
        return true;
    }

    /**
     * 创建与切换数据源
     *
     * @param config 数据库配置类
     * @return
     */
    public synchronized boolean addAndSwitchDataSource(TenantDbConfig config) {
        if (dynamicTargetDataSources.containsKey(config.getName())) {
            CURRENT_DATASOURCE.set(config.getName());
            return true;
        }
        HikariDataSource dataSource = null;
        try {
            dataSource = TenantDbConfig.getDataSource(config);
            dynamicTargetDataSources.put(config.getName(), dataSource);
            afterPropertiesSet();
            CURRENT_DATASOURCE.set(config.getName());
        } catch (Exception e) {
            if (dataSource != null) {
                dataSource.close();
            }
            logger.error("添加数据源失败,失败信息:" + e.getMessage());
            return false;
        }
        return true;
    }

    /**
     * 获取当前所有的Key
     *
     * @return
     */
    public List<String> getDataSourceKeys() {
        List<String> keys = new ArrayList<>();
        for (Map.Entry<Object, Object> s : dynamicTargetDataSources.entrySet()) {
            keys.add(String.valueOf(s.getKey()));
        }
        return keys;
    }

    @Override
    protected Object determineCurrentLookupKey() {
        logger.debug("当前调用的数据库名称标识:" + CURRENT_DATASOURCE.get());
        return CURRENT_DATASOURCE.get();
    }
}

DataSourceConfig.java

package com.xdfsoft.barrel.dynamicds;

import com.xdfsoft.barrel.dao.UserMapper;
import com.xdfsoft.barrel.model.dto.UserDb;
import com.zaxxer.hikari.HikariDataSource;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.annotation.Order;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;

import javax.sql.DataSource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Configuration
@MapperScan(basePackages = {"com.xdfsoft.barrel.dao"}, sqlSessionFactoryRef = "sqlSessionFactory", sqlSessionTemplateRef = "sqlSessionTemplate")
public class DataSourceConfig {
    @Autowired
    private PrimaryDbConfig primaryDbParam;
    @Autowired
    private UserMapper userMapper;

    /**
     * 数据源
     *
     * @return
     */
    @Bean(value = "dataSource")
    @Primary
    public DynamicDataSource dynamicDataSource() throws Exception {
        DynamicDataSource ds = DynamicDataSource.getInstance();
        Map<Object, Object> map = new HashMap<>();
        HikariDataSource primaryDS = TenantDbConfig.getDataSource(TenantDbConfig.builder()
                .driverClassName(primaryDbParam.getDriverClassName())
                .url(primaryDbParam.getUrl())
                .username(primaryDbParam.getUsername())
                .password(primaryDbParam.getPassword())
                .loginTimeout(primaryDbParam.getLoginTimeout())
                .validationQuery(primaryDbParam.getValidationQuery())
                .idleTimeout(primaryDbParam.getIdleTimeout())
                .minimumIdle(primaryDbParam.getMinimumIdle())
                .timeBetweenEvictionRunsMillis(primaryDbParam.getTimeBetweenEvictionRunsMillis())
                .connectionTimeout(primaryDbParam.getConnectionTimeout())
                .maxLifeTime(primaryDbParam.getMaxLifeTime())
                .maximumPoolSize(primaryDbParam.getMaximumPoolSize())
                .testOnBorrow(primaryDbParam.isTestOnBorrow())
                .initializationFailTimeout(primaryDbParam.getInitializationFailTimeout())
                .build());
        map.put(DatasourceKey.PRIMARY_KEY, primaryDS);
        ds.setTargetDataSources(map);
        return ds;
    }

    @Bean(value="userDbMap")
    public Map<String, UserDb> initUserDbMap(){
        DynamicDataSource.getInstance().switchDefault();
        List<UserDb> userDB = userMapper.getUserDB();
        return userDB.stream().collect(Collectors.toMap(a -> a.getOpenId(), a -> a,(k1,k2) -> k1));
    }

    /**
     * 事务处理器
     *
     * @param dataSource
     * @return
     */
    @Bean("transactionManager")
    @Order(3)
    public DataSourceTransactionManager transactionManager(@Qualifier("dataSource") DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }

    /**
     * 编程式事务管理模板
     * @param transactionManager
     * @return
     */
    @Bean("transactionTemplate")
    @Order(4)
    public TransactionTemplate transactionTemplate(@Qualifier("transactionManager") DataSourceTransactionManager transactionManager) {
        TransactionTemplate template = new TransactionTemplate(transactionManager);
        template.setPropagationBehavior(TransactionTemplate.PROPAGATION_REQUIRES_NEW);
        return template;
    }

    /**
     * @param dataSource
     * @return
     * @throws Exception
     */
    @Bean("sqlSessionFactory")
    public SqlSessionFactory getSqlSessionFactory(@Qualifier("dataSource") DataSource dataSource) throws Exception {
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(dataSource);
        bean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:*mappers/*.xml"));
        return bean.getObject();
    }

    /**
     * @param sqlSessionFactory
     * @return
     */
    @Bean("sqlSessionTemplate")
    public SqlSessionTemplate getSqlSessionTemplate(SqlSessionFactory sqlSessionFactory) {
        return new SqlSessionTemplate(sqlSessionFactory);
    }
}

DynamicDataSourceAspect.java

package com.xdfsoft.barrel.aspect;

import com.xdfsoft.barrel.dynamicds.DatasourceKey;
import com.xdfsoft.barrel.dynamicds.TenantDbConfig;
import com.xdfsoft.barrel.dynamicds.DynamicDataSource;
import com.xdfsoft.barrel.model.dto.UserDb;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.servlet.http.HttpServletRequest;
import java.util.Map;

@Aspect
@Slf4j
@Component
public class DynamicDataSourceAspect {
    @Autowired
    private TenantDbConfig tenantDbConfig;

    @Autowired
    @Qualifier("userDbMap")
    private Map<String, UserDb> userDbMap;

    @Pointcut("execution(* com.xdfsoft.barrel.service.*.*(..))")
    private void localDs() {
    }

    @Pointcut("execution(* com.xdfsoft.barrel.controller.*.*(..))")
    private void localControllerDs() {
    }

    @Pointcut("execution(* com.xdfsoft.barrel.controllerCloud.*.*(..))")
    private void cloudControllerDs() {
    }

    @Pointcut("execution(* com.xdfsoft.barrel.serviceCloud.*.*(..))")
    private void cloudDs() {
    }

    /**
     * 连接每个客户数据源
     */
    @Before("localControllerDs()")
    public void beforeController() {
        DynamicDataSource dataSource = DynamicDataSource.getInstance();
        setDataSource(dataSource);
    }

    /**
     * 连接主数据源
     */
    @Before("cloudControllerDs()")
    public void beforeCloudController() {
        DynamicDataSource dataSource = DynamicDataSource.getInstance();
        dataSource.switchDefault();
    }


    /**
     * 主数据源
     *
     * @param proceedingJoinPoint
     * @throws Throwable
     */
    @Around("cloudDs()")
    public Object aroundCloud(ProceedingJoinPoint proceedingJoinPoint) throws Throwable {
        DynamicDataSource dataSource = DynamicDataSource.getInstance();
        String key = dataSource.getCurrentDataSourceName();
        dataSource.switchDefault();
        Object o = proceedingJoinPoint.proceed();
        dataSource.switchDataSource(key);
        return o;
    }

    /**
     * 注入从数据源
     *
     * @param proceedingJoinPoint
     */
    @Around("localDs()")
    public Object aroundLocal(ProceedingJoinPoint proceedingJoinPoint) throws Throwable {
        DynamicDataSource dataSource = DynamicDataSource.getInstance();
        String currentKey = dataSource.getCurrentDataSourceName();
        setDataSource(dataSource);
        Object o = proceedingJoinPoint.proceed();
        dataSource.switchDataSource(currentKey);
        return o;
    }

    private void setDataSource(DynamicDataSource dataSource) {
        String key = getDataSourceLookupKey();
        dataSource.setStorageDatasource(key);

        if (!dataSource.switchDataSource(key)) {
            tenantDbConfig.setName(key);
            tenantDbConfig.setUrl(tenantDbConfig.getJdbcUrl(key));
            if (dataSource.addAndSwitchDataSource(tenantDbConfig)) {
                log.debug("添加并切换到数据库：{}", key);
            }
        }
    }

    /**
     * 抛出异常后操作
     */
    @AfterThrowing("localDs()||cloudDs()")
    public void throwExc() {
        DynamicDataSource dataSource = DynamicDataSource.getInstance();
        dataSource.clearDataSource();
    }

    private String getDataSourceLookupKey() {
        ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
        if(attributes == null) return DatasourceKey.PRIMARY_KEY;
        HttpServletRequest request = attributes.getRequest();
        String openId = request.getHeader("barreluser");
        if(openId == null || openId.length() == 0) return DatasourceKey.PRIMARY_KEY;
        String dbName = userDbMap.get(openId).getDbName();
        return dbName;
    }
}

动态切换数据源及事务

# 概述

# 实现

# 3个基础配置类

# 动态切换

# 多数据源配置类

# 切换数据源AOP

# 源码

# TenantDbConfig.java

# PrimaryDbConfig.java

# DatasourceKey.java

# application.yml

# DynamicDataSource.java

# DataSourceConfig.java

# DynamicDataSourceAspect.java

概述

实现