shardingJdbc源码简析：分片、归并

package org.apache.shardingsphere.spring.boot.datasource;

public final class HikariDataSourcePropertiesSetter implements DataSourcePropertiesSetter {
    public void propertiesSet(final Environment environment, final String prefix, final String dataSourceName, final DataSource dataSource) {
        Properties properties = new Properties();
        String datasourcePropertiesKey = prefix + dataSourceName.trim() + ".data-source-properties";
        if (PropertyUtil.containPropertyPrefix(environment, datasourcePropertiesKey)) {
            Map datasourceProperties = PropertyUtil.handle(environment, datasourcePropertiesKey, Map.class);
            properties.putAll(datasourceProperties);
            Method method = dataSource.getClass().getMethod("setDataSourceProperties", Properties.class);
            method.invoke(dataSource, properties);
        }
    }
    @Override
    public String getType() {
        return "com.zaxxer.hikari.HikariDataSource";
    }
}

通过实现接口 org.apache.shardingsphere.spring.boot.datasource.DataSourcePropertiesSetter 来反射方式给DataSource注入属性。 

• SpringBootConfiguration#getDataSource在创建DataSource时，会根据#getType()指定类型的数据源匹配属性注入文件。

自定义数据源和配置注入器：

package com.noob.shardingJdbc.dataSoure;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import com.zaxxer.hikari.util.PropertyElf;

import java.util.Properties;

public class CustomizeHikariDataSource extends HikariDataSource {

    public CustomizeHikariDataSource() { super(); }

    public CustomizeHikariDataSource(HikariConfig configuration) { super(configuration); }

    
    @Override
    public void setDataSourceProperties(Properties dsProperties) {
        super.setDataSourceProperties(dsProperties); // 只是设置dataSourceProperties
        PropertyElf.setTargetFromProperties(this, this.getDataSourceProperties()); //  还需要真实写入到dataSource的成员属性里
    }
}
----
import lombok.SneakyThrows;
import org.apache.shardingsphere.spring.boot.datasource.DataSourcePropertiesSetter;
import org.apache.shardingsphere.spring.boot.datasource.HikariDataSourcePropertiesSetter;
import org.springframework.core.env.Environment;

import javax.sql.DataSource;

public final class CustomizeHikariDataSourcePropertiesSetter implements DataSourcePropertiesSetter {

    HikariDataSourcePropertiesSetter setter = new HikariDataSourcePropertiesSetter();

    @SneakyThrows(ReflectiveOperationException.class)
    public void propertiesSet(final Environment environment, final String prefix, final String dataSourceName, final DataSource dataSource) {
        setter.propertiesSet(environment, prefix, dataSourceName, dataSource);
    }

    @Override
    public String getType() {
        return "com.noob.shardingJdbc.dataSoure.CustomizeHikariDataSource";
    }
}

package：  org.apache.shardingsphere.shardingjdbc.spring.boot

根据配置差异会选择实例化不同的DataSource类型：

• 按分片策略路由。
  ShardingRuleCondition ->  ShardingDataSource  -> ShardingPreparedStatement -> ShardingConnection
  ShardingRuntimeContext -> ShardingRule -> ShardingRouter#router

• 区分读写请求，写走主库，读按负载策略LoadBalanceAlgorithm路由从库。
  MasterSlaveRuleCondition -> MasterSlaveDataSource ->MasterSlavePreparedStatement  -> MasterSlaveConnection
  MasterSlaveRuntimeContext -> MasterSlaveRule ->  MasterSlaveRouter#router   

2020-03-15 17:24:46.705 [main] INFO  [trace=,span=,parent=] ShardingSphere-SQL - Rule Type: sharding
2020-03-15 17:24:46.705 [main] INFO  [trace=,span=,parent=] ShardingSphere-SQL - Logic SQL: select * from limit_use u , repayment_plan t where t.loan_no ='LOAN683009630195941376' and u.loan_no = t.loan_no
2020-03-15 17:24:46.705 [main] INFO  [trace=,span=,parent=] ShardingSphere-SQL - SQLStatement: SelectSQLStatementContext(super=CommonSQLStatementContext(sqlStatement=org.apache.shardingsphere.sql.parser.sql.statement.dml.SelectStatement@734c98ff, tablesContext=TablesContext(tables=[Table(name=limit_use, alias=Optional.of(u)), Table(name=repayment_plan, alias=Optional.of(t))], schema=Optional.absent())), projectionsContext=ProjectionsContext(startIndex=7, stopIndex=7, distinctRow=false, projections=[ShorthandProjection(owner=Optional.absent())], columnLabels=[id, flow_no, customer_id, customer_name, certificate_no, loan_no, mobile_phone, bank_account_id, bank_account_no, repay_bank_account_id, repay_bank_account_no, contract_no, use_amount, total_periods, term_type, term, channel, loan_type, loan_type_id, repayment_method, due_day, apply_date, accept_date, settle_date, interest_rate, loan_date, interest_start_date, use_status, purpose, trade_channel, trade_status, trade_no, trade_message, business_error_code, reason, version, merge_loan_flag, merge_loan_account, create_time, update_time, id, loan_no, period, should_repayment_date, should_repayment_principal, should_repayment_interest, should_repayment_penalty, should_repayment_fee, actual_repayment_principal, actual_repayment_interest, actual_repayment_penalty, actual_repayment_fee, interest_deduction_amount, penalty_deduction_amount, remaining_principal, is_overdue, actual_repayment_date, repayment_plan_status, grace_days, version, create_time, update_time]), groupByContext=org.apache.shardingsphere.sql.parser.relation.segment.select.groupby.GroupByContext@7a203af7, orderByContext=org.apache.shardingsphere.sql.parser.relation.segment.select.orderby.OrderByContext@360e9b23, paginationContext=org.apache.shardingsphere.sql.parser.relation.segment.select.pagination.PaginationContext@67ee3b1, containsSubquery=false)
2020-03-15 17:24:46.705 [main] INFO  [trace=,span=,parent=] ShardingSphere-SQL - Actual SQL: ds0 ::: select * from limit_use_08 u , repayment_plan_08 t where t.loan_no ='LOAN683009630195941376' and u.loan_no = t.loan_no

下文以4.0.1版本的sharding模式为例 , 4.0.0-RC1 与 4.0.1 版本 在解析sq路由的变动还是挺大的。

从mybatis执行进入ShardingJdbc逻辑："Executor -> PreparedStatementHandler  -> PreparedStatement " 这个流程里拿到的数据源、连接、statement都是ShardingJdbc的： ShardingDataSource -> ShardingConnection -> ShardingPreparedStatement。

#executeQuery的过程：

1. 依据分片规则解析sql确定路由的具体库&表， 重写sql生成真正的脚本<多个库|表会有多个>；
2. 执行sql后的ResultSet选择数据结果集归并方式：内存 | 流式 

    public ResultSet executeQuery() throws SQLException {
        ResultSet result;
        try {
            clearPrevious(); // 清除上次执行各处理缓存
            shard(); // BaseShardingEngine#shard分片解析出路由sqlRouteResult
            initPreparedStatementExecutor(); // 依路由结果初始化各个分库表的执行上下文Statement
            //  数据结果集QueryResult，需要选择归并方式
            MergeEngine mergeEngine = MergeEngineFactory.newInstance(connection.getRuntimeContext().getDatabaseType(), 
                    connection.getRuntimeContext().getRule(), sqlRouteResult, connection.getRuntimeContext().getMetaData().getRelationMetas(), preparedStatementExecutor.executeQuery()); 
            result = getResultSet(mergeEngine);
        } finally {
            clearBatch();
        }
        currentResultSet = result;
        return result;
    }

    private void shard() {
        sqlRouteResult = shardingEngine.shard(sql, getParameters());
    }

分片入口： BaseShardingEngine#shard()

    public SQLRouteResult shard(final String sql, final List