MyBatis-Flex+ShardingSphere-JDBC多数据源分库分表实现_Java

1、目的

用于动态配置shardingsphere-jdbc实现配置化分库分表

2、实现

2-1、导入mybatis-flex和shardingsphere-jdbc的相关依赖

        <dependency>
            <groupid>com.mybatis-flex</groupid>
            <artifactid>mybatis-flex-spring-boot-starter</artifactid>
            <version>1.9.3</version>
        </dependency>
        <dependency>
            <groupid>org.apache.shardingsphere</groupid>
            <artifactid>shardingsphere-jdbc-core-spring-boot-starter</artifactid>
            <version>5.1.1</version>
        </dependency>

2-2、配置初始化的数据库连接用来加载配置，当然用配置中心来保存初始化数据的配置

spring.datasource.ds1.jdbc-url=jdbc:mysql://localhost/test?allowpublickeyretrieval=true
spring.datasource.ds1.driver-class-name=com.mysql.cj.jdbc.driver
spring.datasource.ds1.username=root
spring.datasource.ds1.password=123456
spring.datasource.ds1.type=com.zaxxer.hikari.hikaridatasource

2-3、初始化数据源进行配置查询

初始化数据源配置类：

package com.mochenli.shardingshere.config;
import org.springframework.boot.context.properties.configurationproperties;
import org.springframework.boot.jdbc.datasourcebuilder;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
import javax.sql.datasource;

/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-27 21:23
 */
@configuration
public class datasourceconfig {
    /**
     * 根据配置文件构建数据源
     * @return
     */
    @bean
    @configurationproperties(prefix = "spring.datasource.ds1")
    public datasource datasourceone(){
      return datasourcebuilder.create().build();
    }

}

数据配置表以及数据

/*
 navicat premium data transfer

 source server         : localhost
 source server type    : mysql
 source server version : 80034 (8.0.34)
 source host           : localhost:3306
 source schema         : test

 target server type    : mysql
 target server version : 80034 (8.0.34)
 file encoding         : 65001

 date: 29/06/2024 17:52:36
*/

set names utf8mb4;
set foreign_key_checks = 0;

-- ----------------------------
-- table structure for databaseconfig
-- ----------------------------
drop table if exists `databaseconfig`;
create table `databaseconfig`  (
  `id` int not null auto_increment comment '主键唯一标识',
  `jdbc_url` varchar(255) character set utf8mb4 collate utf8mb4_0900_ai_ci null default null comment '数据库连接',
  `driver_class_name` varchar(255) character set utf8mb4 collate utf8mb4_0900_ai_ci null default null comment '数据库连接驱动',
  `username` varchar(255) character set utf8mb4 collate utf8mb4_0900_ai_ci null default null comment '数据库连接用户名',
  `password` varchar(255) character set utf8mb4 collate utf8mb4_0900_ai_ci null default null comment '数据库连接密码',
  `connection_name` varchar(255) character set utf8mb4 collate utf8mb4_0900_ai_ci null default null comment '数据库连接名称',
  primary key (`id`) using btree
) engine = innodb auto_increment = 3 character set = utf8mb4 collate = utf8mb4_0900_ai_ci row_format = dynamic;

-- ----------------------------
-- records of databaseconfig
-- ----------------------------
insert into `databaseconfig` values (1, 'jdbc:mysql://192.168.157.129:3310/db_order?allowpublickeyretrieval=true', 'com.mysql.cj.jdbc.driver', 'root', '123456', 'server-order0');
insert into `databaseconfig` values (2, 'jdbc:mysql://192.168.157.129:3311/db_order?allowpublickeyretrieval=true', 'com.mysql.cj.jdbc.driver', 'root', '123456', 'server-order1');

set foreign_key_checks = 1;

数据库配置对象类

package com.mochenli.shardingshere.entity;
import com.mybatisflex.annotation.id;
import com.mybatisflex.annotation.keytype;
import com.mybatisflex.annotation.table;
import lombok.allargsconstructor;
import lombok.data;
import lombok.noargsconstructor;
/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-29 17:28
 */
@data
@noargsconstructor
@allargsconstructor
@table("databaseconfig")
public class databaseconfig {

    @id(keytype = keytype.auto)
    private integer id;

    private string jdbcurl;

    private string username;

    private string password;

    private string driverclassname;

    private string connectionname;
}

mapper类

package com.mochenli.shardingshere.mapper;
import com.mochenli.shardingshere.entity.databaseconfig;
import com.mybatisflex.core.basemapper;
import org.apache.ibatis.annotations.mapper;
/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-29 17:31
 */
@mapper
public interface databaseconfigmapper extends basemapper<databaseconfig> {
}

2-4、初始化shardingdatasource数据源实现分库分表

分片算法可查看链接

package com.mochenli.shardingshere.config;
import com.mochenli.shardingshere.entity.databaseconfig;
import com.mochenli.shardingshere.mapper.databaseconfigmapper;
import com.mybatisflex.core.flexglobalconfig;
import com.mybatisflex.core.datasource.flexdatasource;
import com.zaxxer.hikari.hikaridatasource;
import org.apache.shardingsphere.driver.api.shardingspheredatasourcefactory;
import org.apache.shardingsphere.infra.config.algorithm.shardingspherealgorithmconfiguration;
import org.apache.shardingsphere.infra.config.mode.modeconfiguration;
import org.apache.shardingsphere.sharding.api.config.shardingruleconfiguration;
import org.apache.shardingsphere.sharding.api.config.rule.shardingtableruleconfiguration;
import org.apache.shardingsphere.sharding.api.config.strategy.keygen.keygeneratestrategyconfiguration;
import org.apache.shardingsphere.sharding.api.config.strategy.sharding.shardingstrategyconfiguration;
import org.apache.shardingsphere.sharding.api.config.strategy.sharding.standardshardingstrategyconfiguration;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
import javax.sql.datasource;
import java.sql.sqlexception;
import java.util.*;
/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-19 13:53
 */
@configuration
public class shardingconfig {

     private databaseconfigmapper databaseconfigmapper;

    public shardingconfig(databaseconfigmapper databaseconfigmapper) throws sqlexception {
        this.databaseconfigmapper = databaseconfigmapper;
        flexdatasource flexdatasource = flexglobalconfig.getdefaultconfig()
                .getdatasource();
        flexdatasource.adddatasource("getshardingdatasource",getshardingdatasource());
    }
    /**
     * 配置shardingsphere的内存模式
     * @return
     */
    @bean
    public modeconfiguration getmodeconfiguration() {
        modeconfiguration modeconfiguration = new modeconfiguration("memory", null, false);
        return modeconfiguration;
    }
    /**
     * 构建shardingdatasource数据源
     * @return
     * @throws sqlexception
     */
    public datasource getshardingdatasource() throws sqlexception {
        //查询数据库的数据连接配置设置到datasourcemap当中
        map<string, datasource> datasourcemap = new hashmap<>();
        list<databaseconfig> databaseconfigs = databaseconfigmapper.selectall();
        for (databaseconfig databaseconfig : databaseconfigs) {
            datasource datasource = createdatasource(databaseconfig);
            datasourcemap.put(databaseconfig.getconnectionname(), datasource);
        }
        //以下分片配置的规则也可持久化从数据库查询出来进行配置 此处用于演示所需即静态配置
        // 配置分片规则
        shardingruleconfiguration shardingruleconfig = new shardingruleconfiguration();
        //算法相关配置
        map<string, shardingspherealgorithmconfiguration> shardingspherealgorithmconfigurations = new hashmap<>();
        properties properties = new properties();
        properties.put("algorithm-expression","server-order$->{user_id % 2}");
        shardingspherealgorithmconfiguration shardingspherealgorithmconfiguration =  new shardingspherealgorithmconfiguration("inline",properties);
        properties properties1 = new properties();
        properties1.put("sharding-count","2");
        shardingspherealgorithmconfiguration shardingspherealgorithmconfiguration1 =  new shardingspherealgorithmconfiguration("mod",properties1);
        properties properties2 = new properties();
        properties2.put("sharding-count","2");
        shardingspherealgorithmconfiguration shardingspherealgorithmconfiguration2 =  new shardingspherealgorithmconfiguration("hash_mod",properties2);
        shardingspherealgorithmconfigurations.put("alg_inline_userid",shardingspherealgorithmconfiguration);
        shardingspherealgorithmconfigurations.put("alg_mod",shardingspherealgorithmconfiguration1);
        shardingspherealgorithmconfigurations.put("alg_hash_mod",shardingspherealgorithmconfiguration2);
        shardingruleconfig.setshardingalgorithms(shardingspherealgorithmconfigurations);
        //分库分表策略配置
        collection<shardingtableruleconfiguration> shardingtableruleconfigurations = new arraylist<>();
        //分库策略
        shardingtableruleconfiguration shardingtableruleconfiguration = new shardingtableruleconfiguration("t_order","server-order$->{0..1}.t_order$->{0..1}");
        shardingstrategyconfiguration shardingstrategyconfiguration = new standardshardingstrategyconfiguration("user_id","alg_mod");
        shardingtableruleconfiguration.setdatabaseshardingstrategy(shardingstrategyconfiguration);
        //分表策略
        shardingstrategyconfiguration shardingstrategyconfigurationtable = new standardshardingstrategyconfiguration("order_no","alg_hash_mod");
        shardingtableruleconfiguration.settableshardingstrategy(shardingstrategyconfigurationtable);
        shardingtableruleconfigurations.add(shardingtableruleconfiguration);
        shardingruleconfig.settables(shardingtableruleconfigurations);
        //配置主键算法生成策略
        map<string, shardingspherealgorithmconfiguration> keygenerators = new hashmap<>();
        shardingspherealgorithmconfiguration shardingspherealgorithmconfigurationsk =  new shardingspherealgorithmconfiguration("snowflake",null);
        keygenerators.put("alg_snowflake",shardingspherealgorithmconfigurationsk);
        shardingruleconfig.setkeygenerators(keygenerators);
        //分布式id
        keygeneratestrategyconfiguration keygeneratestrategyconfiguration =new keygeneratestrategyconfiguration("id","alg_snowflake");
        shardingtableruleconfiguration.setkeygeneratestrategy(keygeneratestrategyconfiguration);
        //属性设置运行sql打印显示
        properties properties3 = new properties();
        properties3.put("sql-show",true);
        // 创建shardingdatasource
        datasource datasources = shardingspheredatasourcefactory.createdatasource(datasourcemap, collections.singleton(shardingruleconfig),properties3 );
        return datasources;
    }
    /**
     * 创建数据源连接
     * @param databaseconfig
     * @return
     */
    public static datasource createdatasource(databaseconfig databaseconfig)  {
        // 创建数据源，这里需要根据实际情况创建，例如使用hikaricp、druid等连接池
        hikaridatasource datasource = new hikaridatasource();
        datasource.setdriverclassname(databaseconfig.getdriverclassname());
        datasource.setjdbcurl(databaseconfig.getjdbcurl());
        datasource.setusername(databaseconfig.getusername());
        datasource.setpassword(databaseconfig.getpassword());
        //不使用连接池
        //drivermanagerdatasource datasource1 = new drivermanagerdatasource();
        //datasource1.setdriverclassname(databaseconfig.getdriverclassname());
        //datasource1.seturl(databaseconfig.getjdbcurl());
        //datasource1.setusername(databaseconfig.getusername());
        //datasource1.setpassword(databaseconfig.getpassword());
        return datasource;
    }
}

2-5、两个数据库连接server-order0和server-order1的表结构如下：分别在两个库当中运行

set names utf8mb4;
set foreign_key_checks = 0;

-- ----------------------------
-- table structure for t_order0
-- ----------------------------
drop table if exists `t_order0`;
create table `t_order0`  (
  `id` bigint not null,
  `order_no` varchar(30) character set utf8mb4 collate utf8mb4_0900_ai_ci null default null,
  `user_id` bigint null default null,
  `amount` decimal(10, 2) null default null,
  primary key (`id`) using btree
) engine = innodb character set = utf8mb4 collate = utf8mb4_0900_ai_ci row_format = dynamic;


-- ----------------------------
-- table structure for t_order1
-- ----------------------------
drop table if exists `t_order1`;
create table `t_order1`  (
  `id` bigint not null,
  `order_no` varchar(30) character set utf8mb4 collate utf8mb4_0900_ai_ci null default null,
  `user_id` bigint null default null,
  `amount` decimal(10, 2) null default null,
  primary key (`id`) using btree
) engine = innodb character set = utf8mb4 collate = utf8mb4_0900_ai_ci row_format = dynamic;

set foreign_key_checks = 1;

2-6、实体这个@table(“t_order”)是逻辑表名shardingconfig分库策略时设置的

package com.mochenli.shardingshere.entity;
import com.mybatisflex.annotation.id;
import com.mybatisflex.annotation.keytype;
import com.mybatisflex.annotation.table;
import lombok.data;
import java.math.bigdecimal;
/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-18 17:15
 */
@table("t_order")
@data
public class order {
    @id(keytype = keytype.auto)
    private long id;
    private string orderno;
    private long userid;
    private bigdecimal amount;
}

2-7、mapper

package com.mochenli.shardingshere.mapper;
import com.mochenli.shardingshere.entity.order;
import com.mybatisflex.core.basemapper;
import org.apache.ibatis.annotations.mapper;
/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-18 17:18
 */
@mapper
public interface ordermapper extends basemapper<order> {
}

3、测试

3-1、测试分库分表的新增

package com.mochenli.shardingshere.controller;

import com.mochenli.shardingshere.entity.order;
import com.mochenli.shardingshere.mapper.ordermapper;
import com.mybatisflex.core.datasource.datasourcekey;
import lombok.allargsconstructor;
import org.springframework.web.bind.annotation.getmapping;
import org.springframework.web.bind.annotation.restcontroller;

/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-29 18:11
 */
@restcontroller
@allargsconstructor
public class testcontroller {
   private final   ordermapper ordermapper;
    /**
     * id是用分布式id雪花算法 所以不填
     * 测试新增   因为前面配置是根据user_id进行分库 分库算法是:取模分片，算法类型：mod  分片数量是 2
     * 分表算法是根据 order_no 进行分表  分表算法是 哈希取模分片算法，类型：hash_mod  分片数量是 2
     */
    @getmapping("/testaddorder")
    public string addtorder(){
        //切换数据源 使用shardingdatasource数据源
        datasourcekey.use("getshardingdatasource");
        //进行分库分表插入
        for (int i = 1; i <=60; i++) {
            order order = new order();
            order.setuserid(long.valueof(i));
            order.setorderno("分表算法"+i);
            ordermapper.insert(order);
        }
        return "success";
    }
}

结果如下所示：

server_order0.t_order0

server_order0.t_order1

server_order1.t_order0

server_order1.t_order1

3-2、测试分页查询：

package com.mochenli.shardingshere.controller;

import com.mochenli.shardingshere.entity.order;
import com.mochenli.shardingshere.mapper.ordermapper;
import com.mybatisflex.core.datasource.datasourcekey;
import com.mybatisflex.core.paginate.page;
import com.mybatisflex.core.query.querywrapper;
import lombok.allargsconstructor;
import org.springframework.web.bind.annotation.getmapping;
import org.springframework.web.bind.annotation.restcontroller;

/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-29 18:11
 */
@restcontroller
@allargsconstructor
public class testcontroller {
   private final   ordermapper ordermapper;
    
      /**
     * 测试分页查询
     * @return
     */
    @getmapping("/testpageorder")
    public page<order> getpage(){
        //切换数据源 使用shardingdatasource数据源
        datasourcekey.use("getshardingdatasource");
        //进行分页查询
        page<order> page = new page<>();
        page.setpagenumber(1);
        page.setpagesize(27);
        page<order> paginate = ordermapper.paginate(page, new querywrapper());
        return paginate;
    

}

结果如下:

3-3、测试事务问题

1、正常情况

package com.mochenli.shardingshere.controller;

import com.mochenli.shardingshere.entity.databaseconfig;
import com.mochenli.shardingshere.entity.order;
import com.mochenli.shardingshere.mapper.databaseconfigmapper;
import com.mochenli.shardingshere.mapper.ordermapper;
import com.mybatisflex.core.datasource.datasourcekey;
import com.mybatisflex.core.paginate.page;
import com.mybatisflex.core.query.querywrapper;
import lombok.allargsconstructor;
import org.springframework.web.bind.annotation.getmapping;
import org.springframework.web.bind.annotation.restcontroller;
import java.util.hashmap;
import java.util.list;
import java.util.map;

/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-29 18:11
 */
@restcontroller
@allargsconstructor
public class testcontroller {
   private final   ordermapper ordermapper;
   private  final databaseconfigmapper databaseconfigmapper;
  
    /**
     * 测试事务问题
     * 正常情况
     */
    @getmapping("/testtransaction")
    public map<string,object> testtransaction(){
        map<string,object> map = new hashmap<>();
        list<databaseconfig> databaseconfigs = databaseconfigmapper.selectall();
        map.put("databaseconfigs",databaseconfigs);
        datasourcekey.clear();//清除数据源
        //切换数据源 使用shardingdatasource数据源
        datasourcekey.use("getshardingdatasource");
        list<order> orders = ordermapper.selectall();
        map.put("orders",orders);
        return map;
    }

}

结果：

2、出错进行事务回滚情况一

package com.mochenli.shardingshere.controller;

import com.mochenli.shardingshere.entity.databaseconfig;
import com.mochenli.shardingshere.entity.order;
import com.mochenli.shardingshere.mapper.databaseconfigmapper;
import com.mochenli.shardingshere.mapper.ordermapper;
import com.mybatisflex.core.datasource.datasourcekey;
import com.mybatisflex.core.paginate.page;
import com.mybatisflex.core.query.querywrapper;
import lombok.allargsconstructor;
import org.springframework.transaction.annotation.transactional;
import org.springframework.web.bind.annotation.getmapping;
import org.springframework.web.bind.annotation.restcontroller;

import java.util.hashmap;
import java.util.list;
import java.util.map;

/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-29 18:11
 */
@restcontroller
@allargsconstructor
public class testcontroller {
    private final ordermapper ordermapper;
    private final databaseconfigmapper databaseconfigmapper;
    /**
     * 测试事务问题
     * 报错情况一  不分库的数据源报错  回滚
     */
    @getmapping("/testtransactionerror")
    @transactional
    public void testtransactionerror() {
        databaseconfig databaseconfig = new databaseconfig();
        databaseconfig.setconnectionname("连接名称");
        databaseconfig.setjdbcurl("连接字符串");
        databaseconfig.setusername("用户名");
        databaseconfig.setpassword("密码");
        databaseconfig.setdriverclassname("驱动");
        //进行新增
        databaseconfigmapper.insert(databaseconfig);
        //模拟报错
        int i = 10 / 0;
        datasourcekey.clear();//清除数据源
        //切换数据源 使用shardingdatasource数据源
        datasourcekey.use("getshardingdatasource");
        list<order> orders = ordermapper.selectall();
    }

}

3、出错进行事务回滚情况二

package com.mochenli.shardingshere.controller;

import com.mochenli.shardingshere.entity.databaseconfig;
import com.mochenli.shardingshere.entity.order;
import com.mochenli.shardingshere.mapper.databaseconfigmapper;
import com.mochenli.shardingshere.mapper.ordermapper;
import com.mybatisflex.core.datasource.datasourcekey;
import com.mybatisflex.core.paginate.page;
import com.mybatisflex.core.query.querywrapper;
import lombok.allargsconstructor;
import org.springframework.transaction.annotation.transactional;
import org.springframework.web.bind.annotation.getmapping;
import org.springframework.web.bind.annotation.restcontroller;

import java.util.hashmap;
import java.util.list;
import java.util.map;

/**
 * @author: mochenli
 * @description:
 * @createtime: 2024-06-29 18:11
 */
@restcontroller
@allargsconstructor
public class testcontroller {
    private final ordermapper ordermapper;
    private final databaseconfigmapper databaseconfigmapper;

    /**
     * 测试事务问题
     * 报错情况二  分库分表的数据源报错  回滚
     */
    @getmapping("/testtransactionerror1")
    @transactional
    public void testtransactionerror1() {
        //切换数据源 使用shardingdatasource数据源
        datasourcekey.use("getshardingdatasource");
        for (int i = 1; i <= 5; i++) {
            order order = new order();
            //不出错 1%2=1应该在server_order1库当中
            order.setuserid(1l);
           //不出错  "事务回滚测试".hashcode() % 2 = 0 应该在 t_order0表当中;
            order.setorderno("事务回滚测试");
            ordermapper.insert(order);
        }
        //模拟报错
        int k = 10 / 0;
        datasourcekey.clear();//清除数据源
        list<databaseconfig> databaseconfigs = databaseconfigmapper.selectall();
    }

}

结果回滚了