Mybatis-Plus集成Sharding-JDBC与Flyway实现多租户分库分表

背景

公司产品部收到了一些重要客户的需求，他们希望能够依赖独立的数据库存储来支持他们的业务数据。与此同时，仍有许多中小客户，可以继续使用公共库以满足其需求。技术实现方面，此前持久层框架使用的Mybatis-plus，部分业务场景使用到了Sharding-JDBC用于分表，另外，我们的数据库版本控制工具使用的是Flyway。
方案说明

这里将方案进行简要说明，配置统一通过Nacos管理(有需要的可以自行定义租户配置页面)。

• 1.首先多数据源管理使用Mybatis-Plus官方推荐的dynamic-datasource-spring-boot-starter组件，需要注意的是构建动态多数据源时，我们要把Sharding-JDBC数据源也纳入管理。因为我们的库里面毕竟只有部分表用到了Sharding-JDBC，这样可以复用数据源。
  
  
• 2.其次，租户与数据源之间在Nacos建立关系配置，确保根据租户ID能够路由到唯一的租户数据源。我们需要自定义Sharding分片策略和多数据源切换逻辑，根据http请求传入的租户ID，设置正确的数据源。
  
  
• 3.动态数据源与Sharding数据源配置做为公共配置在Nacos维护，在业务服务启动时，读取公共配置初始化多数据源，并添加对公共多数据源配置的监听。当配置变更时，重新构造Sharding数据源，并并更新动态多数据源。另外数据库脚本通过自定义flyway配置执行。
  
  

技术实现

前提

需要在Nacos提前维护租户与数据源关系配置。
不使用Sharding-JDBC场景

• 1.引入相关组件依赖。
  

1.         <dependency>
2.             <groupId>com.alibaba.nacos</groupId>
3.             <artifactId>nacos-client</artifactId>
4.             <version>2.1.0</version>
5.         </dependency>
6.         <dependency>
7.             <groupId>org.flywaydb</groupId>
8.             <artifactId>flyway-core</artifactId>
9.             <version>7.15.0</version>
10.         </dependency>
11.         <dependency>
12.             <groupId>com.baomidou</groupId>
13.             <artifactId>mybatis-plus-boot-starter</artifactId>
14.             <version>3.4.1</version>
15.         </dependency>
16.         <dependency>
17.             <groupId>com.baomidou</groupId>
18.             <artifactId>dynamic-datasource-spring-boot-starter</artifactId>
19.             <version>3.4.1</version>
20.         </dependency>
21.         <dependency>
22.             <groupId>com.alibaba</groupId>
23.             <artifactId>druid</artifactId>
24.             <version>1.2.6</version>
25.         </dependency>

复制代码

• 2.关闭Flyway自动配置和配置多数据源。
  

1. spring:
2.   flyway:
3.     #关闭flyway自动配置，自定义实现
4.     enabled: false
5.   datasource:
6.     dynamic:
7.       #默认数据源
8.       primary: ds0
9.       datasource:
10.         ds0:
11.           type: com.alibaba.druid.pool.DruidDataSource
12.           driverClassName: org.postgresql.Driver
13.           url: jdbc:postgresql://127.0.0.1:5432/ds0
14.           username: ds0
15.           password: ds0123
16.         ds1:
17.           type: com.alibaba.druid.pool.DruidDataSource
18.           driverClassName: org.postgresql.Driver
19.           url: jdbc:postgresql://127.0.0.1:5432/ds1
20.           username: ds1
21.           password: ds1123

复制代码

• 3.自定义实现Flyway配置类，对应的flyway脚本目录结构见下图，主库和租户库SQL脚本独立维护。
  

1. Java
2. @Slf4j
3. @Configuration
4. @EnableTransactionManagement
5. public class FlywayConfig {
6.     @Value("${spring.application.name}")
7.     private String appName;
8.     @Autowired
9.     private DataSource dataSource;
11.     @Bean
12.     public void migrate() {
13.         log.info("flyway开始逐数据源执行脚本");
14.         DynamicRoutingDataSource ds = (DynamicRoutingDataSource) dataSource;
15.         Map<String, DataSource> dataSources = ds.getDataSources();
16.         dataSources.forEach((k, v) -> {
17.             if (!"sharding".equals(k)) {
18.                                                     // Flyway相关参数建议通过配置管理，以下代码仅供参考
19.                 Flyway flyway = Flyway.configure()
20.                         .dataSource(v)
21.                         .table("t_" + k + "_" + appName + "_version")
22.                         .baselineOnMigrate(true)
23.                         .outOfOrder(true)
24.                         .baselineVersion("1.0.0")
25.                         .baselineDescription(k + "初始化")
26.                         .locations(CommonConstant.SQL_BASE_LOCATION + (CommonConstant.DEFAULT_DS_NAME.equals(k) ? CommonConstant.MASTER_DB : CommonConstant.TENANT_DB))
27.                         .load();
28.                 flyway.migrate();
29.                 log.info("flyway在 {} 数据源执行脚本成功", k);
30.             }
31.         });
32.     }
33. }

复制代码

• 4.自定义实现数据源切换Filter类。
  

1. @Slf4j
2. @Component
3. @WebFilter(filterName = "dynamicDatasourceFilter", urlPatterns = {"/*"})
4. public class DynamicDatasourceFilter implements Filter {
5.     // 构建演示用租户与数据源关系配置
6.                 private static Map<String, String> tenantDsMap = new HashMap<>();
7.     static {
8.         tenantDsMap.put("tenant123", "ds0");
9.         tenantDsMap.put("tenant456", "ds0");
10.                                 tenantDsMap.put("tenant789", "ds1");
11.     }
13.     @Override
14.     public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
15.         HttpServletRequest httpRequest = (HttpServletRequest) request;
16.         // 从请求头获取租户ID
17.         String tenantId = httpRequest.getHeader(CommonConstant.TENANT_HEADER);
18.         try {
19.             // 设置数据源
20.             if (tenantDsMap.get(tenantId) == null) {
21.                 // 如果根据租户ID未找到租户数据源配置，默认走主库
22.                 DynamicDataSourceContextHolder.push(CommonConstant.DEFAULT_DS_NAME);
23.             } else {
24.                 //注意，如果是分片表，那么需要在分片表Service类或方法上加@DS("sharding")注解，最终由sharding的库分片策略决定SQL在哪个库执行。而这里的设置将会被@DS注解配置覆盖
25.                 DynamicDataSourceContextHolder.push(tenantDsMap.get(tenantId));
26.             }
27.             // 执行
28.             chain.doFilter(request, response);
29.         } catch (Exception e) {
30.             log.error("切换数据源失败，tenantId={}，请求接口uri={}，异常原因：{}", tenantId, httpRequest.getRequestURI(), ExceptionUtils.getStackTrace(e));
31.         } finally {
32.             // 清空当前线程数据源
33.             DynamicDataSourceContextHolder.poll();
34.         }
35.     }

复制代码

使用Sharding-JDBC

如果微服务还需要使用Sharding分片，那么还需要引入sharding-jdbc组件依赖，并配置sharding数据源和分片规则。如果是多个服务共用数据库，那么建议将Sharding数据源配置做为公共配置在Nacos管理，而Sharding分片规则则做为服务个性化配置单独维护(分片规则基本不需要动态变更)，这样当有新租户需要申请开通独立租户库的时候，直接变更Sharding数据源公共配置，服务在监听到公共配置变更后，即可重新构建新的Sharding数据源实例和动态数据源更新，无需重启服务。

• 1.引入sharding-jdbc组件依赖。
  

1.         <dependency>
2.             <groupId>org.apache.shardingsphere</groupId>
3.             <artifactId>sharding-jdbc-core</artifactId>
4.             <version>4.1.1</version>
5.         </dependency>

复制代码

• 2.配置Sharding数据源和分片规则。
  

1. # sharding数据源配置
2. dataSources:
3.   ds0: !!com.alibaba.druid.pool.DruidDataSource
4.     driverClassName: org.postgresql.Driver
5.     url: jdbc:postgresql://127.0.0.1:5432/ds0
6.     username: ds0
7.     password: ds0123
8.   ds1: !!com.alibaba.druid.pool.DruidDataSource
9.     driverClassName: org.postgresql.Driver
10.     url: jdbc:postgresql://127.0.0.1:5432/ds1
11.     username: ds1
12.     password: ds1123
13.   ds2: !!com.alibaba.druid.pool.DruidDataSource
14.     driverClassName: org.postgresql.Driver
15.     url: jdbc:postgresql://127.0.0.1:5432/ds2
16.     username: ds2
17.     password: ds2123
18. # sharding分片规则配置
19. shardingRule:
20.   tables:
21.     t_order:
22.       actualDataNodes: ds$->{0..2}.t_order$->{0..1}
23.       tableStrategy:
24.         inline:
25.           shardingColumn: order_no
26.           algorithmExpression: t_order$->{order_no.toBigInteger() % 2}
27.   defaultDataSourceName: ds0
28.   # 默认库分片策略
29.   defaultDatabaseStrategy:
30.     standard:
31.       shardingColumn: tenant_id
32.                         # 自定义精确分片策略
33.       preciseAlgorithmClassName: cn.xtstu.demo.config.CustomDataSourcePreciseShardingAlgorithm
34.     #hint:
35.                     #
36.     #  algorithmClassName: cn.xtstu.demo.config.CustomHintShardingAlgorithm
37.   defaultTableStrategy:
38.     none:
39. props:
40.   sql.show: true

复制代码

• 3.自定义精确分片策略。
  

1. public class CustomDataSourcePreciseShardingAlgorithm implements PreciseShardingAlgorithm<String> {
3.     // 构建演示用租户与数据源关系配置
4.                 private static Map<String, String> tenantDsMap = new HashMap<>();
5.     static {
6.         tenantDsMap.put("tenant123", "ds0");
7.         tenantDsMap.put("tenant456", "ds0");
8.                                 tenantDsMap.put("tenant789", "ds1");
9.     }
10.                
11.     @Override
12.     public String doSharding(Collection<String> dataSourceNames, PreciseShardingValue<String> shardingValue) {
13.                     // 库分片策略配置的分片键是字段tenant_id，根据分片键查询配置的数据源
14.                     String dsName = tenantDsMap.get(shardingValue.getValue());
15.         // 如果如前文所属，Sharding子数据源key与dynamic数据源key保持一致的话，这里直接返回就行了
16.                                 return dsName;
17.         // TODO 需要处理未匹配到数据源的情况
18.     }
19. }

复制代码

• 4.自定义Hint分片策略(可选)，适用于分片键与SQL无关的场景。
  

1. public class CustomHintShardingAlgorithm implements HintShardingAlgorithm<Integer> {
3.     // 构建演示用租户与数据源关系配置
4.                 private static Map<String, String> tenantDsMap = new HashMap<>();
5.     static {
6.         tenantDsMap.put("tenant123", "ds0");
7.         tenantDsMap.put("tenant456", "ds0");
8.                                 tenantDsMap.put("tenant789", "ds1");
9.     }
10.                
11.     @Override
12.     public Collection<String> doSharding(Collection<String> collection, HintShardingValue<Integer> hintShardingValue) {
13.         Collection<String> result = new ArrayList<>();
14.         // 从请求头取到当前租户ID
15.                                 HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
16.         result.add(tenantDsMap.get(request.getHeader("tenantId")));
17.                                 // TODO  需要处理未匹配到数据源的情况
18.         return result;
19.     }
20. }

复制代码

• 5.自定义动态数据源配置(核心就是将sharding数据源及其子数据源添加到动态数据源一起管理)。
  

1. @Slf4j
2. @Configuration
3. public class CustomDynamicDataSourceConfig {
4.     @Value("${spring.cloud.nacos.config.extension-configs[0].data-id}")
5.     private String dataId;
6.     @Value("${spring.cloud.nacos.config.group:DEFAULT_GROUP}")
7.     private String group;
8.     @Resource
9.     private DynamicDataSourceProperties properties;
10.     @Resource
11.     private NacosHelper nacosHelper;
13.     /**
14.      * 启动时通过查询Nacos上sharding数据源及分片规则yaml配置初始化sharding-jdbc数据源
15.      *
16.      * @return
17.      */
18.     @Bean
19.     public ShardingDataSource shardingDataSource() {
20.         ConfigService configService = nacosHelper.getConfigService();
21.         if (configService == null) {
22.             log.error("连接nacos失败");
23.         }
24.         String configInfo = null;
25.         try {
26.             configInfo = configService.getConfig(dataId, group, 5000);
27.         } catch (NacosException e) {
28.             log.error("获取{}配置失败，异常原因：{}", dataId, ExceptionUtils.getStackTrace(e));
29.         }
30.         if (StringUtils.isBlank(configInfo)) {
31.             log.error("{}配置为空，启动失败", dataId);
32.             throw new NullPointerException(dataId + "配置为空");
33.         }
34.         try {
35.             // 通过工厂类和yaml配置创建Sharding数据源
36.             return (ShardingDataSource) YamlShardingDataSourceFactory.createDataSource(configInfo.getBytes(StandardCharsets.UTF_8));
37.         } catch (Exception e) {
38.             log.error("创建sharding-jdbc数据源异常：{}", ExceptionUtils.getStackTrace(e));
39.             throw new NullPointerException("sharding-jdbc数据源为空");
40.         }
41.     }
43.     /**
44.      * 将动态数据源设置为首选的
45.      * 当spring存在多个数据源时, 自动注入的是首选的对象
46.      * 设置为主要的数据源之后，就可以支持shardingJdbc原生的配置方式了
47.      */
48.     @Primary
49.     @Bean
50.     public DataSource dataSource() {
51.         DynamicRoutingDataSource dataSource = new DynamicRoutingDataSource();
52.         dataSource.setPrimary(properties.getPrimary());
53.         dataSource.setStrict(properties.getStrict());
54.         dataSource.setStrategy(properties.getStrategy());
55.         dataSource.setP6spy(properties.getP6spy());
56.         dataSource.setSeata(properties.getSeata());
57.         return dataSource;
58.     }
60.     /**
61.      * 初始化动态数据源
62.      *
63.      * @return
64.      */
65.     @Bean
66.     public DynamicDataSourceProvider dynamicDataSourceProvider(ShardingDataSource shardingDataSource) {
67.         return new AbstractDataSourceProvider() {
68.             @Override
69.             public Map<String, DataSource> loadDataSources() {
70.                 Map<String, DataSource> dataSourceMap = new HashMap<>();
71.                 // 将sharding数据源整体添加到动态数据源里
72.                 dataSourceMap.put(CommonConstant.SHARDING_DS_NAME, shardingDataSource);
73.                 // 同时把sharding内部管理的子数据源也添加到动态数据源里
74.                 Map<String, DataSource> shardingInnerDataSources = shardingDataSource.getDataSourceMap();
75.                 dataSourceMap.putAll(shardingInnerDataSources);
76.                 return dataSourceMap;
77.             }
78.         };
79.     }
80. }

复制代码

• 6.最后给出一份通过监听Nacos配置变更动态更新数据源的示例代码。注意：这份示例代码中只给出了Sharding配置变更时的处理逻辑，如果是dynamic数据源配置的话，有需要的可以参考着自行实现。
  

1. @Slf4j
2. @Configuration
3. public class NacosShardingConfigListener {
4.     @Value("${spring.cloud.nacos.config.extension-configs[0].data-id}")
5.     private String dataId;
6.     @Value("${spring.cloud.nacos.config.group:DEFAULT_GROUP}")
7.     private String group;
8.     @Value("${spring.application.name}")
9.     private String appName;
10.     @Autowired
11.     private DataSource dataSource;
12.     @Autowired
13.     private NacosHelper nacosHelper;
15.     @PostConstruct
16.     public void shardingConfigListener() throws Exception {
17.         ConfigService configService = nacosHelper.getConfigService();
18.         if (configService == null) {
19.             return;
20.         }
21.         configService.addListener(dataId, group, new Listener() {
22.             @Override
23.             public Executor getExecutor() {
24.                 return null;
25.             }
27.             @Override
28.             public void receiveConfigInfo(String configInfo) {
29.                 log.info("configInfo:\n{}", configInfo);
30.                 if (StringUtils.isBlank(configInfo)) {
31.                     log.warn("sharding-jdbc配置为空，不会刷新数据源");
32.                     return;
33.                 }
34.                 try {
35.                     if (StringUtils.isNotBlank(configInfo)) {
36.                         // 通过yaml配置创建sharding数据源(注意：如果分片规则是独立配置文件，那么需要提前合并数据源和分片规则配置)
37.                         ShardingDataSource shardingDataSource = (ShardingDataSource) YamlShardingDataSourceFactory.createDataSource(configInfo.getBytes(StandardCharsets.UTF_8));
38.                         Map<String, DataSource> shardingInnerDataSources = shardingDataSource.getDataSourceMap();
39.                         DynamicRoutingDataSource ds = (DynamicRoutingDataSource) dataSource;
40.                         // 遍历sharding子数据源
41.                         for (String poolName : shardingInnerDataSources.keySet()) {
42.                             // TODO 这里还有个细节，如果yaml配置删减了数据源，对应数据源应该要从ds中remove掉，且主数据源不能被remove。另外其实只有新增的数据源才需要执行flyway脚本
43.                             // 将sharding子数据源逐个添加到动态数据源
44.                             ds.addDataSource(poolName, shardingInnerDataSources.get(poolName));
45.                             // 通过代码完成数据源Flyway配置，并执行迁移操作
46.                                                                                                                 Flyway flyway = Flyway.configure()
47.                                     .dataSource(dataSource)
48.                                     .table("t_" + poolName + "_" + appName + "_version")
49.                                     .baselineOnMigrate(true)
50.                                     .outOfOrder(true)
51.                                     .baselineVersion("1.0.0")
52.                                     .baselineDescription(poolName + "初始化")
53.                                     .locations(CommonConstant.SQL_BASE_LOCATION + CommonConstant.TENANT_DB)
54.                                     .load();
55.                             flyway.migrate();
56.                         }
57.                         // 将sharding数据源自身也添加到动态数据源
58.                         ds.addDataSource(CommonConstant.SHARDING_DS_NAME, shardingDataSource);
59.                         log.info("动态数据源刷新完成，现有数据源：{}", JSONUtil.toJsonStr(ds.getDataSources().keySet()));
60.                     }
61.                 } catch (Exception e) {
62.                     log.error("创建sharding-jdbc数据源异常：{}", ExceptionUtils.getStackTrace(e));
63.                 }
64.             }
65.         });
66.     }
67. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！