Spring使用sharding-jdbc实现读写分离或分库分表

逻辑表与实际表之间的对应关系,均匀分布

使用sharding-jdbc来实现无论是读写分离,还是分库分表,都是很简单易用的。

如下图,其中order被拆分为两个表:

 

001

 

//使用默认的分表配置

TableRuleConfiguration orderTableRuleConfig = new TableRuleConfiguration();

orderTableRuleConfig.setLogicTable(“t_order”);

orderTableRuleConfig.setActualDataNodes(“db0.t_order_0, db0.t_order_1, db1.t_order_0, db1.t_order_1″);

LogicTable and ActualTable

数据库分库分表的目的是将数据从原始表传播到不同数据库中的不同表,并在不更改原始sql的情况下查询数据。

这种映射关系将通过使用LogicTable和ActualTable来说明。假设使用PreparedStatement访问数据库,SQL如下:

select * from t_order where user_id = ? and order_id = ?;

当条件user_id = 0 并且 order_id= 0时,Sharding-JDBC会改变这个SQL为以下目标SQL:

select * from db0.t_order_0 where user_id = ? and order_id = ?;

第一个SQL中的t_order是LogicTable和db0。第二个SQL中的t_order_0是ActualTable。

规则配置

我们可以通过配置规则来实现上述功能,本部分将介绍详细的规则配置:

ShardingRuleConfiguration shardingRuleConfig = new ShardingRuleConfiguration(); shardingRuleConfig.getTableRuleConfigs().add(orderTableRule);

shardingRuleConfig.getTableRuleConfigs().add(orderItemTableRule);

shardingRuleConfig.setDefaultDatabaseShardingStrategyConfig(new ComplexShardingStrategyConfiguration(“user_id”, “xxx.ModuloDatabaseShardingAlgorithm”));

shardingRuleConfig.setDefaultTableShardingStrategyConfig(new ComplexShardingStrategyConfiguration(“order_id”, “xxx.ModuloTableShardingAlgorithm”));

数据源配置

我们需要创建至少一个数据源映射对象,用于描述数据源名称和数据源的映射。如果使用了分库,那么是需要两个创建两个BasicDataSource对象:

private BasicDataSource dataSource1() {
    BasicDataSource dataSource = new BasicDataSource();
    dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
    dataSource.setUrl("jdbc:mysql://127.0.0.1:3306/db0?serverTimezone=GMT&characterEncoding=utf8");
    dataSource.setUsername("root");
    dataSource.setPassword("root");

    dataSource.setInitialSize(0);
    dataSource.setMaxIdle(5);
    dataSource.setMinIdle(100);
    dataSource.setMaxOpenPreparedStatements(100);
    dataSource.setTestWhileIdle(true);
    dataSource.setValidationQuery("SELECT 1");
    dataSource.setTimeBetweenEvictionRunsMillis(3600000);
    dataSource.setMinEvictableIdleTimeMillis(18000000);
    dataSource.setTestOnBorrow(true);
    dataSource.setMaxWaitMillis(300000);

    return dataSource;
}

private BasicDataSource dataSource2() {
    BasicDataSource dataSource = new BasicDataSource();
    dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
    dataSource.setUrl("jdbc:mysql://127.0.0.1:3306/db1?serverTimezone=GMT&characterEncoding=utf8");
    dataSource.setUsername("root");
    dataSource.setPassword("root");

    dataSource.setInitialSize(0);
    dataSource.setMaxIdle(5);
    dataSource.setMinIdle(100);
    dataSource.setMaxOpenPreparedStatements(100);
    dataSource.setTestWhileIdle(true);
    dataSource.setValidationQuery("SELECT 1");
    dataSource.setTimeBetweenEvictionRunsMillis(3600000);
    dataSource.setMinEvictableIdleTimeMillis(18000000);
    dataSource.setTestOnBorrow(true);
    dataSource.setMaxWaitMillis(300000);

    return dataSource;
}

 

以下是数据源集合的代码:

Map<String, DataSource> dataSourceMap = new HashMap<>();

dataSourceMap.put(“ds_0″, datasource1());

dataSourceMap.put(“ds_1″, datasource());

*注:如果只是为了分表,那么无需创建两个数据源,但是如果你想实现读写分离或者是分库,那么则需要至少个数据源。

策略配置

一共有两个策越,分表是针对数据库跟数据库表

在sharding-jdbc中有两个用于分库分表的策略:

  • DatabaseShardingStrategy
  • TableShardingStrategy

DatabaseShardingStrategy用于分布式数据库的数据源的策略。

TableShardingStrategy用于分布数据库表的策略。

此外,这两种策略的API是相同的,因此我们只要对其中一种API进行详细的介绍就可以了。

特定表规则的全局默认策略

策略与数据表(t_order)规则密切相关,因为策略适用于特定的表规则。

TableRuleConfiguration orderTableRuleConfig = new TableRuleConfiguration();

orderTableRuleConfig.setLogicTable(“t_order”);

orderTableRuleConfig.setActualDataNodes(“ds_0.t_order_0, ds_0.t_order_1, ds_1.t_order_0, ds_1.t_order_1″);

orderTableRuleConfig.setDatabaseShardingStrategyConfig(new ComplexShardingStrategyConfiguration(“user_id”, “xxx.ModuloDatabaseShardingAlgorithm”));

orderTableRuleConfig.setTableShardingStrategyConfig(new ComplexShardingStrategyConfiguration(“order_id”, “xxx.ModuloTableShardingAlgorithm”));

上述的代码,有两种策略,第一种就是通过user_id进行数据库的分配;第二种就是根据order_id再对数据表进行分配。最终实现的逻辑代码其实是:xxx.ModuloDatabaseShardingAlgorithm、xxx.ModuloTableShardingAlgorithm。

如果所有或大部分数据表都使用相同的分片策略,则可以使用默认策略来简化配置。

 

TableRuleConfiguration orderTableRuleConfig = new TableRuleConfiguration();//第个表的策略

orderTableRuleConfig.setLogicTable(“t_order”);

orderTableRuleConfig.setActualDataNodes(“ds_0.t_order_0, ds_0.t_order_1, ds_1.t_order_0, ds_1.t_order_1″);

 

 

TableRuleConfiguration orderItemTableRuleConfig = new TableRuleConfiguration();//第二个表的策略

orderItemTableRuleConfig.setLogicTable(“t_order_item”);

orderItemTableRuleConfig.setActualDataNodes(“ds_0.t_order_item_0,ds_0.t_order_item_1,ds_1.t_order_item_0,ds_1.t_order_item_1″);

 

 

ShardingRuleConfiguration shardingRuleConfig = new ShardingRuleConfiguration();

shardingRuleConfig.getTableRuleConfigs().add(orderTableRuleConfig);

shardingRuleConfig.getTableRuleConfigs().add(orderItemTableRuleConfig);

shardingRuleConfig.setDefaultDatabaseShardingStrategyConfig(new ComplexShardingStrategyConfiguration(“user_id”, “xxx.ModuloDatabaseShardingAlgorithm”));

shardingRuleConfig.setDefaultTableShardingStrategyConfig(new ComplexShardingStrategyConfiguration(“order_id”, “xxx.ModuloTableShardingAlgorithm”));

上述代码其实跟前一段是一致的,但是因为配置了两个数据表的分配策略,所以创建了两个TableRuleConfiguration,然后通过ShardingRuleConfiguration的方法getTableRuleConfigs().add()把规则添加进去。

分表或分库的列字段

分片策略中设置为第一个参数为分库分表的列字段(user_id、order_id)是SQL中WHERE中的条件列。如果你的SQL语句的WHERE中可能会没有这两个列字段,那么你最好在xxx.ModuloDatabaseShardingAlgorithm、xxx.ModuloTableShardingAlgorithm逻辑代码中特殊处理一下,当然你也可以配置多个分片列。

分表分库的算法(以下来自官方的翻译文件)

Sharding-JDBC provides 5 kinds of sharding strategies. Because of the closely connection between specific business and specific sharding algorithms, Sharding-JDBC not carry out sharding algorithm. Instead, after making a higher level of abstraction, we provide API to allow developers to implement sharding algorithms as they need.

Sharding-JDBC提供了5种切分策略。由于特定业务和特定的切分算法之间的紧密联系,Sharding-JDBC没有执行切分算法。相反,在进行了更高级别的抽象之后,我们提供了允许开发人员根据需要实现切分算法的API。

  • StandardShardingStrategy (标准共享策略)

Support =, IN, BETWEEN AND in SQLs for sharding operation. StandardShardingStrategy only supports single sharding column, and provides two sharding algorithms of PreciseShardingAlgorithm and RangeShardingAlgorithm. The PreciseShardingAlgorithm is required to handle the sharding operation of = and IN. The RangeShardingAlgorithm is optional to handle BETWEEN AND. If the RangeShardingAlgorithm is not configured, the BETWEEN-AND SQLs will be executed in all tables.

在SQL语句中支持 =, IN, BETWEEN AND,以便进行切分操作。标准分片策略只支持单分片列,提供了两种分片算法:精确分片算法(PreciseShardingAlgorithm)和测距分片算法(RangeShardingAlgorithm)。精确分片算法(PreciseShardingAlgorithm)需要使用精确的硬件算法来处理“=”和in的切分操作。RangeShardingAlgorithm是在BETWEEN AND处理的可选方法。如果未配置RangeShardingAlgorithm,那么在查询SQL语句中,将在所有表中执行BETWEEN-AND SQLs。

  • ComplexShardingStrategy(综合硬件策略)

Support =, IN, BETWEEN AND in SQLs for sharding operation. ComplexShardingStrategy supports multiple sharding columns. Due to the complex relationship among the multiple sharding columns, Sharding-JDBC only provide algorithm API to allow developers combine different sharding columns and implement the specific algorithm.

在SQL中支持 =, IN, BETWEEN AND in,以便进行切分操作。ComplexShardingStrategy 支持多个切分列。由于多个切分列之间的复杂关系,Sharding-JDBC只提供算法API,允许开发人员组合不同的切分列,实现特定的算法。

  • InlineShardingStrategy (内部共享策略)

This strategy provides sharding support for =, IN in SQLs by means of Groovy’s Inline expression. InlineShardingStrategy only supports single sharding column. Some simple sharding algorithm can be configured, e.g. tuser $ {user_id% 8} shows us the t_user table is divided into 8 tables via mod(user_id), and the child tables is t_user_0 to t_user_7.

此策略通过内联表达式提供支持SQL中的 =, IN ,提供分片支持。InlineShardingStrategy 只支持单个分片的列字段。可以配置一些简单的切分算法,例如, t_user $ {user_id % 8} 向我们显示,t_user表通过取模(user_id)分为8个表,子表分别是t_user_0到t_user_7。

  • HintShardingStrategy

Support spliting table by means of Hint method, not SQL Parsing.

通过提示方法支持分表,而不对SQL进行解析。

  • NoneShardingStrategy

注:这种策略不要拆分数据库或表。

联表操作

它由一组表组成,其中逻辑表和实际表之间的映射关系是相同的。例如order table与order ID进行了分割,order_item table也与order ID进行了分割,因此可以将order table与order_item table配置为彼此的BindingTable。

在这种情况下,SQL语句应该如下:

SELECT i.* FROM t_order o JOIN t_order_item i ON o.order_id=i.order_id WHERE o.user_id=? AND o.order_id=?

t_order位于FROM的左侧,Sharding-JDBC将它视为绑定表的驱动表。所有计算将只使用已配置的驱动表策略。因此,t_order_item的路由计算也将使用t_order的条件。这个实现的核心在于它们相同的分片列。

最后提交我的实验代码:

@Configuration
@EnableTransactionManagement
@MapperScan(value = "com.lanxinbase.repository.mapper")
public class MybatisConfig implements TransactionManagementConfigurer {

    @Resource
    private ShardingDataSource shardingDataSource;

    public MybatisConfig(){
    }


    @Bean(value = "sessionFactoryBean")
    public SqlSessionFactoryBean sessionFactoryBean() throws Exception {
        SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
        factoryBean.setDataSource(shardingDataSource);
        factoryBean.setConfiguration(this.getConfiguration());

        String locationPattern = ResourcePatternResolver.CLASSPATH_URL_PREFIX + "com/lanxinbase/repository/resource/**.xml";
        PathMatchingResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
        factoryBean.setMapperLocations(resolver.getResources(locationPattern));
        return factoryBean;
    }

    @Bean
    public SqlSessionTemplate sqlSessionTemplate(SqlSessionFactory sqlSessionFactory) throws Exception {
        return new SqlSessionTemplate(sqlSessionFactory);
    }


    @Override
    public PlatformTransactionManager annotationDrivenTransactionManager() {
        DataSourceTransactionManager manager = new DataSourceTransactionManager();
        manager.setDataSource(shardingDataSource);
        manager.setDefaultTimeout(40);
        manager.setRollbackOnCommitFailure(true);
        return manager;
    }

    public org.apache.ibatis.session.Configuration getConfiguration() {
        org.apache.ibatis.session.Configuration configuration = new org.apache.ibatis.session.Configuration();
        configuration.setLogImpl(StdOutImpl.class);
        configuration.setLocalCacheScope(LocalCacheScope.SESSION);
        configuration.setCacheEnabled(true);
        return configuration;
    }

    @Bean
    public ShardingDataSource shardingDataSource() throws SQLException {
        ShardingRuleConfiguration conf = new ShardingRuleConfiguration();

        /**
         * 添加分表策略
         */
        conf.getTableRuleConfigs().add(tableRuleConfiguration());
        conf.getBindingTableGroups().add(Constant.SHARDING_TABLE_GPS);

        //http://shardingsphere.apache.org/document/legacy/2.x/en/02-guide/master-slave/
       //conf.getMasterSlaveRuleConfigs().add();配置读写分离

        //这个是分库的
       //conf.setDefaultDatabaseShardingStrategyConfig(new StandardShardingStrategyConfiguration("age", ShardingPreciseShardingAlgorithm.class.getName()));

        /**
         * 分表处理对象类
         */
        conf.setDefaultTableShardingStrategyConfig(new StandardShardingStrategyConfiguration("imei", ShardingPreciseShardingAlgorithm.class.getName()));
        conf.build(this.getDataSourceMap());

        ShardingDataSource dataSource = new ShardingDataSource(conf.build(this.getDataSourceMap()));
        return dataSource;
    }

    /**
     * 数据表配置,我这里只做了分表
     * 如果有多个表,就重复创建tableRuleConfiguration方法,
     * 然后通过getTableRuleConfigs.add(tableRuleConfiguration());
     *
     * @return
     */
    @Bean
    public TableRuleConfiguration tableRuleConfiguration() {
        TableRuleConfiguration ruleConfiguration = new TableRuleConfiguration();

        /**
         * 要逻辑表
         */
        ruleConfiguration.setLogicTable(Constant.SHARDING_TABLE_GPS);

        /**
         * 区分规则
         */
        ruleConfiguration.setActualDataNodes("dataSource2.lx_dev_gps_${0..9}");

        /**
         * 用于区分的字段
         */
        ruleConfiguration.setKeyGeneratorColumnName("imei");

        return ruleConfiguration;
    }

    /**
     * 可以配置多个dataSource,如dataSource1、dataSource2
     * 然后就可以把读写分离分开了
     * @return
     */
    private Map<String, DataSource> getDataSourceMap() {
        Map<String, DataSource> result = new HashMap<>();
        result.put("dataSource2", this.dataSource2());
        return result;
    }

    /**
     * 创建一个BasicDataSource给sharding-jdbc
     * @return
     */
    private BasicDataSource dataSource2() {
        BasicDataSource dataSource = new BasicDataSource();
        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
        dataSource.setUrl("jdbc:mysql://127.0.0.1:3306/test?serverTimezone=GMT&characterEncoding=utf8");
        dataSource.setUsername("root");
        dataSource.setPassword("root");

        dataSource.setInitialSize(0);
        dataSource.setMaxIdle(5);
        dataSource.setMinIdle(100);
        dataSource.setMaxOpenPreparedStatements(100);
        dataSource.setTestWhileIdle(true);
        dataSource.setValidationQuery("SELECT 1");
        dataSource.setTimeBetweenEvictionRunsMillis(3600000);
        dataSource.setMinEvictableIdleTimeMillis(18000000);
        dataSource.setTestOnBorrow(true);
        dataSource.setMaxWaitMillis(300000);

        return dataSource;
    }
}
//这里才是最终处理分表的逻辑代码
public class ShardingPreciseShardingAlgorithm implements PreciseShardingAlgorithm<Integer> {

    private static final Logger logger = Logger.getLogger(ShardingPreciseShardingAlgorithm.class.getName());

    @Override
    public String doSharding(Collection<String> availableTargetNames, PreciseShardingValue<Integer> shardingValue) {
        int flag = shardingValue.getValue() % 10;

        //只对特别的表进行拦截处理
        if (shardingValue.getLogicTableName().contains(Constant.SHARDING_TABLE_GPS)){
            for (String tableName : availableTargetNames) {
                if (tableName.endsWith(flag + "")) {
                    logger.info(">>>>tableName:" + tableName);
                    if (flag == 0) {
                        return tableName.substring(0, tableName.lastIndexOf("_"));
                    }
                    return tableName;
                }
            }
            throw new IllegalArgumentException("No match to the table.");
        }

        return shardingValue.getLogicTableName();
    }
}

 

*注:本人只做了分表的策略,没有做分库,是用于存放硬件设备GPS数据的表,一共10个。

更多文献:http://shardingsphere.apache.org/document/legacy/2.x/en/02-guide/sharding/

 

利用Sharding-Jdbc实现分表

看到了当当开源的Sharding-JDBC组件,它可以在几乎不修改代码的情况下完成分库分表的实现。摘抄其中一段介绍:

      Sharding-JDBC直接封装JDBC API,可以理解为增强版的JDBC驱动,旧代码迁移成本几乎为零:

  • 可适用于任何基于java的ORM框架,如:JPA, Hibernate, Mybatis, Spring JDBC Template或直接使用JDBC
  • 可基于任何第三方的数据库连接池,如:DBCP, C3P0, BoneCP, Druid等。
  • 理论上可支持任意实现JDBC规范的数据库。虽然目前仅支持MySQL,但已有支持Oracle,SQLServer,DB2等数据库的计划。

先做一个最简单的试用,不做分库,仅做分表。选择数据表bead_information,首先复制成三个表:bead_information_0、bead_information_1、bead_information_2

1552822689-9656-20180725

测试实现过程

前提:已经实现srping+mybatis对单库单表做增删改查的项目。

 1、修改pom.xml增加dependency

        <dependency>
            <groupId>com.dangdang</groupId>
            <artifactId>sharding-jdbc-core</artifactId>
            <version>1.4.2</version>
        </dependency>
        <dependency>
            <groupId>com.dangdang</groupId>
            <artifactId>sharding-jdbc-config-spring</artifactId>
            <version>1.4.0</version>
        </dependency>

2、新建一个sharding-jdbc.xml文件,实现分库分表的配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context" 
    xmlns:tx="http://www.springframework.org/schema/tx"
    xmlns:rdb="http://www.dangdang.com/schema/ddframe/rdb"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
                        http://www.springframework.org/schema/beans/spring-beans.xsd 
                        http://www.springframework.org/schema/tx 
                        http://www.springframework.org/schema/tx/spring-tx.xsd
                        http://www.springframework.org/schema/context 
                        http://www.springframework.org/schema/context/spring-context.xsd
                        http://www.dangdang.com/schema/ddframe/rdb 
                        http://www.dangdang.com/schema/ddframe/rdb/rdb.xsd">
    
 
    
     <!-- 配置数据源 -->
    <bean name="dataSource" class="com.alibaba.druid.pool.DruidDataSource" init-method="init" destroy-method="close">
        <property name="url" value="jdbc:mysql://localhost:3306/beadhouse" />
        <property name="username" value="root" />
        <property name="password" value="123456" />
    </bean>
                 
    <rdb:strategy id="tableShardingStrategy" sharding-columns="id" algorithm-class="com.springdemo.utill.MemberSingleKeyTableShardingAlgorithm"/>
    
    <rdb:data-source id="shardingDataSource">
        <rdb:sharding-rule data-sources="dataSource">
            <rdb:table-rules>
                <rdb:table-rule logic-table="bead_information" actual-tables="bead_information_${0..2}"  table-strategy="tableShardingStrategy"/>
            </rdb:table-rules>
        </rdb:sharding-rule>
    </rdb:data-source>
    
    <bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
        <property name="dataSource" ref="shardingDataSource" />
    </bean>
</beans>

3、将文件引入spring配置文件中。

需要修改几个地方,把sqlSessionFactory和transactionManager原来关联的dataSource统一修改为shardingDataSource(这一步作用就是把数据源全部托管给sharding去管理)

1552822688-6796-20180725

 

4、实现分表(分库)逻辑,我们的分表逻辑类需要实现SingleKeyTableShardingAlgorithm接口的三个方法doBetweenSharding、doEqualSharding、doInSharding

(取模除数需要按照自己需求改变,我这里分3个表,所以除以3)

import java.util.Collection;
import java.util.LinkedHashSet;
import com.dangdang.ddframe.rdb.sharding.api.ShardingValue;
import com.dangdang.ddframe.rdb.sharding.api.strategy.table.SingleKeyTableShardingAlgorithm;
import com.google.common.collect.Range;
public class MemberSingleKeyTableShardingAlgorithm implements SingleKeyTableShardingAlgorithm<Integer> {

    @Override
    public Collection<String> doBetweenSharding(Collection<String> tableNames, ShardingValue<Integer> shardingValue) {
        Collection<String> result = new LinkedHashSet<String>(tableNames.size());
        Range<Integer> range = (Range<Integer>) shardingValue.getValueRange();
        for (Integer i = range.lowerEndpoint(); i <= range.upperEndpoint(); i++) {
            Integer modValue = i % 3;
            String modStr = modValue < 3 ? "" + modValue : modValue.toString();
            for (String each : tableNames) {
                if (each.endsWith(modStr)) {
                    result.add(each);
                }
            }
        }
        return result;
    }
    @Override
    public String doEqualSharding(Collection<String> tableNames, ShardingValue<Integer> shardingValue) {
        Integer modValue = shardingValue.getValue() % 3;
        String modStr = modValue < 3 ? "" + modValue : modValue.toString();
        for (String each : tableNames) {
            if (each.endsWith(modStr)) {
                return each;
            }
        }
        throw new IllegalArgumentException();
    }
    @Override
    public Collection<String> doInSharding(Collection<String> tableNames, ShardingValue<Integer> shardingValue) {
        Collection<String> result = new LinkedHashSet<String>(tableNames.size());
        for (Integer value : shardingValue.getValues()) {
            Integer modValue = value % 3;
            String modStr = modValue < 3 ? "" + modValue : modValue.toString();
            for (String tableName : tableNames) {
                if (tableName.endsWith(modStr)) {
                    result.add(tableName);
                }
            }
        }
        return result;
    }
}

5、配置完成,可以实现增删改查测试。

理解Sharding jdbc原理

相比于Spring基于AbstractRoutingDataSource实现的分库分表功能,Sharding jdbc在单库单表扩展到多库多表时,兼容性方面表现的更好一点。例如,spring实现的分库分表sql写法如下:

select id, name, price, publish, intro
from book${tableIndex}
where id = #{id,jdbcType=INTEGER}

sql中的表名book需要加一个分表的后缀tableIndex,也就是需要在sql注入的参数中指定插入哪个表。相比,Sharding jdbc在这一块封装的更好一点。其sql中,根本不需要指定tableIndex,而是根据分库分表策略自动路由。

select id, name, price, publish, intro
from book
where id = #{id,jdbcType=INTEGER}

Sharding jdbc的这种特性,在水平扩展的时候无疑更具有吸引力。试想一下,一个项目开发一段时间后,单库单表数据量急剧上升,需要分库分表解决数据库的访问压力。而现有sql配置都是基于单库单表实现的,如果基于spring的AbstractRoutingDataSource实现,需要修改每一个相关表的sql,修改涉及较多地方,出错概率较大。而基于Sharding jdbc实现时,sql无需修改,只需要在spring中添加Sharding jdbc的相关配置即可,减少了修改面,大大简化分库分表的实现难度。

那么,Sharding jdbc是如何实现这种分库分表的逻辑呢?下面我们用一段简单、易懂的代码描述Sharding jdbc的原理。

通常我们在写一段访问数据库的数据时,逻辑是这样的:

ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext(“application.xml”);
DataSource dataSource = ctx.getBean(“dataSource”, DataSource.class);
Connection connection = dataSource.getConnection();

String sql = “select id, name, price, publish, intro from book where id = 111″;
PreparedStatement ps = connection.prepareStatement(sql);
ResultSet rs = ps.executeQuery();
// handle ResultSet…

Sharding jdbc是基于JDBC协议实现的,当我们获得dataSource时,这个dataSource是Sharding jdbc自己定义的一个SpringShardingDataSource类型的数据源,该数据源在返回getConnection()及prepareStatement()时,分别返回ShardingConnection和ShardingPreparedStatement的实例对象。然后在executeQuery()时,ShardingPreparedStatement做了这样的一件事:

根据逻辑sql,经过分库分表策略逻辑计算,获得分库分表的路由结果SQLRouteResult;
SQLRouteResult中包含真实的数据源以及转换后的真正sql,利用真实的数据源去执行获得ResultSet;
将ResultSet列表封装成一个可以顺序读的ResultSet对象IteratorReducerResultSet。

class ShardingPreparedStatement implements PreparedStatement {

@Override
public ResultSet executeQuery() throws SQLException {
List<SQLRouteResult> routeResults = routeSql(logicSql);

List<ResultSet> resultSets = new ArrayList<>(routeResults.size());
for (SQLRouteResult routeResult : routeResults) {
PreparedStatement ps = routeResult.getDataSource().getConnection.prepareStatement(routeResult.getParsedSql());
ResultSet rs = ps.executeQuery();
resultSets.add(rs);
}

return new IteratorReducerResultSet(resultSets);
}
…..

}

其中,分库分表策略的sql路由过程,我们将Sharding jdbc中的相关代码全部抽出来,放到一起来观看这个过程的实现:

// 环境准备
@SuppressWarnings(“resource”)
ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext(“application.xml”);
SpringShardingDataSource dataSource = ctx.getBean(SpringShardingDataSource.class);
Field field = SpringShardingDataSource.class.getSuperclass().getDeclaredField(“shardingContext”);
field.setAccessible(true);
ShardingContext sctx = (ShardingContext)field.get(dataSource);
ShardingRule shardingRule = sctx.getShardingRule();

String logicSql = “select id, name, price, publish, intro from book where id = ?”;
List<Object> parameters = new ArrayList<>();
parameters.add(2000);

// sql解析
MySqlStatementParser parser = new MySqlStatementParser(logicSql);
MySQLSelectVisitor visitor = new MySQLSelectVisitor();
SQLStatement statement = parser.parseStatement();
visitor.getParseContext().setShardingRule(shardingRule);
statement.accept(visitor);

SQLParsedResult parsedResult = visitor.getParseContext().getParsedResult();
if (visitor.getParseContext().isHasOrCondition()) {
new OrParser(statement, visitor).fillConditionContext(parsedResult);
}
visitor.getParseContext().mergeCurrentConditionContext();
System.out.println(“Parsed SQL result: ” + parsedResult);
System.out.println(“Parsed SQL: ” + visitor.getSQLBuilder());
parsedResult.getRouteContext().setSqlBuilder(visitor.getSQLBuilder());
parsedResult.getRouteContext().setSqlStatementType(SQLStatementType.SELECT);

// 分库分表路由
SQLRouteResult result = new SQLRouteResult(parsedResult.getRouteContext().getSqlStatementType(), parsedResult.getMergeContext(), parsedResult.getGeneratedKeyContext());
for (ConditionContext each : parsedResult.getConditionContexts()) {
Collection<Table> tables = parsedResult.getRouteContext().getTables();
final Set<String> logicTables = new HashSet<>();
tables.forEach(a -> logicTables.add(a.getName()));

SingleTableRouter router = new SingleTableRouter(shardingRule,
logicTables.iterator().next(),
each,
parsedResult.getRouteContext().getSqlStatementType());

RoutingResult routingResult = router.route();

// sql改写 –> routingResult.getSQLExecutionUnits()
// —> SingleRoutingTableFactor.replaceSQL(sqlBuilder).buildSQL()
// 结果合并
result.getExecutionUnits().addAll(routingResult.getSQLExecutionUnits(parsedResult.getRouteContext().getSqlBuilder()));
}
// amendSQLAccordingToRouteResult(parsedResult, parameters, result);
for (SQLExecutionUnit each : result.getExecutionUnits()) {
System.out.println(each.getDataSource() + ” ” + each.getSql() + ” ” + parameters);
}

  • 准备环境。由于Sharding jdbc分库分表中ShardingRule这个类是贯穿整个路由过程,我们在Spring中写好Sharding jdbc的配置,利用反射获取一个这个对象。(Sharding jdbc版本以及配置,在文章最后列出,方便debug这个过程)
  • sql解析。Sharding jdbc使用阿里的Druid库解析sql。在这个过程中,Sharding jdbc实现了一个自己的sql解析内容缓存容器SqlBuilder。当语法分析中解析到一个表名的时候,在SqlBuilder中缓存一个sql相关的逻辑表名的token。并且,Sharding jdbc会将sql按照语义解析为多个segment。例如,”select id, name, price, publish, intro from book where id = ?”将解析为,”select id, name, price, publish, intro | from | book | where | id = ?”。
  • 分库分表路由。根据ShardingRule中指定的分库分表列的参数值,以及分库分表策略,实行分库分表,得到一个RoutingResult 。RoutingResult 中包含一个真实数据源,以及逻辑表名和实际表名。
  • sql改写。在SqlBuilder中,查找sql中解析的segment,将和逻辑表名一致的segment替换成实际表名。(segment中可以标注该地方是不是表名)
    以上代码执行结果如下:

Parsed SQL result: SQLParsedResult(routeContext=RouteContext(tables=[Table(name=book, alias=Optional.absent())], sqlStatementType=null, sqlBuilder=null), generatedKeyContext=GeneratedKeyContext(columns=[], columnNameToIndexMap={}, valueTable={}, rowIndex=0, columnIndex=0, autoGeneratedKeys=0, columnIndexes=null, columnNames=null), conditionContexts=[ConditionContext(conditions={})], mergeContext=MergeContext(orderByColumns=[], groupByColumns=[], aggregationColumns=[], limit=null))
Parsed SQL: SELECT id, name, price, publish, intro FROM [Token(book)] WHERE id = ?
dataSource1 SELECT id, name, price, publish, intro FROM book_00 WHERE id = ? [2000]
dataSource2 SELECT id, name, price, publish, intro FROM book_02 WHERE id = ? [2000]
dataSource1 SELECT id, name, price, publish, intro FROM book_02 WHERE id = ? [2000]
dataSource2 SELECT id, name, price, publish, intro FROM book_01 WHERE id = ? [2000]
dataSource0 SELECT id, name, price, publish, intro FROM book_00 WHERE id = ? [2000]
dataSource0 SELECT id, name, price, publish, intro FROM book_01 WHERE id = ? [2000]
dataSource2 SELECT id, name, price, publish, intro FROM book_00 WHERE id = ? [2000]
dataSource1 SELECT id, name, price, publish, intro FROM book_01 WHERE id = ? [2000]
dataSource0 SELECT id, name, price, publish, intro FROM book_02 WHERE id = ? [2000]

实际上,我们可以用更通俗易懂的代码表示sql改写的这个过程:

String logicSql = “select id, name, price, publish, intro from book where id = 111″;
MySqlStatementParser parser = new MySqlStatementParser(logicSql);
SQLStatement statement = parser.parseStatement();
MySQLSimpleVisitor visitor = new MySQLSimpleVisitor();
statement.accept(visitor);

String logicTable = “book”;
String realTable = “book_00″;
String token = “\\$\\{” + logicTable + “\\}”;

String sqlBuilder = visitor.getAppender().toString();
String sql = sqlBuilder.replaceAll(token, realTable);

System.out.println(sqlBuilder);
System.out.println(sql);

 

结果如下:

SELECT id, name, price, publish, intro
FROM ${book}
WHERE id = 111
SELECT id, name, price, publish, intro
FROM book_00
WHERE id = 111

以上,大致将Sharding jdbc的原理及实现过程介绍了一下,如果想要了解正真的实现过程和细节,还需要对照代码仔细推敲。

本文的实现环境:

<dependency>
<groupId>com.dangdang</groupId>
<artifactId>sharding-jdbc-core</artifactId>
<version>1.4.2</version>
</dependency>
<dependency>
<groupId>com.dangdang</groupId>
<artifactId>sharding-jdbc-config-spring</artifactId>
<version>1.4.0</version>
</dependency>

application.xml:

<?xml version=”1.0″ encoding=”UTF-8″?>
<beans xmlns=”http://www.springframework.org/schema/beans”
xmlns:xsi=”http://www.w3.org/2001/XMLSchema-instance”
xmlns:tx=”http://www.springframework.org/schema/tx”
xmlns:context=”http://www.springframework.org/schema/context”
xmlns:rdb=”http://www.dangdang.com/schema/ddframe/rdb”
xsi:schemaLocation=”
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.0.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
http://www.dangdang.com/schema/ddframe/rdb
http://www.dangdang.com/schema/ddframe/rdb/rdb.xsd”>

<context:property-placeholder location=”classpath:jdbc.properties” ignore-unresolvable=”true” />

<bean id=”dataSource0″ class=”org.springframework.jdbc.datasource.DriverManagerDataSource”>
<property name=”driverClassName” value=”com.mysql.jdbc.Driver” />
<property name=”url” value=”${jdbc.mysql.url0}” />
<property name=”username” value=”${jdbc.mysql.username0}” />
<property name=”password” value=”${jdbc.mysql.password0}” />
</bean>

<bean id=”dataSource1″ class=”org.springframework.jdbc.datasource.DriverManagerDataSource”>
<property name=”driverClassName” value=”${driver}” />
<property name=”url” value=”${jdbc.mysql.url1}” />
<property name=”username” value=”${jdbc.mysql.username1}” />
<property name=”password” value=”${jdbc.mysql.password1}” />
</bean>

<bean id=”dataSource2″ class=”org.springframework.jdbc.datasource.DriverManagerDataSource”>
<property name=”driverClassName” value=”${driver}” />
<property name=”url” value=”${jdbc.mysql.url2}” />
<property name=”username” value=”${jdbc.mysql.username2}” />
<property name=”password” value=”${jdbc.mysql.password2}” />
</bean>

<!– sharding jdbc –>
<rdb:strategy id=”tableShardingStrategy” sharding-columns=”id”
algorithm-class=”com.wy.sharding.MemberSingleKeyTableShardingAlgorithm” />

<rdb:data-source id=”shardingDataSource”>
<rdb:sharding-rule data-sources=”dataSource0,dataSource1,dataSource2″>
<rdb:table-rules>
<rdb:table-rule logic-table=”book”
actual-tables=”book_0${0..2}”
table-strategy=”tableShardingStrategy”/>
</rdb:table-rules>
</rdb:sharding-rule>
</rdb:data-source>
</beans>

MemberSingleKeyTableShardingAlgorithm.java:

public class MemberSingleKeyTableShardingAlgorithm implements SingleKeyTableShardingAlgorithm<Integer> {

public String doEqualSharding(Collection<String> availableTargetNames, ShardingValue<Integer> shardingValue) {
String routeDBSuffix = getRouteDBSuffix(shardingValue.getValue());
for (String each : availableTargetNames) {
if (each.endsWith(routeDBSuffix)) {
return each;
}
}
throw new IllegalArgumentException();
}

public Collection<String> doInSharding(Collection<String> availableTargetNames, ShardingValue<Integer> shardingValue) {
Collection<String> result = new LinkedHashSet<String>(availableTargetNames.size());
for (int value : shardingValue.getValues()) {
String routeDBSuffix = getRouteDBSuffix(value);
for (String tableName : availableTargetNames) {
if (tableName.endsWith(routeDBSuffix)) {
result.add(tableName);
}
}
}
return result;
}

public Collection<String> doBetweenSharding(Collection<String> availableTargetNames,
ShardingValue<Integer> shardingValue) {
Collection<String> result = new LinkedHashSet<String>(availableTargetNames.size());
Range<Integer> range = (Range<Integer>) shardingValue.getValueRange();
for (int i = range.lowerEndpoint(); i <= range.upperEndpoint(); i++) {
String routeDBSuffix = getRouteDBSuffix(i);
for (String each : availableTargetNames) {
if (each.endsWith(routeDBSuffix)) {
result.add(each);
}
}
}
return result;
}

public String getRouteDBSuffix(Integer shardingCode) {
int modValue = shardingCode % 3;
return “0” + modValue;
}

}

 

 
Copyright © 2008-2021 lanxinbase.com Rights Reserved. | 粤ICP备14086738号-3 |