shardingsphere源码解析

观看该文建议浏览：https://blog.csdn.net/qq_36851469/article/details/125449241

1、SpringBootConfiguration

@Configuration
@ComponentScan("org.apache.shardingsphere.spring.boot.converter")
// EnableConfigurationProperties都是用于加载配置信息
@EnableConfigurationProperties({
    
        SpringBootShardingRuleConfigurationProperties.class,
        SpringBootMasterSlaveRuleConfigurationProperties.class, SpringBootEncryptRuleConfigurationProperties.class,
        SpringBootPropertiesConfigurationProperties.class, SpringBootShadowRuleConfigurationProperties.class})
@ConditionalOnProperty(prefix = "spring.shardingsphere", name = "enabled", havingValue = "true", matchIfMissing = true)
// shardingsphere初始化 DataSource 优先于DataSourceAutoConfiguration，同时MybatisAutoConfiguration初始化落后于 DataSourceAutoConfiguration
@AutoConfigureBefore(DataSourceAutoConfiguration.class)
@RequiredArgsConstructor
public class SpringBootConfiguration implements EnvironmentAware {
    
	
	@Bean
	@Conditional(ShardingRuleCondition.class)
	public DataSource shardingDataSource() throws SQLException {
    
	    return ShardingDataSourceFactory.createDataSource(dataSourceMap, new 	
	    ShardingRuleConfigurationYamlSwapper().swap(shardingRule), props.getProps());
	}
}

public final class ShardingDataSourceFactory {
    
  public static DataSource createDataSource(
          final Map<String, DataSource> dataSourceMap, final ShardingRuleConfiguration shardingRuleConfig, final 
          Properties props) throws SQLException {
    
      return new ShardingDataSource(dataSourceMap, new ShardingRule(shardingRuleConfig, dataSourceMap.keySet()), 
      		props);
  }
}
// ShardingRule 分库分表的分片策略信息
public ShardingDataSource(final Map<String, DataSource> dataSourceMap, final ShardingRule shardingRule, final 
		Properties props) throws SQLException {
    
		// 其中dataSourceMap最终由 抽象类 AbstractDataSourceAdapter持有
        super(dataSourceMap);
        checkDataSourceType(dataSourceMap);
        runtimeContext = new ShardingRuntimeContext(dataSourceMap, shardingRule, props, getDatabaseType());
    }

2、EnvironmentAware

目的加载配置文件信息。

@Override
public final void setEnvironment(final Environment environment) {
    
    String prefix = "spring.shardingsphere.datasource.";
    // 获取 prefix + name 配置「spring.shardingsphere.datasource.name」的所有库名
    for (String each : getDataSourceNames(environment, prefix)) {
    
        // dataSourceMap 维护配置文件中配置的所有 DataSource
        dataSourceMap.put(each, getDataSource(environment, prefix, each));
    }
}

private DataSource getDataSource(final Environment environment, final String prefix, final String dataSourceName) 	
	throws ReflectiveOperationException, NamingException {
    
    Map<String, Object> dataSourceProps = PropertyUtil.handle(environment, prefix + dataSourceName.trim(), Map.class);
    // 根据配置的 type属性初始化对应类型的 DataSource 「HikariDataSource」
    DataSource result = DataSourceUtil.getDataSource(dataSourceProps.get("type").toString(), dataSourceProps);
 	...
    return result;
}

public static DataSource getDataSource(final String dataSourceClassName, final Map<String, Object> 
	dataSourceProperties) throws ReflectiveOperationException {
    
	    // 根据配置的driver-class-name实例化 DataSource
        DataSource result = (DataSource) Class.forName(dataSourceClassName).newInstance();
        // 设置DataSource的属性包括：用户名、密码、jdbcUrl等
        for (Entry<String, Object> entry : dataSourceProperties.entrySet()) {
    
            callSetterMethod(result, getSetterMethodName(entry.getKey()), null == entry.getValue() ? null : 	
            entry.getValue().toString());
        }
        return result;
    }

3、ShardingConnection

private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {
    
  Statement stmt;
  //ShardingConnection：通过之前Mybatis源码分析可知，最终通过 SpringManagedTransaction 持有的 ShardingDataSource获取连接ShardingConnection
  Connection connection = getConnection(statementLog);
  stmt = handler.prepare(connection, transaction.getTimeout());
  handler.parameterize(stmt);
  return stmt;
}

public class ShardingDataSource extends AbstractDataSourceAdapter {
    
	@Override
    public final ShardingConnection getConnection() {
    
        // getDataSourceMap() 获取 抽象类AbstractDataSourceAdapter持有的DataSourceMap
        return new ShardingConnection(getDataSourceMap(), runtimeContext, TransactionTypeHolder.get());
    }
}

public ShardingConnection(final Map<String, DataSource> dataSourceMap, final ShardingRuntimeContext runtimeContext, 	
			final TransactionType transactionType) {
    
	// 持有当前应用所有的dataSourceMap
    this.dataSourceMap = dataSourceMap;
    this.runtimeContext = runtimeContext;
    this.transactionType = transactionType;
    shardingTransactionManager = 	
    runtimeContext.getShardingTransactionManagerEngine().getTransactionManager(transactionType);
}

4、ShardingPreparedStatement

//返回的为 ShardingPreparedStatement
public class PreparedStatementHandler extends BaseStatementHandler {
    
	protected Statement instantiateStatement(Connection connection) throws SQLException {
    
    String sql = boundSql.getSql();
    if (mappedStatement.getKeyGenerator() instanceof Jdbc3KeyGenerator) {
    
      String[] keyColumnNames = mappedStatement.getKeyColumns();
      if (keyColumnNames == null) {
    
        return connection.prepareStatement(sql, PreparedStatement.RETURN_GENERATED_KEYS);
      } else {
    
        return connection.prepareStatement(sql, keyColumnNames);
      }
    } else if (mappedStatement.getResultSetType() == ResultSetType.DEFAULT) {
    
      return connection.prepareStatement(sql);
    } else {
    
      return connection.prepareStatement(sql, mappedStatement.getResultSetType().getValue(), 
      	ResultSet.CONCUR_READ_ONLY);
    }
  }
}

@Override
public <E> List<E> query(Statement statement, ResultHandler resultHandler) throws SQLException {
    
  // ShardingPreparedStatement
  PreparedStatement ps = (PreparedStatement) statement;
  ps.execute();
  return resultSetHandler.handleResultSets(ps);
}

public boolean execute() throws SQLException {
    
    // 此处开始执行 分库分表插件 的逻辑
    prepare();
    // #5 初始化 StatementExecuteUnit，就是 PreparedStatement 最终执行SQL的最小单位。其实就是初始化AbstractStatementExecutor属性#inputGroups
    initPreparedStatementExecutor();
    // 通过回调真正执行SQL
    return preparedStatementExecutor.execute();
}

public final class SQLExecuteCallbackFactory {
    
	public static SQLExecuteCallback<Boolean> getPreparedSQLExecuteCallback(final DatabaseType databaseType, final 
				boolean isExceptionThrown) {
    
	    return new SQLExecuteCallback<Boolean>(databaseType, isExceptionThrown) {
    
	        
	        @Override
	        protected Boolean executeSQL(final String sql, final Statement statement, final ConnectionMode 
	        			connectionMode) throws SQLException {
    
	        	// 真正执行原生jdbc
	            return ((PreparedStatement) statement).execute();
	        }
	    };
	}
}

5、PreparedStatementExecutor

AbstractStatementExecutor#inputGroups包含SQL执行需要的SQL、sql参数 & PrepareStatement等。PrepareStatement存在数据库连接、jdbcUrl等SQL执行需要必须条件。

public final class PreparedStatementExecutor extends AbstractStatementExecutor {
    
	public void init(final ExecutionContext executionContext) throws SQLException {
    
        setSqlStatementContext(executionContext.getSqlStatementContext());
        //获取 AbstractStatementExecutor类属性 inputGroups
        getInputGroups().addAll(obtainExecuteGroups(executionContext.getExecutionUnits()));
        cacheStatements();
    }

	private Collection<InputGroup<StatementExecuteUnit>> obtainExecuteGroups(final Collection<ExecutionUnit> 
		executionUnits) throws SQLException {
    
        return getSqlExecutePrepareTemplate().getExecuteUnitGroups(executionUnits, new SQLExecutePrepareCallback() {
    
            @Override
            public List<Connection> getConnections(final ConnectionMode connectionMode, final String dataSourceName, 
            	final int connectionSize) throws SQLException {
    
            	//dataSourceName：是用户实现PreciseShardingAlgorithm接口返回的目标库名
            	// 通过 dataSourceName 从 AbstractDataSourceAdapter中dataSourceMap中获取对应的dataSource
            	// 通过 dataSource 获取对应的数据库连接
                return PreparedStatementExecutor.super.getConnection().getConnections(connectionMode, dataSourceName, 	
                	connectionSize);
            }
            
            @Override
            public StatementExecuteUnit createStatementExecuteUnit(final Connection connection, final ExecutionUnit 
            	executionUnit, final ConnectionMode connectionMode) throws SQLException {
    
            	// 创建 StatementExecuteUnit：createPreparedStatement通过
                return new StatementExecuteUnit(executionUnit, createPreparedStatement(connection, 
                	executionUnit.getSqlUnit().getSql()), connectionMode);
            }
        });
    }
	
	private PreparedStatement createPreparedStatement(final Connection connection, final String sql) throws 
		SQLException {
    
		// 通过 HikariProxyConnection 获取 prepareStatement
        return returnGeneratedKeys ? connection.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS)
                : connection.prepareStatement(sql, getResultSetType(), getResultSetConcurrency(), 
                	getResultSetHoldability());
    }
}

StatementExecuteUnit

public final class StatementExecuteUnit {
    
   private final ExecutionUnit executionUnit;
   private final Statement statement;
   private final ConnectionMode connectionMode;
}

public final class ExecutionUnit {
    
    private final String dataSourceName;
    private final SQLUnit sqlUnit;
}

public final class SQLUnit {
    
    private final String sql;
    private final List<Object> parameters;
}

原生的jdbc

public static void main(String[] args) throws Exception {
    
    //1、注册驱动
    Class.forName("com.mysql.jdbc.Driver");
    //2、获取连接对象
    Connection con = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/day01_db",
            "root", "1234");
    //3、获取发送SQL语句的对象
    PreparedStatement ps = con.prepareStatement("select * from category");
    //4、发送SQL语句，返回结果集
    ResultSet rs = ps.executeQuery();
    //5、遍历结果集
    while (rs.next()){
    
        //遍历该行数据
        int cid = rs.getInt("cid");
        String cname = rs.getString("cname");
        System.out.println("cid:"+cid+"\t cname:"+cname);
    }
    //6、关闭资源
    rs.close();
    ps.close();
    con.close();
}