摘要: 原创出处 http://www.iocoder.cn/Sharding-JDBC/jdbc-implement-and-read-write-splitting/ 「芋道源码」欢迎转载，保留摘要，谢谢！

排版已崩....建议点击原文，抱歉~~~~

本文主要基于 Sharding-JDBC 1.5.0 正式版

1. 概述
2. unspported 包
3. adapter 包
3.1 WrapperAdapter
3.2 AbstractDataSourceAdapter
3.3 AbstractConnectionAdapter
3.4 AbstractStatementAdapter
3.5 AbstractPreparedStatementAdapter
3.6 AbstractResultSetAdapter
4. 插入流程
5. 查询流程
6. 读写分离
666. 彩蛋

🙂🙂🙂关注微信公众号：【芋道源码】有福利：
RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。
新的源码解析文章实时收到通知。每周更新一篇左右。
认真的源码交流微信群。

1. 概述

本文主要分享 JDBC 与 读写分离 的实现。为什么会把这两个东西放在一起讲呢？客户端直连数据库的读写分离主要通过获取读库和写库的不同连接来实现，和 JDBC Connection 刚好放在一块。

OK，我们先来看一段 Sharding-JDBC 官方对自己的定义和定位

Sharding-JDBC定位为轻量级java框架，使用客户端直连数据库，以jar包形式提供服务，未使用中间层，无需额外部署，无其他依赖，DBA也无需改变原有的运维方式，可理解为增强版的JDBC驱动，旧代码迁移成本几乎为零。

可以看出，Sharding-JDBC 通过实现 JDBC规范，对上层提供透明化数据库分库分表的访问。😈 黑科技？实际我们使用的数据库连接池也是通过这种方式实现对上层无感知的提供连接池。甚至还可以通过这种方式实现对 Lucene、MongoDB 等等的访问。

扯远了，下面来看看 Sharding-JDBC jdbc 包的结构：

unsupported：声明不支持的数据操作方法
adapter：适配类，实现和分库分表无关的方法
core：核心类，实现和分库分表相关的方法

根据 core 包，可以看出分到四种我们超级熟悉的对象

Datasource
Connection
Statement
ResultSet

实现层级如下：JDBC 接口 <=(继承)== unsupported抽象类 <=(继承)== unsupported抽象类 <=(继承)== core类。

本文内容顺序

unspported 包
adapter 包
插入流程，分析的类：

ShardingDataSource
ShardingConnection
ShardingPreparedStatement（ShardingStatement 类似，不重复分析）
GeneratedKeysResultSet、GeneratedKeysResultSetMetaData

查询流程，分析的类：

ShardingPreparedStatement
ShardingResultSet

读写分离，分析的类：

MasterSlaveDataSource

Sharding-JDBC 正在收集使用公司名单：传送门。

🙂 你的登记，会让更多人参与和使用 Sharding-JDBC。传送门

Sharding-JDBC 也会因此，能够覆盖更多的业务场景。传送门

登记吧，骚年！传送门

2. unspported 包

unspported 包内的抽象类，声明不支持操作的数据对象，所有方法都是 thrownewSQLFeatureNotSupportedException() 方式。

publicabstractclassAbstractUnsupportedGeneratedKeysResultSetextendsAbstractUnsupportedOperationResultSet{
@Override
publicboolean getBoolean(finalint columnIndex)throwsSQLException{
thrownewSQLFeatureNotSupportedException("getBoolean");
}
// .... 省略其它类似方法
}
publicabstractclassAbstractUnsupportedOperationConnectionextendsWrapperAdapterimplementsConnection{
@Override
publicfinalCallableStatement prepareCall(finalString sql)throwsSQLException{
thrownewSQLFeatureNotSupportedException("prepareCall");
}
// .... 省略其它类似方法
}

3. adapter 包

adapter 包内的抽象类，实现和分库分表无关的方法。

考虑到第4、5两小节更容易理解，本小节贴的代码会相对多

3.1 WrapperAdapter

WrapperAdapter，JDBC Wrapper 适配类。

对 Wrapper 接口实现如下两个方法：

@Override
publicfinal<T> T unwrap(finalClass<T> iface)throwsSQLException{
if(isWrapperFor(iface)){
return(T)this;
}
thrownewSQLException(String.format("[%s] cannot be unwrapped as [%s]", getClass().getName(), iface.getName()));
}
@Override
publicfinalboolean isWrapperFor(finalClass<?> iface)throwsSQLException{
return iface.isInstance(this);
}

提供子类 #recordMethodInvocation() 记录方法调用， #replayMethodsInvocation() 回放记录的方法调用：

/**
* 记录的方法数组
*/
privatefinalCollection<JdbcMethodInvocation> jdbcMethodInvocations =newArrayList<>();
/**
* 记录方法调用.
* 
* @param targetClass 目标类
* @param methodName 方法名称
* @param argumentTypes 参数类型
* @param arguments 参数
*/
publicfinalvoid recordMethodInvocation(finalClass<?> targetClass,finalString methodName,finalClass<?>[] argumentTypes,finalObject[] arguments){
try{
       jdbcMethodInvocations.add(newJdbcMethodInvocation(targetClass.getMethod(methodName, argumentTypes), arguments));
}catch(finalNoSuchMethodException ex){
thrownewShardingJdbcException(ex);
}
}
/**
* 回放记录的方法调用.
* 
* @param target 目标对象
*/
publicfinalvoid replayMethodsInvocation(finalObject target){
for(JdbcMethodInvocation each : jdbcMethodInvocations){
       each.invoke(target);
}
}

这两个方法有什么用途呢？例如下文会提到的 AbstractConnectionAdapter 的 #setAutoCommit()，当它无数据库连接时，先记录；等获得到数据连接后，再回放：

// AbstractConnectionAdapter.java
@Override
publicfinalvoid setAutoCommit(finalboolean autoCommit)throwsSQLException{
this.autoCommit = autoCommit;
if(getConnections().isEmpty()){// 无数据连接时，记录方法调用
recordMethodInvocation(Connection.class,"setAutoCommit",newClass[]{boolean.class},newObject[]{autoCommit});
return;
}
for(Connection each : getConnections()){
each.setAutoCommit(autoCommit);
}
}

JdbcMethodInvocation，反射调用JDBC相关方法的工具类：

publicclassJdbcMethodInvocation{
/**
* 方法
*/
@Getter
privatefinalMethod method;
/**
* 方法参数
*/
@Getter
privatefinalObject[] arguments;
/**
* 调用方法.
*
* @param target 目标对象
*/
publicvoid invoke(finalObject target){
try{
method.invoke(target, arguments);// 反射调用
}catch(finalIllegalAccessException|InvocationTargetException ex){
thrownewShardingJdbcException("Invoke jdbc method exception", ex);
}
}
}

提供子类 #throwSQLExceptionIfNecessary() 抛出异常链：

protectedvoid throwSQLExceptionIfNecessary(finalCollection<SQLException> exceptions)throwsSQLException{
if(exceptions.isEmpty()){// 为空不抛出异常
return;
}
SQLException ex =newSQLException();
for(SQLException each : exceptions){
       ex.setNextException(each);// 异常链
}
throw ex;
}

3.2 AbstractDataSourceAdapter

AbstractDataSourceAdapter，数据源适配类。

直接点击链接查看源码。

3.3 AbstractConnectionAdapter

AbstractConnectionAdapter，数据库连接适配类。

我们来瞅瞅大家最关心的事务相关方法的实现。

/**
* 是否自动提交
*/
privateboolean autoCommit =true;
/**
* 获得链接
*
* @return 链接
*/
protectedabstractCollection<Connection> getConnections();
@Override
publicfinalboolean getAutoCommit()throwsSQLException{
return autoCommit;
}
@Override
publicfinalvoid setAutoCommit(finalboolean autoCommit)throwsSQLException{
this.autoCommit = autoCommit;
if(getConnections().isEmpty()){// 无数据连接时，记录方法调用
       recordMethodInvocation(Connection.class,"setAutoCommit",newClass[]{boolean.class},newObject[]{autoCommit});
return;
}
for(Connection each : getConnections()){
       each.setAutoCommit(autoCommit);
}
}

#setAutoCommit() 调用时，实际会设置其所持有的 Connection 的 autoCommit 属性
#getConnections() 和分库分表相关，因而仅抽象该方法，留给子类实现

@Override
publicfinalvoid commit()throwsSQLException{
for(Connection each : getConnections()){
       each.commit();
}
}
@Override
publicfinalvoid rollback()throwsSQLException{
Collection<SQLException> exceptions =newLinkedList<>();
for(Connection each : getConnections()){
try{
           each.rollback();
}catch(finalSQLException ex){
           exceptions.add(ex);
}
}
   throwSQLExceptionIfNecessary(exceptions);
}

#commit()、 #rollback() 调用时，实际调用其所持有的 Connection 的方法
异常情况下， #commit() 和 #rollback() 处理方式不同，笔者暂时不知道答案，求证后会进行更新

#commit() 处理方式需要改成和 #rollback() 一样。代码如下：

@Override
publicfinalvoid commit()throwsSQLException{
Collection<SQLException> exceptions =newLinkedList<>();
for(Connection each : getConnections()){
try{
each.commit();
}catch(finalSQLException ex){
exceptions.add(ex);
}
}
throwSQLExceptionIfNecessary(exceptions);
}

事务级别和是否只读相关代码如下：

/**
* 只读
*/
privateboolean readOnly =true;
/**
* 事务级别
*/
privateint transactionIsolation = TRANSACTION_READ_UNCOMMITTED;
@Override
publicfinalvoid setReadOnly(finalboolean readOnly)throwsSQLException{
this.readOnly = readOnly;
if(getConnections().isEmpty()){
       recordMethodInvocation(Connection.class,"setReadOnly",newClass[]{boolean.class},newObject[]{readOnly});
return;
}
for(Connection each : getConnections()){
       each.setReadOnly(readOnly);
}
}
@Override
publicfinalvoid setTransactionIsolation(finalint level)throwsSQLException{
   transactionIsolation = level;
if(getConnections().isEmpty()){
       recordMethodInvocation(Connection.class,"setTransactionIsolation",newClass[]{int.class},newObject[]{level});
return;
}
for(Connection each : getConnections()){
       each.setTransactionIsolation(level);
}
}

3.4 AbstractStatementAdapter

AbstractStatementAdapter，静态语句对象适配类。

@Override
publicfinalint getUpdateCount()throwsSQLException{
long result =0;
boolean hasResult =false;
for(Statement each : getRoutedStatements()){
if(each.getUpdateCount()>-1){
           hasResult =true;
}
       result += each.getUpdateCount();
}
if(result >Integer.MAX_VALUE){
       result =Integer.MAX_VALUE;
}
return hasResult ?Long.valueOf(result).intValue():-1;
}
/**
* 获取路由的静态语句对象集合.
* 
* @return 路由的静态语句对象集合
*/
protectedabstractCollection<?extendsStatement> getRoutedStatements();

#getUpdateCount() 调用持有的 Statement 计算更新数量
#getRoutedStatements() 和分库分表相关，因而仅抽象该方法，留给子类实现

3.5 AbstractPreparedStatementAdapter

AbstractPreparedStatementAdapter，预编译语句对象的适配类。

#recordSetParameter()实现对占位符参数的设置：

/**
* 记录的设置参数方法数组
*/
privatefinalList<SetParameterMethodInvocation> setParameterMethodInvocations =newLinkedList<>();
/**
* 参数
*/
@Getter
privatefinalList<Object> parameters =newArrayList<>();
@Override
publicfinalvoid setInt(finalint parameterIndex,finalint x)throwsSQLException{
   setParameter(parameterIndex, x);
   recordSetParameter("setInt",newClass[]{int.class,int.class}, parameterIndex, x);
}
/**
* 记录占位符参数
*
* @param parameterIndex 占位符参数位置
* @param value 参数
*/
privatevoid setParameter(finalint parameterIndex,finalObject value){
if(parameters.size()== parameterIndex -1){
       parameters.add(value);
return;
}
for(int i = parameters.size(); i <= parameterIndex -1; i++){// 用 null 填充前面未设置的位置
       parameters.add(null);
}
   parameters.set(parameterIndex -1, value);
}
/**
* 记录设置参数方法调用
*
* @param methodName 方法名，例如 setInt、setLong 等
* @param argumentTypes 参数类型
* @param arguments 参数
*/
privatevoid recordSetParameter(finalString methodName,finalClass[] argumentTypes,finalObject... arguments){
try{
       setParameterMethodInvocations.add(newSetParameterMethodInvocation(PreparedStatement.class.getMethod(methodName, argumentTypes), arguments, arguments[1]));
}catch(finalNoSuchMethodException ex){
thrownewShardingJdbcException(ex);
}
}
/**
* 回放记录的设置参数方法调用
*
* @param preparedStatement 预编译语句对象
*/
protectedvoid replaySetParameter(finalPreparedStatement preparedStatement){
   addParameters();
for(SetParameterMethodInvocation each : setParameterMethodInvocations){
       updateParameterValues(each, parameters.get(each.getIndex()-1));// 同一个位置多次设置，值可能不一样，需要更新下
       each.invoke(preparedStatement);
}
}
/**
* 当使用分布式主键时，生成后会添加到 parameters，此时 parameters 数量多于 setParameterMethodInvocations，需要生成该分布式主键的 SetParameterMethodInvocation
*/
privatevoid addParameters(){
for(int i = setParameterMethodInvocations.size(); i < parameters.size(); i++){
       recordSetParameter("setObject",newClass[]{int.class,Object.class}, i +1, parameters.get(i));
}
}
privatevoid updateParameterValues(finalSetParameterMethodInvocation setParameterMethodInvocation,finalObject value){
if(!Objects.equals(setParameterMethodInvocation.getValue(), value)){
       setParameterMethodInvocation.changeValueArgument(value);// 修改占位符参数
}
}

逻辑类似 WrapperAdapter 的 #recordMethodInvocation()， #replayMethodsInvocation()，请认真阅读代码注释
SetParameterMethodInvocation，继承 JdbcMethodInvocation，反射调用参数设置方法的工具类：

publicfinalclassSetParameterMethodInvocationextendsJdbcMethodInvocation{
/**
* 位置
*/
@Getter
privatefinalint index;
/**
* 参数值
*/
@Getter
privatefinalObject value;
/**
* 设置参数值.
*
* @param value 参数值
*/
publicvoid changeValueArgument(finalObject value){
getArguments()[1]= value;
}
}

3.6 AbstractResultSetAdapter

AbstractResultSetAdapter，代理结果集适配器。

publicabstractclassAbstractResultSetAdapterextendsAbstractUnsupportedOperationResultSet{
/**
     * 结果集集合
     */
@Getter
privatefinalList<ResultSet> resultSets;
@Override
// TODO should return sharding statement in future
publicfinalStatement getStatement()throwsSQLException{
return getResultSets().get(0).getStatement();
}
@Override
publicfinalResultSetMetaData getMetaData()throwsSQLException{
return getResultSets().get(0).getMetaData();
}
@Override
publicint findColumn(finalString columnLabel)throwsSQLException{
return getResultSets().get(0).findColumn(columnLabel);
}
// .... 省略其它方法
}

4. 插入流程

插入使用分布式主键例子代码如下：

// 代码仅仅是例子，生产环境下请注意异常处理和资源关闭
String sql ="INSERT INTO t_order(uid, nickname, pid) VALUES (1, '2', ?)";
DataSource dataSource =newShardingDataSource(shardingRule);
Connection conn = dataSource.getConnection();
PreparedStatement ps = conn.prepareStatement(sql,Statement.RETURN_GENERATED_KEYS);// 返回主键需要  Statement.RETURN_GENERATED_KEYS
ps.setLong(1,100);
ps.executeUpdate();
ResultSet rs = ps.getGeneratedKeys();
if(rs.next()){
System.out.println("id:"+ rs.getLong(1));
}

调用 #executeUpdate() 方法，内部过程如下：

是不是对上层完全透明？！我们来看看内部是怎么实现的。

// ShardingPreparedStatement.java
@Override
publicint executeUpdate()throwsSQLException{
try{
Collection<PreparedStatementUnit> preparedStatementUnits = route();
returnnewPreparedStatementExecutor(
               getShardingConnection().getShardingContext().getExecutorEngine(), getRouteResult().getSqlStatement().getType(), preparedStatementUnits, getParameters()).executeUpdate();
}finally{
       clearBatch();
}
}

#route() 分库分表路由，获得预编译语句对象执行单元( PreparedStatementUnit )集合。

publicfinalclassPreparedStatementUnitimplementsBaseStatementUnit{
/**
* SQL 执行单元
*/
privatefinalSQLExecutionUnit sqlExecutionUnit;
/**
* 预编译语句对象
*/
privatefinalPreparedStatement statement;
}

#executeUpdate() 调用执行引擎并行执行多个预编译语句对象。执行时，最终调用预编译语句对象( PreparedStatement )。我们来看一个例子：

// PreparedStatementExecutor.java
publicint executeUpdate(){
Context context =MetricsContext.start("ShardingPreparedStatement-executeUpdate");
try{
List<Integer> results = executorEngine.executePreparedStatement(sqlType, preparedStatementUnits, parameters,newExecuteCallback<Integer>(){
@Override
publicInteger execute(finalBaseStatementUnit baseStatementUnit)throwsException{
// 调用 PreparedStatement#executeUpdate()
return((PreparedStatement) baseStatementUnit.getStatement()).executeUpdate();
}
});
return accumulate(results);
}finally{
MetricsContext.stop(context);
}
}

// ShardingPreparedStatement.java
privateCollection<PreparedStatementUnit> route()throwsSQLException{
Collection<PreparedStatementUnit> result =newLinkedList<>();
// 路由
   setRouteResult(routingEngine.route(getParameters()));
// 遍历 SQL 执行单元
for(SQLExecutionUnit each : getRouteResult().getExecutionUnits()){
SQLType sqlType = getRouteResult().getSqlStatement().getType();
Collection<PreparedStatement> preparedStatements;
// 创建实际的 PreparedStatement
if(SQLType.DDL == sqlType){
           preparedStatements = generatePreparedStatementForDDL(each);
}else{
           preparedStatements =Collections.singletonList(generatePreparedStatement(each));
}
       getRoutedStatements().addAll(preparedStatements);
// 回放设置占位符参数到 PreparedStatement
for(PreparedStatement preparedStatement : preparedStatements){
           replaySetParameter(preparedStatement);
           result.add(newPreparedStatementUnit(each, preparedStatement));
}
}
return result;
}
/**
* 创建 PreparedStatement
*
* @param sqlExecutionUnit SQL 执行单元
* @return PreparedStatement
* @throws SQLException 当 JDBC 操作发生异常时
*/
privatePreparedStatement generatePreparedStatement(finalSQLExecutionUnit sqlExecutionUnit)throwsSQLException{
Optional<GeneratedKey> generatedKey = getGeneratedKey();
// 获得连接
Connection connection = getShardingConnection().getConnection(sqlExecutionUnit.getDataSource(), getRouteResult().getSqlStatement().getType());
// 声明返回主键
if(isReturnGeneratedKeys()|| isReturnGeneratedKeys()&& generatedKey.isPresent()){
return connection.prepareStatement(sqlExecutionUnit.getSql(), RETURN_GENERATED_KEYS);
}
return connection.prepareStatement(sqlExecutionUnit.getSql(), getResultSetType(), getResultSetConcurrency(), getResultSetHoldability());
}

调用 #generatePreparedStatement() 创建 PreparedStatement，后调用 #replaySetParameter() 回放设置占位符参数到 PreparedStatement
当 声明返回主键 时，即 #isReturnGeneratedKeys() 返回 true 时，调用 connection.prepareStatement(sqlExecutionUnit.getSql(),RETURN_GENERATED_KEYS)。为什么该方法会返回 true？上文例子 conn.prepareStatement(sql,Statement.RETURN_GENERATED_KEYS)
声明返回主键后，插入执行完成，我们调用 #getGeneratedKeys() 可以获得主键：

// ShardingStatement.java
@Override
publicResultSet getGeneratedKeys()throwsSQLException{
Optional<GeneratedKey> generatedKey = getGeneratedKey();
// 分布式主键
if(generatedKey.isPresent()&& returnGeneratedKeys){
returnnewGeneratedKeysResultSet(routeResult.getGeneratedKeys().iterator(), generatedKey.get().getColumn(),this);
}
// 数据库自增
if(1== getRoutedStatements().size()){
return getRoutedStatements().iterator().next().getGeneratedKeys();
}
returnnewGeneratedKeysResultSet();
}
// ShardingConnection.java
@Override
publicPreparedStatement prepareStatement(finalString sql,finalString[] columnNames)throwsSQLException{
returnnewShardingPreparedStatement(this, sql,Statement.RETURN_GENERATED_KEYS);
}
// ShardingPreparedStatement.java
publicShardingPreparedStatement(finalShardingConnection shardingConnection,finalString sql,finalint autoGeneratedKeys){
this(shardingConnection, sql);
if(RETURN_GENERATED_KEYS == autoGeneratedKeys){
markReturnGeneratedKeys();
}
}
protectedfinalvoid markReturnGeneratedKeys(){
returnGeneratedKeys =true;
}

调用 ShardingConnection#getConnection() 方法获得该 PreparedStatement 对应的真实数据库连接( Connection )：

调用 #getCachedConnection() 尝试获得已缓存的数据库连接；如果缓存中不存在，获取到连接后会进行缓存
从 ShardingRule 配置的 DataSourceRule 获取真实的数据源( DataSource )
MasterSlaveDataSource 实现主从数据源封装，我们在下小节分享
调用 #replayMethodsInvocation() 回放记录的 Connection 方法

// ShardingConnection.java
/**
* 根据数据源名称获取相应的数据库连接.
*
* @param dataSourceName 数据源名称
* @param sqlType SQL语句类型
* @return 数据库连接
* @throws SQLException SQL异常
*/
publicConnection getConnection(finalString dataSourceName,finalSQLType sqlType)throwsSQLException{
// 从连接缓存中获取连接
Optional<Connection> connection = getCachedConnection(dataSourceName, sqlType);
if(connection.isPresent()){
return connection.get();
}
Context metricsContext =MetricsContext.start(Joiner.on("-").join("ShardingConnection-getConnection", dataSourceName));
//
DataSource dataSource = shardingContext.getShardingRule().getDataSourceRule().getDataSource(dataSourceName);
Preconditions.checkState(null!= dataSource,"Missing the rule of %s in DataSourceRule", dataSourceName);
String realDataSourceName;
if(dataSource instanceofMasterSlaveDataSource){
dataSource =((MasterSlaveDataSource) dataSource).getDataSource(sqlType);
realDataSourceName =MasterSlaveDataSource.getDataSourceName(dataSourceName, sqlType);
}else{
realDataSourceName = dataSourceName;
}
Connection result = dataSource.getConnection();
MetricsContext.stop(metricsContext);
// 添加到连接缓存
connectionMap.put(realDataSourceName, result);
// 回放 Connection 方法
replayMethodsInvocation(result);
return result;
}
privateOptional<Connection> getCachedConnection(finalString dataSourceName,finalSQLType sqlType){
String key = connectionMap.containsKey(dataSourceName)? dataSourceName :MasterSlaveDataSource.getDataSourceName(dataSourceName, sqlType);
returnOptional.fromNullable(connectionMap.get(key));
}

插入实现的代码基本分享完了，因为是不断代码下钻的方式分析，可以反向向上在理理，会更加清晰。

5. 查询流程

单纯从 core 包里的 JDBC 实现，查询流程 #executeQuery() 和 #execute() 基本一致，差别在于执行和多结果集归并。

@Override
publicResultSet executeQuery()throwsSQLException{
ResultSet result;
try{
// 路由
Collection<PreparedStatementUnit> preparedStatementUnits = route();
// 执行
List<ResultSet> resultSets =newPreparedStatementExecutor(
               getShardingConnection().getShardingContext().getExecutorEngine(), getRouteResult().getSqlStatement().getType(), preparedStatementUnits, getParameters()).executeQuery();
// 结果归并
       result =newShardingResultSet(resultSets,newMergeEngine(
               getShardingConnection().getShardingContext().getDatabaseType(), resultSets,(SelectStatement) getRouteResult().getSqlStatement()).merge());
}finally{
       clearBatch();
}
// 设置结果集
   setCurrentResultSet(result);
return result;
}

SQL执行 感兴趣的同学可以看：《Sharding-JDBC 源码分析 —— SQL 执行》
结果归并 感兴趣的同学可以看：《Sharding-JDBC 源码分析 —— 结果归并》
结果归并 #merge() 完后，创建分片结果集( ShardingResultSet )

publicfinalclassShardingResultSetextendsAbstractResultSetAdapter{
/**
* 归并结果集
*/
privatefinalResultSetMerger mergeResultSet;
@Override
publicint getInt(finalint columnIndex)throwsSQLException{
Object result = mergeResultSet.getValue(columnIndex,int.class);
wasNull =null== result;
return(int)ResultSetUtil.convertValue(result,int.class);
}
@Override
publicint getInt(finalString columnLabel)throwsSQLException{
Object result = mergeResultSet.getValue(columnLabel,int.class);
wasNull =null== result;
return(int)ResultSetUtil.convertValue(result,int.class);
}
// .... 隐藏其他类似 getXXXX() 方法
}

6. 读写分离

建议前置阅读：《官方文档 —— 读写分离》

当你有读写分离的需求时，将 ShardingRule 配置对应的数据源 从 ShardingDataSource 替换成 MasterSlaveDataSource。我们来看看 MasterSlaveDataSource 的功能和实现。

支持一主多从的读写分离配置，可配合分库分表使用

// MasterSlaveDataSourceFactory.java
publicfinalclassMasterSlaveDataSourceFactory{
/**
     * 创建读写分离数据源.
     * 
     * @param name 读写分离数据源名称
     * @param masterDataSource 主节点数据源
     * @param slaveDataSource 从节点数据源
     * @param otherSlaveDataSources 其他从节点数据源
     * @return 读写分离数据源
     */
publicstaticDataSource createDataSource(finalString name,finalDataSource masterDataSource,finalDataSource slaveDataSource,finalDataSource... otherSlaveDataSources){
returnnewMasterSlaveDataSource(name, masterDataSource,Lists.asList(slaveDataSource, otherSlaveDataSources));
}
}
// MasterSlaveDataSource.java
publicfinalclassMasterSlaveDataSourceextendsAbstractDataSourceAdapter{
/**
     * 数据源名
     */
privatefinalString name;
/**
     * 主数据源
     */
@Getter
privatefinalDataSource masterDataSource;
/**
     * 从数据源集合
     */
@Getter
privatefinalList<DataSource> slaveDataSources;
}

同一线程且同一数据库连接内，如有写入操作，以后的读操作均从主库读取，用于保证数据一致性。

// ShardingConnection.java
publicConnection getConnection(finalString dataSourceName,finalSQLType sqlType)throwsSQLException{
// .... 省略部分代码
String realDataSourceName;
if(dataSource instanceofMasterSlaveDataSource){// 读写分离
       dataSource =((MasterSlaveDataSource) dataSource).getDataSource(sqlType);
       realDataSourceName =MasterSlaveDataSource.getDataSourceName(dataSourceName, sqlType);
}else{
       realDataSourceName = dataSourceName;
}
Connection result = dataSource.getConnection();
// .... 省略部分代码
}
// MasterSlaveDataSource.java
/**
* 当前线程是否是 DML 操作标识
*/
privatestaticfinalThreadLocal<Boolean> DML_FLAG =newThreadLocal<Boolean>(){
@Override
protectedBoolean initialValue(){
returnfalse;
}
};
/**
* 从库负载均衡策略
*/
privatefinalSlaveLoadBalanceStrategy slaveLoadBalanceStrategy =newRoundRobinSlaveLoadBalanceStrategy();
/**
* 获取主或从节点的数据源.
*
* @param sqlType SQL类型
* @return 主或从节点的数据源
*/
publicDataSource getDataSource(finalSQLType sqlType){
if(isMasterRoute(sqlType)){
       DML_FLAG.set(true);
return masterDataSource;
}
return slaveLoadBalanceStrategy.getDataSource(name, slaveDataSources);
}
privatestaticboolean isMasterRoute(finalSQLType sqlType){
returnSQLType.DQL != sqlType || DML_FLAG.get()||HintManagerHolder.isMasterRouteOnly();
}

ShardingConnection 获取到的数据源是 MasterSlaveDataSource 时，调用 MasterSlaveDataSource#getConnection() 方法获取真实的数据源
通过 #isMasterRoute() 判断是否读取主库，以下三种情况会访问主库：

非查询语句 (DQL)
该数据源在当前线程访问过主库：通过线程变量 DML_FLAG 实现
强制主库：程序里调用 HintManager.getInstance().setMasterRouteOnly() 实现

访问从库时，会通过负载均衡策略( SlaveLoadBalanceStrategy ) 选择一个从库

MasterSlaveDataSource 默认使用 RoundRobinSlaveLoadBalanceStrategy，暂时不支持配置
RoundRobinSlaveLoadBalanceStrategy，轮询负载均衡策略，每个从节点访问次数均衡，暂不支持数据源故障移除

// SlaveLoadBalanceStrategy.java
publicinterfaceSlaveLoadBalanceStrategy{
/**
* 根据负载均衡策略获取从库数据源.
*
* @param name 读写分离数据源名称
* @param slaveDataSources 从库数据源列表
* @return 选中的从库数据源
*/
DataSource getDataSource(String name,List<DataSource> slaveDataSources);
}
// RoundRobinSlaveLoadBalanceStrategy.java
publicfinalclassRoundRobinSlaveLoadBalanceStrategyimplementsSlaveLoadBalanceStrategy{
privatestaticfinalConcurrentHashMap<String,AtomicInteger> COUNT_MAP =newConcurrentHashMap<>();
@Override
publicDataSource getDataSource(finalString name,finalList<DataSource> slaveDataSources){
AtomicInteger count = COUNT_MAP.containsKey(name)? COUNT_MAP.get(name):newAtomicInteger(0);
COUNT_MAP.putIfAbsent(name, count);
count.compareAndSet(slaveDataSources.size(),0);
return slaveDataSources.get(count.getAndIncrement()% slaveDataSources.size());
}
}

666. 彩蛋

没有彩蛋

没有彩

没有

没

下一篇，《分布式事务（一）之最大努力型》走起。老司机，赶紧上车。

道友，分享一个朋友圈可好？不然交个道姑那~~敏感词~~你。

继续阅读

阅读原文

数据库中间件 Sharding-JDBC 源码分析 —— JDBC实现与读写分离