ShardingSphere多租户分库分表动态加载、切换数据源

1.需求说明要实现多租户动态加载、切换数据源，并进行分表操作。表结构参考：CREATE TABLE `tenant_info` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`TENANT_ID` varchar(255) DEFAULT NULL COMMENT '租户id',`TENANT_NAME` varchar(255) DEFAULT NULL CO

拽风追梦

5111人浏览 · 2021-10-21 11:26:20

拽风追梦 · 2021-10-21 11:26:20 发布

1.需求说明

要实现多租户动态加载、切换数据源，并进行分表操作。

表结构参考：

CREATE TABLE `tenant_info` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`TENANT_ID` varchar(255) DEFAULT NULL COMMENT '租户id',
`TENANT_NAME` varchar(255) DEFAULT NULL COMMENT '租户名称',
`DATASOURCE_URL` varchar(255) DEFAULT NULL COMMENT '数据源url',
`DATASOURCE_USERNAME` varchar(255) DEFAULT NULL COMMENT '数据源用户名',
`DATASOURCE_PASSWORD` varchar(255) DEFAULT NULL COMMENT '数据源密码',
`DATASOURCE_DRIVER` varchar(255) DEFAULT NULL COMMENT '数据源驱动',
`SYSTEM_ACCOUNT` varchar(255) DEFAULT NULL COMMENT '系统账号',
`SYSTEM_PASSWORD` varchar(255) DEFAULT NULL COMMENT '账号密码',
`SYSTEM_PROJECT` varchar(255) DEFAULT NULL COMMENT '系统PROJECT',
`STATUS` tinyint(1) DEFAULT NULL COMMENT '是否启用（1是0否）',
`CREATE_TIME` datetime DEFAULT NULL COMMENT '创建时间',
`UPDATE_TIME` datetime DEFAULT NULL COMMENT '更新时间',
`type` int(255) DEFAULT NULL,
PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8 COMMENT='多租户表';

2.相关依赖

<dependency>
    <groupId>com.github.pagehelper</groupId>
    <artifactId>pagehelper-spring-boot-starter</artifactId>
    <version>1.2.12</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>mybatis-plus-boot-starter</artifactId>
    <version>3.2.0</version>
</dependency>

3.MybatisConfig文件

package com.ctl.mes.service.execute.config;

import com.alibaba.druid.filter.Filter;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.wall.WallConfig;
import com.alibaba.druid.wall.WallFilter;
import com.baomidou.mybatisplus.core.parser.ISqlParser;
import com.baomidou.mybatisplus.extension.parsers.BlockAttackSqlParser;
import com.baomidou.mybatisplus.extension.plugins.PaginationInterceptor;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.plugin.Interceptor;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Qualifier;
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 org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement;

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

/**
 * 初始化配置，提供默认数据源
 * @author  wyf
 * @date  2021/10/9 11:31
 */
@Slf4j
@EnableTransactionManagement
@Configuration
@MapperScan("com.ctl.mes.service.execute.mapper")
public class MybatisConfig {

    @Primary
    @Bean("master")
    @ConfigurationProperties(prefix = "spring.shardingsphere.datasource")
    public DataSource master() {
        DruidDataSource druidDataSource = new DruidDataSource();//如果不是分库分表的话，可以new DataSource()。不然会找不到初始的数据源
        List<Filter> filterList = new ArrayList<>();
        filterList.add(wallFilter());
        druidDataSource.setProxyFilters(filterList);
        System.out.println("初始化yml配置的默认数据库...");
        return druidDataSource;
    }
/*
    @Bean("slave")
    @ConfigurationProperties(prefix = "spring.datasource.slave")
    public DataSource slave() {
        return new DruidDataSource();
    }*/

    @Bean("dynamicDataSource")
    public DataSource dynamicDataSource() {
        Map<Object, Object> map = new HashMap<>(1);
        map.put("master", master());

        return new DynamicDataSource(master(), map);
    }

    @Bean(name="ibatiesConfig")
    @ConfigurationProperties(prefix = "mybatis.configuration")
    public org.apache.ibatis.session.Configuration ibatiesConfig(){
        return new org.apache.ibatis.session.Configuration();
    }

    @Bean
    public WallFilter wallFilter() {
        WallFilter wallFilter = new WallFilter();
        wallFilter.setConfig(wallConfig());
        return wallFilter;
    }

    @Bean
    public WallConfig wallConfig() {
        WallConfig config = new WallConfig();
        //允许一次执行多条语句
        config.setMultiStatementAllow(true);
        //允许非基本语句的其他语句
        config.setNoneBaseStatementAllow(true);
        //允许包含注释的 SQL 执行
        config.setCommentAllow(true);
        return config;
    }

    @Bean
    public SqlSessionFactoryBean sqlSessionFactoryBean(@Qualifier("dynamicDataSource") DataSource dynamicDataSource, @Qualifier("ibatiesConfig") org.apache.ibatis.session.Configuration ibatiesConfig) throws Exception {
        SqlSessionFactoryBean sessionFactory = new SqlSessionFactoryBean();
        //重点，使分页插件生效
        Interceptor[] plugins = new Interceptor[1];
        plugins[0] = paginationInterceptor();
        sessionFactory.setPlugins(plugins);
        // 配置数据源，此处配置为关键配置，如果没有将 dynamicDataSource作为数据源则不能实现切换
        sessionFactory.setDataSource(dynamicDataSource);
        // 扫描Model
        sessionFactory.setTypeAliasesPackage("com.ctl.mes.service.execute.domain");
        // 扫描映射文件
        sessionFactory.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath*:mapper/*.xml"));
        sessionFactory.setConfiguration(ibatiesConfig);
        return sessionFactory;
    }

    @Bean
    public PlatformTransactionManager transactionManager() {
        // 配置事务管理, 使用事务时在方法头部添加@Transactional注解即可
        return new DataSourceTransactionManager(dynamicDataSource());
    }

    /**
     * 加载分页插件
     *
     * @return
     */
    @Bean
    public PaginationInterceptor paginationInterceptor() {
        PaginationInterceptor paginationInterceptor = new PaginationInterceptor();

        List<ISqlParser> sqlParserList = new ArrayList<>();
        // 攻击 SQL 阻断解析器、加入解析链
        sqlParserList.add(new BlockAttackSqlParser());
        paginationInterceptor.setSqlParserList(sqlParserList);
        return paginationInterceptor;
    }
}

3.DynamicDataSource文件

package com.ctl.mes.service.execute.config;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.util.StringUtils;
import com.ctl.mes.service.execute.domain.TenantInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.api.config.sharding.ShardingRuleConfiguration;
import org.apache.shardingsphere.api.config.sharding.TableRuleConfiguration;
import org.apache.shardingsphere.api.config.sharding.strategy.StandardShardingStrategyConfiguration;
import org.apache.shardingsphere.shardingjdbc.api.ShardingDataSourceFactory;
import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;

import javax.sql.DataSource;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * （切换数据源必须在调用service之前进行，也就是开启事务之前）
 * 动态数据源实现类
 * @author  weiyongfu
 * @date  2021/10/9 13:31
 */
@Slf4j
public class DynamicDataSource extends AbstractRoutingDataSource {

    private Map<Object, Object> dynamicTargetDataSources;

    /**
     * 决定使用哪个数据源之前需要把多个数据源的信息以及默认数据源信息配置好
     *
     * @param defaultTargetDataSource 默认数据源
     * @param targetDataSources       目标数据源
     */
    DynamicDataSource(DataSource defaultTargetDataSource, Map<Object, Object> targetDataSources) {
        setDefaultTargetDataSource(defaultTargetDataSource);
        setTargetDataSources(targetDataSources);
        this.dynamicTargetDataSources = targetDataSources;
        super.afterPropertiesSet();
    }

    public DynamicDataSource() {
    }

    /**
     * 如果不希望数据源在启动配置时就加载好，可以定制这个方法，从任何你希望的地方读取并返回数据源
     * 比如从数据库、文件、外部接口等读取数据源信息，并最终返回一个DataSource实现类对象即可
     */
    @Override
    protected DataSource determineTargetDataSource() {
        return super.determineTargetDataSource();
    }

    /**
     * 如果希望所有数据源在启动配置时就加载好，这里通过设置数据源Key值来切换数据，定制这个方法
     *
     * 实现数据源切换要扩展的方法，该方法的返回值就是项目中所要用的DataSource的key值，
     * 拿到该key后就可以在resolvedDataSource中取出对应的DataSource，如果key找不到对应的DataSource就使用默认的数据源。
     */
    @Override
    protected Object determineCurrentLookupKey() {
        String dataSourceName = DynamicDataSourceContextHolder.getDataSourceKey();
        if (!StringUtils.isEmpty(dataSourceName)) {
            Map<Object, Object> dynamicTargetDataSources2 = this.dynamicTargetDataSources;
            if (dynamicTargetDataSources2.containsKey(dataSourceName)) {
                log.info("当前数据源为：" + dataSourceName);
            } else {
                log.info("不存在的数据源："+ dataSourceName);
            }
        } else {
            log.info("当前数据源为：默认数据源");
        }
        return dataSourceName;
    }

    /**
     * 设置默认数据源
     * @param defaultDataSource Object
     */
    @Override
    public void setDefaultTargetDataSource(Object defaultDataSource) {
        super.setDefaultTargetDataSource(defaultDataSource);
    }

    /**
     * 设置数据源
     * @param dataSources Map<Object, Object>
     */
    @Override
    public void setTargetDataSources(Map<Object, Object> dataSources) {

        super.setTargetDataSources(dataSources);

        this.dynamicTargetDataSources = dataSources;

        // 将数据源的 key 放到数据源上下文的 key 集合中，用于切换时判断数据源是否有效
        DynamicDataSourceContextHolder.addDataSourceKeys(dataSources.keySet());
    }

    void setDataSources(TenantInfo tenantInfo) {
        String tenantId = tenantInfo.getTenantId();
        String dataSourceDriver = tenantInfo.getDataSourceDriver();
        String url = tenantInfo.getDataSourceUrl();
        String username = tenantInfo.getDataSourceUsername();
        String password = tenantInfo.getDataSourcePassword();
        try {
            // 排除连接不上的错误
            Class.forName(dataSourceDriver);
            // 相当于连接数据库
            DriverManager.getConnection(url, username, password);

            DruidDataSource dataSource = new DruidDataSource();
            dataSource.setDriverClassName(dataSourceDriver);
            dataSource.setUrl(url);
            dataSource.setUsername(username);
            dataSource.setPassword(password);
            this.dynamicTargetDataSources.put(tenantId, buildShardDatasources(dataSource,tenantId));
            setTargetDataSources(this.dynamicTargetDataSources);
            super.afterPropertiesSet();
            log.info("数据源初始化成功!租户标识:" + tenantId);
        } catch (Exception e) {
            throw new IllegalStateException("[" + tenantId + "]" + "：数据源连接不上， 可能是连接参数有误！");
        }
    }


    public DataSource buildShardDatasources(DataSource dataSource,String type) {
        // 配置多数据源
        Map<String, DataSource> dsMap = new HashMap<>();
        dsMap.put(type, dataSource);
        //tableRuleConfiguration.setDatabaseShardingStrategyConfig();分库规则，要分库的自己加进去
        //act_test_data表分片规则添加
        /*TableRuleConfiguration manuHisTableRuleConfig = new TableRuleConfiguration("act_test_data", type+".act_test_data_$->{2020..2023}");
        StandardShardingStrategyConfiguration manu_workorder = new StandardShardingStrategyConfiguration("testtime", new TableShardingAlgorithm());
        manuHisTableRuleConfig.setTableShardingStrategyConfig(manu_workorder);
        ShardingRuleConfiguration shardingRuleConfig = new ShardingRuleConfiguration();
        shardingRuleConfig.getTableRuleConfigs().add(manuHisTableRuleConfig);

        //manu_order表分片规则添加
        TableRuleConfiguration manuHisTableRuleConfig = new TableRuleConfiguration("manu_order", type+".manu_order_$->{2020..2023}");
        StandardShardingStrategyConfiguration manu_workorder = new StandardShardingStrategyConfiguration("created_date", new TableShardingAlgorithm());
        manuHisTableRuleConfig.setTableShardingStrategyConfig(manu_workorder);
        shardingRuleConfig = new ShardingRuleConfiguration();*/
        String [] tableNames="act_test_data,manu_workorder,manu_order,manu_workorder_list".split(",");
        String [] columNames="testtime,integrate_time,created_date,create_time".split(",");
        ShardingRuleConfiguration shardingRuleConfig = new ShardingRuleConfiguration();
        for(int i=0;i<4;i++){
            TableRuleConfiguration tableRuleConfig = this.getTableRuleConfig(tableNames[i], columNames[i], type);
            shardingRuleConfig.getTableRuleConfigs().add(tableRuleConfig);
        }
        //shardingRuleConfig.getTableRuleConfigs().add(manuHisTableRuleConfig);

        try {
            Properties properties = new Properties();
            properties.setProperty("sql.show", "true");
            DataSource dataSource1 = ShardingDataSourceFactory.createDataSource(dsMap, shardingRuleConfig, properties);
            return dataSource1;
        } catch (SQLException throwables) {
            throwables.printStackTrace();
            throw new IllegalArgumentException();
        }
    }

    /**
     * 根据表名、分列名、租户标识,获取表的分片规则
     * @param tableName
     * @param columName
     * @param type
     * @return
     */
    private TableRuleConfiguration getTableRuleConfig(String tableName,String columName,String type){
        TableRuleConfiguration tableRuleConfiguration = new TableRuleConfiguration(tableName, type+"."+tableName+"_$->{2020..2023}");
        StandardShardingStrategyConfiguration shardingRuleConfig = new StandardShardingStrategyConfiguration(columName, new TableShardingAlgorithm());
        tableRuleConfiguration.setTableShardingStrategyConfig(shardingRuleConfig);
        return tableRuleConfiguration;
    }

}

4.DynamicDataSourceContextHolder上下文切换

package com.ctl.mes.service.execute.config;


import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 *动态数据源上下文（切换数据源必须在调用service之前进行，也就是开启事务之前）
 * @author  weiyongfu
 * @date  2021/10/9 11:31
 */
public class DynamicDataSourceContextHolder {

    public static final String DEFAULT_DATASOURCE = "master";

    private static final ThreadLocal<String> CONTEXT_HOLDER_EXE = new ThreadLocal<String>() {
        /**
         * 将 master 数据源的 key作为默认数据源的 key
         */
        @Override
        protected String initialValue() {
            return DEFAULT_DATASOURCE;
        }
    };

    /**
     数据源的 key集合，用于切换时判断数据源是否存在
     */
    private static List<Object> dataSourceKeys = new ArrayList<>();

    /**
     切换数据源
     @param key String
     */
    public static void setDataSourceKey(String key) {
        CONTEXT_HOLDER_EXE.set(key);
    }

    /**
     获取数据源
     @return String
     */
    public static String getDataSourceKey() {
        String s = CONTEXT_HOLDER_EXE.get();
        return s;
    }

    /**
     重置数据源
     */
    public static void clearDataSourceKey() {
        CONTEXT_HOLDER_EXE.remove();
    }

    /**
     判断是否包含数据源
     @param key 数据源key
     @return boolean
     */
    public static boolean containDataSourceKey(String key) {
        return dataSourceKeys.contains(key);
    }

    /**
     添加数据源keys
     @param keys Collection
     @return boolean
     */
    public static boolean addDataSourceKeys(Collection<?> keys) {
        return dataSourceKeys.addAll(keys);
    }

}

5.DynamicDataSourceInit初始化数据源

package com.ctl.mes.service.execute.config;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONObject;

import com.ctl.mes.service.execute.advice.RemoteOauthService;
import com.ctl.mes.service.execute.domain.TenantInfo;
import com.ctl.mes.service.execute.service.RemoteBasicService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

/**
 * 初始化动态数据源
 * @author  weiyongfu
 * @date  2021/10/9 11:31
 */
@Slf4j
@Configuration
public class DynamicDataSourceInit {
    @Autowired
    private RemoteOauthService remoteOauthService;


    @Resource
    private DynamicDataSource dynamicDataSource;


    @PostConstruct
    public void initDataSource() throws SQLException {
        log.info("=====初始化动态数据源=====");

        //加载master数据源除外的其他数据源
        //List<TenantInfo> tenantList = remoteOauthService.findList(2);
        List<TenantInfo> tenantList = new ArrayList<>();
        TenantInfo tenantInfo1 = new TenantInfo();
        tenantInfo1.setTenantId("huawei");
        tenantInfo1.setDataSourceUrl("jdbc:mysql://xxx:3306/ctlmes_execute_data?useUnicode=true&characterEncoding=utf-8&serverTimezone=GMT%2B8&useSSL=false&allowMultiQueries=true");
        tenantInfo1.setDataSourceUsername("root");
        tenantInfo1.setDataSourcePassword("xx");
        tenantInfo1.setDataSourceDriver("com.mysql.cj.jdbc.Driver");
        tenantList.add(tenantInfo1);

        TenantInfo tenantInfo2 = new TenantInfo();
        tenantInfo2.setTenantId("xiaomi");
        tenantInfo2.setDataSourceUrl("jdbc:mysql://xxx:3306/ctlmes_execute_data?useUnicode=true&characterEncoding=utf-8&serverTimezone=GMT%2B8&useSSL=false&allowMultiQueries=true");
        tenantInfo2.setDataSourceUsername("root");
        tenantInfo2.setDataSourcePassword("xxx");
        tenantInfo2.setDataSourceDriver("com.mysql.cj.jdbc.Driver");
        tenantList.add(tenantInfo2);
        for (TenantInfo tenantInfo : tenantList) {
            log.info(tenantInfo.toString());
            DruidDataSource dataSource = new DruidDataSource();
            dataSource.setDriverClassName(tenantInfo.getDataSourceDriver());
            dataSource.setUrl(tenantInfo.getDataSourceUrl());
            dataSource.setUsername(tenantInfo.getDataSourceUsername());
            dataSource.setPassword(tenantInfo.getDataSourcePassword());
            dataSource.init();
            dynamicDataSource.setDataSources(tenantInfo);
        }
        log.info("====动态数据源信息===="+ JSONObject.toJSON(tenantList));
        log.info("====初始化动态数据源结束====");
    }

}

6.TableShardingAlgorithm分表规则

package com.ctl.mes.service.execute.config;

import groovy.util.logging.Slf4j;
import org.apache.shardingsphere.api.sharding.standard.PreciseShardingAlgorithm;
import org.apache.shardingsphere.api.sharding.standard.PreciseShardingValue;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Collection;
import java.util.Date;

/**
 * 数据库分表策略
 * @author weiyongfu
 * @version 1.0
 * @date 2021/10/13 9:37
 */
@Slf4j
public class TableShardingAlgorithm implements PreciseShardingAlgorithm<Date> {

    @Override
    public String doSharding(Collection<String> availableTargetNames, PreciseShardingValue<Date> shardingValue) {
        System.out.println("table PreciseShardingAlgorithm ");
        String tb_name = shardingValue.getLogicTableName() + "_";
        // 根据当前日期 来 分库分表
        Date date = shardingValue.getValue();
        String year = String.format("%tY", date);
        // 选择表
        tb_name = tb_name + year;
        System.out.println("tb_name:" + tb_name);
        for (String each : availableTargetNames) {
            if (each.equals(tb_name)) {
                return each;

            }
        }
        throw new IllegalArgumentException();
    }
}

7.aop文件DynamicDataSourceAspect

package com.ctl.mes.service.execute.aspect;

import com.ctl.mes.service.execute.config.DynamicDataSource;
import com.ctl.mes.service.execute.config.DynamicDataSourceContextHolder;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.*;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

/**
 Aop动态切换多数据源
 * 请注意：这里order一定要小于tx:annotation-driven的order，即先执行DynamicDataSourceAspect切面，再执行事务切面，才能获取到最终的数据源
 * @author  weiyongfu
 * @date  2021/10/9 13:31
 */
@Slf4j
@Aspect
@Component
@Order(-1)
public class DynamicDataSourceAspect {
   /* @Autowired
    RemoteCoreService remoteCoreService;*/

    /**
     * 切点: 所有controller方法进去切面，根据需要看service设置切点不
     */
    @Pointcut("execution(* com.ctl.mes.service.execute.controller.*.*(..)) || execution(* com.ctl.mes.service.execute.service.*.*(..))")
    public void dataSourcePointCut() {
    }

    @Around("dataSourcePointCut()")
    public Object doAround(ProceedingJoinPoint point) throws Throwable {
        String dataSourceKey = DynamicDataSourceContextHolder.getDataSourceKey();
        DynamicDataSourceContextHolder.setDataSourceKey("GTS110");
        log.info("动态数据源切换成功:【"+dataSourceKey+"】切换为【"+"GTS110】");
        return point.proceed();
    }




}
8.DataSource文件

package com.ctl.mes.service.execute.aspect;


import com.ctl.mes.service.execute.config.DynamicDataSourceContextHolder;

import java.lang.annotation.*;

/**
 * @author  weiyongfu
 * @date  2021/10/9 13:31
 */
@Documented
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface DataSource {

    String value() default DynamicDataSourceContextHolder.DEFAULT_DATASOURCE;

}

9.项目结构以及效果图