shardingsphere-readwrite-splitting

读写分离被单独拆分了出来，加载初始化配置逻辑都一样，读写分离api有一个扩展点，可以实现自己的ReplicaLoadBalanceAlgorithm逻辑，目前已有的只有轮询和随机

Datasouce选择中，如果是包含锁或不是读操作或hint或事务中或同一线程中已经有其他操作访问主库那么这些情况都路由到主库，其他情况按照定义的选择器做选择

@RequiredArgsConstructor
public final class ReadwriteSplittingDataSourceRouter {
    
    private final ReadwriteSplittingDataSourceRule rule;
    
    public String route(final SQLStatement sqlStatement) {
        if (isPrimaryRoute(sqlStatement)) {
            PrimaryVisitedManager.setPrimaryVisited();
            String autoAwareDataSourceName = rule.getAutoAwareDataSourceName();
            if (Strings.isNullOrEmpty(autoAwareDataSourceName)) {
                return rule.getWriteDataSourceName();
            }
            Optional<DataSourceNameAware> dataSourceNameAware = DataSourceNameAwareFactory.getInstance().getDataSourceNameAware();
            if (dataSourceNameAware.isPresent()) {
                return dataSourceNameAware.get().getPrimaryDataSourceName(autoAwareDataSourceName);
            }
        }
        String autoAwareDataSourceName = rule.getAutoAwareDataSourceName();
        if (Strings.isNullOrEmpty(autoAwareDataSourceName)) {
            return rule.getLoadBalancer().getDataSource(rule.getName(), rule.getWriteDataSourceName(), rule.getReadDataSourceNames());
        }
        Optional<DataSourceNameAware> dataSourceNameAware = DataSourceNameAwareFactory.getInstance().getDataSourceNameAware();
        if (dataSourceNameAware.isPresent()) {
            Collection<String> replicaDataSourceNames = dataSourceNameAware.get().getReplicaDataSourceNames(autoAwareDataSourceName);
            return rule.getLoadBalancer().getDataSource(rule.getName(), rule.getWriteDataSourceName(), new ArrayList<>(replicaDataSourceNames));
        }
        return rule.getLoadBalancer().getDataSource(rule.getName(), rule.getWriteDataSourceName(), rule.getReadDataSourceNames());
    }
    
    private boolean isPrimaryRoute(final SQLStatement sqlStatement) {
        return containsLockSegment(sqlStatement) || !(sqlStatement instanceof SelectStatement)
                || PrimaryVisitedManager.getPrimaryVisited() || HintManager.isWriteRouteOnly() || TransactionHolder.isTransaction();
    }
    
    private boolean containsLockSegment(final SQLStatement sqlStatement) {
        return sqlStatement instanceof SelectStatement && SelectStatementHandler.getLockSegment((SelectStatement) sqlStatement).isPresent();
    }
}

 

影子库功能

shadow功能，首先看下加载了那些东西，除了配置相关的之外，实现了两个重要的类，ShadowSQLRewriteContextDecorator和ShadowSQLRouter

ShadowRule构建，shadowMappings中包含了key-真实db；value-影子db

public ShadowRule(final ShadowRuleConfiguration shadowRuleConfig) {
        column = shadowRuleConfig.getColumn();
        shadowMappings = new HashMap<>(shadowRuleConfig.getShadowDataSourceNames().size());
        for (int i = 0; i < shadowRuleConfig.getSourceDataSourceNames().size(); i++) {
            shadowMappings.put(shadowRuleConfig.getSourceDataSourceNames().get(i), shadowRuleConfig.getShadowDataSourceNames().get(i));
        }
    }

ShadowSQLRouter 结合shadowMappings内容可知，下面一段代码路由选择，如果是DML需要走双库，否则按照是否影子库路由到对应的逻辑

    @Override
    public RouteContext createRouteContext(final LogicSQL logicSQL, final ShardingSphereMetaData metaData, final ShadowRule rule, final ConfigurationProperties props) {
        RouteContext result = new RouteContext();
        if (!(logicSQL.getSqlStatementContext().getSqlStatement() instanceof DMLStatement)) {
            rule.getShadowMappings().forEach((key, value) -> {
                result.getRouteUnits().add(new RouteUnit(new RouteMapper(key, key), Collections.emptyList()));
                result.getRouteUnits().add(new RouteUnit(new RouteMapper(value, value), Collections.emptyList()));
            });
            return result;
        }
        if (isShadow(logicSQL.getSqlStatementContext(), logicSQL.getParameters(), rule)) {
            rule.getShadowMappings().values().forEach(each -> result.getRouteUnits().add(new RouteUnit(new RouteMapper(each, each), Collections.emptyList())));
        } else {
            rule.getShadowMappings().keySet().forEach(each -> result.getRouteUnits().add(new RouteUnit(new RouteMapper(each, each), Collections.emptyList())));
        }
        return result;
    }

对传入的 SQL 进行路由并改写，删除影子字段与字段值，该逻辑位于路由链路的最后一个环节

@Getter
@Setter
public abstract class ShadowParameterRewriter<T extends SQLStatementContext> implements ParameterRewriter<T>, ShadowRuleAware {
    
    private ShadowRule shadowRule;
    
    @Override
    public final boolean isNeedRewrite(final SQLStatementContext sqlStatementContext) {
        return isNeedRewriteForShadow(sqlStatementContext);
    }
    
    protected abstract boolean isNeedRewriteForShadow(SQLStatementContext sqlStatementContext);
}

影子参数删除

public final class ShadowPredicateParameterRewriter extends ShadowParameterRewriter<SQLStatementContext> {
    
    @Override
    protected boolean isNeedRewriteForShadow(final SQLStatementContext sqlStatementContext) {
        return true;
    }
    
    @Override
    public void rewrite(final ParameterBuilder parameterBuilder, final SQLStatementContext sqlStatementContext, final List<Object> parameters) {
        new ShadowConditionEngine(getShadowRule()).createShadowCondition(sqlStatementContext).ifPresent(
            shadowCondition -> replaceShadowParameter(parameterBuilder, shadowCondition.getPositionIndexMap()));
    }
    
    private void replaceShadowParameter(final ParameterBuilder parameterBuilder, final Map<Integer, Integer> positionIndexes) {
        if (!positionIndexes.isEmpty()) {
            for (Entry<Integer, Integer> entry : positionIndexes.entrySet()) {
                ((StandardParameterBuilder) parameterBuilder).addRemovedParameters(entry.getValue());
            }
        }
    }
}