如何写一个读写分离中间件

-javaagent:/{Path}/transmittable-thread-local-2.6.0-SNAPSHOT.jar

@Aspect@Componentpublic class DataSourceAspect{    @Around("execution(* *(..)) && @annotation(dataSourceType)")    public Object aroundMethod(ProceedingJoinPoint pjd, DataSourceType dataSourceType) throws Throwable {      DataSourceContextHolder.setMultiSqlDataSourceType(dataSourceType.name());        Object result = pjd.proceed();        DataSourceContextHolder.clearMultiSqlDataSourceType();        return result;    }}

public final class BackendConnection extends AbstractConnectionAdapter {
    private AbstractRoutingDataSource abstractRoutingDataSource;
    // 用于缓存一条sql（可能对应多个statement）或者一次事务中的连接    private final Map<String, Connection> connectionMap = new HashMap<String, Connection>();
    //构造函数    public BackendConnection(AbstractRoutingDataSource abstractRoutingDataSource) {        this.abstractRoutingDataSource = abstractRoutingDataSource;    }
    @Override    public PreparedStatement prepareStatement(String sql) throws SQLException {        return getConnectionInternal(sql).prepareStatement(sql);    }
    @Override    public DatabaseMetaData getMetaData() throws SQLException {        if(connectionMap == null || connectionMap.isEmpty()){            return abstractRoutingDataSource.getResolvedDefaultDataSource().getConnection().getMetaData();        }        return fetchCachedConnection(connectionMap.keySet().iterator().next().toString()).get().getMetaData();    }
    @Override    public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency)            throws SQLException {        return getConnectionInternal(sql).prepareStatement(sql,resultSetType,resultSetConcurrency);    }
    @Override    public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency,            int resultSetHoldability) throws SQLException {        return getConnectionInternal(sql).prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);    }
    @Override    public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {        return getConnectionInternal(sql).prepareStatement(sql, autoGeneratedKeys);    }
    @Override    public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {        return getConnectionInternal(sql).prepareStatement(sql, columnIndexes);    }
    @Override    public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {        return getConnectionInternal(sql).prepareStatement(sql, columnNames);    }
    @Override    protected Collection<Connection> getConnections() {        return connectionMap.values();    }
    /**     * 根据sql获取连接，对连接进行缓存     * @param sql     * @return     * @throws SQLException     */    private Connection getConnectionInternal(final String sql) throws SQLException {        //设置线程环境遍历        if (ExecutionEventUtil.isDML(sql)) {            DataSourceContextHolder.setSingleSqlDataSourceType(DataSourceType.MASTER);        } else if (ExecutionEventUtil.isDQL(sql)) {            DataSourceContextHolder.setSingleSqlDataSourceType(DataSourceType.SLAVE);        }        //根据上面设置的环境变量，选择相应的数据源        Object dataSourceKey = abstractRoutingDataSource.determineCurrentLookupKey();        String dataSourceName = dataSourceKey.toString();
        //看缓存中是否已经含有相应数据源的连接        Optional<Connection> connectionOptional = fetchCachedConnection(dataSourceName);        if (connectionOptional.isPresent()) {            return connectionOptional.get();        }        //缓存中没有相应连接，建立相应连接，并放入缓存        Connection connection = abstractRoutingDataSource.getTargetDataSource(dataSourceKey).getConnection();        connection.setAutoCommit(super.getAutoCommit());        connection.setTransactionIsolation(super.getTransactionIsolation());
        connectionMap.put(dataSourceKey.toString(), connection);
        return connection;    }
    /**     * 从缓存中取数据源     * @param dataSourceName     * @return     */    private Optional<Connection> fetchCachedConnection(final String dataSourceName) {        if (connectionMap.containsKey(dataSourceName)) {            return Optional.of(connectionMap.get(dataSourceName));        }        return Optional.absent();    }
}

/** *  * @Type AbstractRoutingDataSource * @Desc 数据源路由器（spring的AbstractRoutingDataSource将resolvedDataSources的注入放在bean初始化） * @Version V1.0 */public abstract class AbstractRoutingDataSource extends AbstractDataSource {
    private boolean lenientFallback = true;    private Map<Object, Object> targetDataSources;    private Object defaultTargetDataSource;    private Map<Object, DataSource> resolvedDataSources = new HashMap<Object, DataSource>();    private DataSource resolvedDefaultDataSource;    private Logger logger = LoggerFactory.getLogger(AbstractRoutingDataSource.class);
    public BackendConnection getConnection() throws SQLException {        return new BackendConnection(this);    }
    public BackendConnection getConnection(String username, String password)            throws SQLException {        return new BackendConnection(this);    }
    public void afterPropertiesSet() {        if (this.targetDataSources == null) {            throw new IllegalArgumentException("Property 'targetDataSources' is required");        }        this.resolvedDataSources = new HashMap<Object, DataSource>(this.targetDataSources.size());        for (Map.Entry entry : this.targetDataSources.entrySet()) {            Object lookupKey = resolveSpecifiedLookupKey(entry.getKey());            DataSource dataSource = resolveSpecifiedDataSource(entry.getValue());            this.resolvedDataSources.put(lookupKey, dataSource);        }        if (this.defaultTargetDataSource != null) {            this.resolvedDefaultDataSource = resolveSpecifiedDataSource(this.defaultTargetDataSource);        }    }
    public void putNewDataSource(Object key, DataSource dataSource){        if(this.resolvedDataSources == null){            this.resolvedDataSources = new HashMap<Object, DataSource>();        }        if(this.resolvedDataSources.containsKey(key)){            this.resolvedDataSources.remove(key);            logger.info("remove old key:" + key);        }        logger.info("add key:" + key + ", value=" + dataSource);        this.resolvedDataSources.put(key, dataSource);    }
    /**     * 数据源选择逻辑     */    public DataSource determineTargetDataSource() {        Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
        Object lookupKey = determineCurrentLookupKey();        DataSourceContextHolder.clearSingleSqlDataSourceType();
        int index = 0;        for (Entry<Object, DataSource> element : resolvedDataSources.entrySet()) {            logger.debug("myAbstractDS, index:" + index + ", key:" + element.getKey() + ", value:" + element.getValue().toString());            index++;        }        DataSource dataSource = this.resolvedDataSources.get(lookupKey);        if (dataSource == null && (this.lenientFallback || lookupKey == null)) {            dataSource = this.resolvedDefaultDataSource;        }        if (dataSource == null) {            throw new IllegalStateException("Cannot determine target DataSource for lookup key [" + lookupKey + "]");        }        logger.debug("myAbstractDS, hit DS is " + dataSource.toString());        return dataSource;    }
    public DataSource getTargetDataSource(Object lookupKey) {        Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
        if(lookupKey == null){            lookupKey = determineCurrentLookupKey();        }        DataSourceContextHolder.clearSingleSqlDataSourceType();
        int index = 0;        for (Entry<Object, DataSource> element : resolvedDataSources.entrySet()) {            logger.debug("myAbstractDS, index:" + index + ", key:" + element.getKey() + ", value:" + element.getValue().toString());            index++;        }        DataSource dataSource = this.resolvedDataSources.get(lookupKey);        if (dataSource == null && (this.lenientFallback || lookupKey == null)) {            dataSource = this.resolvedDefaultDataSource;        }        if (dataSource == null) {            throw new IllegalStateException("Cannot determine target DataSource for lookup key [" + lookupKey + "]");        }        logger.debug("myAbstractDS, hit DS is " + dataSource.toString());        return dataSource;    }
    public abstract Object determineCurrentLookupKey();    public abstract Object getCurrentSlaveKey();
    @Override    public boolean isWrapperFor(Class<?> iface) throws SQLException {        return (iface.isInstance(this) || determineTargetDataSource().isWrapperFor(iface));    }
    @SuppressWarnings("unchecked")    @Override    public <T> T unwrap(Class<T> iface) throws SQLException {        if (iface.isInstance(this)){            return (T) this;        }        return determineTargetDataSource().unwrap(iface);    }
    protected Object resolveSpecifiedLookupKey(Object lookupKey) {        return lookupKey;    }
    protected DataSource resolveSpecifiedDataSource(Object dataSource) throws IllegalArgumentException {        if (dataSource instanceof DataSource) {            return (DataSource) dataSource;        }        else {            throw new IllegalArgumentException(                    "Illegal data source value - only [javax.sql.DataSource] and String supported: " + dataSource);        }    }    //get set方法省略}

/** *  * @Type AbstractRoutingDataSource * @Desc 数据源路由器（spring的AbstractRoutingDataSource将resolvedDataSources的注入放在bean初始化） * @Version V1.0 */public abstract class AbstractRoutingDataSource extends AbstractDataSource {
    private boolean lenientFallback = true;    private Map<Object, Object> targetDataSources;    private Object defaultTargetDataSource;    private Map<Object, DataSource> resolvedDataSources = new HashMap<Object, DataSource>();    private DataSource resolvedDefaultDataSource;    private Logger logger = LoggerFactory.getLogger(AbstractRoutingDataSource.class);
    public BackendConnection getConnection() throws SQLException {        return new BackendConnection(this);    }
    public BackendConnection getConnection(String username, String password)            throws SQLException {        return new BackendConnection(this);    }
    public void afterPropertiesSet() {        if (this.targetDataSources == null) {            throw new IllegalArgumentException("Property 'targetDataSources' is required");        }        this.resolvedDataSources = new HashMap<Object, DataSource>(this.targetDataSources.size());        for (Map.Entry entry : this.targetDataSources.entrySet()) {            Object lookupKey = resolveSpecifiedLookupKey(entry.getKey());            DataSource dataSource = resolveSpecifiedDataSource(entry.getValue());            this.resolvedDataSources.put(lookupKey, dataSource);        }        if (this.defaultTargetDataSource != null) {            this.resolvedDefaultDataSource = resolveSpecifiedDataSource(this.defaultTargetDataSource);        }    }
    public void putNewDataSource(Object key, DataSource dataSource){        if(this.resolvedDataSources == null){            this.resolvedDataSources = new HashMap<Object, DataSource>();        }        if(this.resolvedDataSources.containsKey(key)){            this.resolvedDataSources.remove(key);            logger.info("remove old key:" + key);        }        logger.info("add key:" + key + ", value=" + dataSource);        this.resolvedDataSources.put(key, dataSource);    }
    /**     * 数据源选择逻辑     */    public DataSource determineTargetDataSource() {        Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
        Object lookupKey = determineCurrentLookupKey();        DataSourceContextHolder.clearSingleSqlDataSourceType();
        int index = 0;        for (Entry<Object, DataSource> element : resolvedDataSources.entrySet()) {            logger.debug("myAbstractDS, index:" + index + ", key:" + element.getKey() + ", value:" + element.getValue().toString());            index++;        }        DataSource dataSource = this.resolvedDataSources.get(lookupKey);        if (dataSource == null && (this.lenientFallback || lookupKey == null)) {            dataSource = this.resolvedDefaultDataSource;        }        if (dataSource == null) {            throw new IllegalStateException("Cannot determine target DataSource for lookup key [" + lookupKey + "]");        }        logger.debug("myAbstractDS, hit DS is " + dataSource.toString());        return dataSource;    }
    public DataSource getTargetDataSource(Object lookupKey) {        Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
        if(lookupKey == null){            lookupKey = determineCurrentLookupKey();        }        DataSourceContextHolder.clearSingleSqlDataSourceType();
        int index = 0;        for (Entry<Object, DataSource> element : resolvedDataSources.entrySet()) {            logger.debug("myAbstractDS, index:" + index + ", key:" + element.getKey() + ", value:" + element.getValue().toString());            index++;        }        DataSource dataSource = this.resolvedDataSources.get(lookupKey);        if (dataSource == null && (this.lenientFallback || lookupKey == null)) {            dataSource = this.resolvedDefaultDataSource;        }        if (dataSource == null) {            throw new IllegalStateException("Cannot determine target DataSource for lookup key [" + lookupKey + "]");        }        logger.debug("myAbstractDS, hit DS is " + dataSource.toString());        return dataSource;    }
    public abstract Object determineCurrentLookupKey();    public abstract Object getCurrentSlaveKey();
    @Override    public boolean isWrapperFor(Class<?> iface) throws SQLException {        return (iface.isInstance(this) || determineTargetDataSource().isWrapperFor(iface));    }
    @SuppressWarnings("unchecked")    @Override    public <T> T unwrap(Class<T> iface) throws SQLException {        if (iface.isInstance(this)){            return (T) this;        }        return determineTargetDataSource().unwrap(iface);    }
    protected Object resolveSpecifiedLookupKey(Object lookupKey) {        return lookupKey;    }
    protected DataSource resolveSpecifiedDataSource(Object dataSource) throws IllegalArgumentException {        if (dataSource instanceof DataSource) {            return (DataSource) dataSource;        }        else {            throw new IllegalArgumentException(                    "Illegal data source value - only [javax.sql.DataSource] and String supported: " + dataSource);        }    }    //get set方法省略}

public class DataSourceContextHolder {
    private static final TransmittableThreadLocal<String> singleSqlContextHolder = new TransmittableThreadLocal<String>();    private static final TransmittableThreadLocal<String> multiSqlContextHolder = new TransmittableThreadLocal<String>();
    /**     * @Description: 设置单条sql数据源类型     * @param dataSourceType  数据库类型     * @return void     * @throws     */    public static void setSingleSqlDataSourceType(String dataSourceType) {        singleSqlContextHolder.set(dataSourceType);    }
    /**     * @Description: 获取单条sql数据源类型     * @param      * @return String     * @throws     */    public static String getSingleSqlDataSourceType() {        return singleSqlContextHolder.get();    }
    /**     * @Description: 清除单条sql数据源类型     * @param      * @return void     * @throws     */    public static void clearSingleSqlDataSourceType() {        singleSqlContextHolder.remove();    }
    /**     * @Description: 设置多条sql数据源类型     * @param dataSourceType  数据库类型     * @return void     * @throws     */    public static void setMultiSqlDataSourceType(String dataSourceType) {        multiSqlContextHolder.set(dataSourceType);    }
    /**     * @Description: 获取多条sql数据源类型     * @param      * @return String     * @throws     */    public static String getMultiSqlDataSourceType() {        return multiSqlContextHolder.get();    }
    /**     * @Description: 清除多条sql数据源类型     * @param      * @return void     * @throws     */    public static void clearMultiSqlDataSourceType() {        multiSqlContextHolder.remove();    }
    /**     * 判断当前线程是否为使用从库为数据源. 最外层service有slave的aop标签  或者 service没有aop标签且单条sql为DQL     *      * @return     */    public static boolean isSlave() {        return "slave".equals(multiSqlContextHolder.get()) || (multiSqlContextHolder.get()==null && "slave".equals(singleSqlContextHolder.get())) ;    }  }

public class DynamicDataSource extends AbstractRoutingDataSource implements InitializingBean{  
    private static final Logger logger = LoggerFactory.getLogger(DynamicDataSource.class);     private Integer slaveCount = 0;  
    // 轮询计数,初始为-1,AtomicInteger是线程安全的      private AtomicInteger counter = new AtomicInteger(-1); 
    // 记录读库的key      private List<Object> slaveDataSources = new ArrayList<Object>(0); 
    // slave库的权重    private  Map<Object,Integer>  slaveDataSourcesWeight;
    private Object currentSlaveKey;        public DynamicDataSource() {        super();    }
    /**     * 构造函数     * @param defaultTargetDataSource     * @param targetDataSources     * @param slaveDataSourcesWeight     */    public DynamicDataSource(Object defaultTargetDataSource, Map<Object,Object> targetDataSources, Map<Object,Integer> slaveDataSourcesWeight) {        this.setResolvedDataSources(new HashMap<Object, DataSource>(targetDataSources.size()));        for (Map.Entry<Object, Object> entry : targetDataSources.entrySet()) {            DataSource dataSource = resolveSpecifiedDataSource(entry.getValue());            this.putNewDataSource(entry.getKey(), dataSource);        }        if (defaultTargetDataSource != null) {            this.setResolvedDefaultDataSource(resolveSpecifiedDataSource(defaultTargetDataSource));        }        this.setSlaveDataSourcesWeight(slaveDataSourcesWeight);        this.afterPropertiesSet();    }
    @Override      public Object determineCurrentLookupKey() {          // 使用DataSourceContextHolder保证线程安全，并且得到当前线程中的数据源key          if (DataSourceContextHolder.isSlave()) {              currentSlaveKey = getSlaveKey();              return currentSlaveKey;          }          //TODO        Object key = "master";          return key;      }  
    @Override      public void afterPropertiesSet() {          try {              super.afterPropertiesSet();            Map<Object, DataSource> resolvedDataSources = this.getResolvedDataSources();              //清空从库节点，重新生成            slaveDataSources.clear();            slaveCount = 0;            for (Map.Entry<Object, DataSource> entry : resolvedDataSources.entrySet()) {                 if(slaveDataSourcesWeight.get(entry.getKey())==null){                    continue;                }                for(int i=0; i<slaveDataSourcesWeight.get(entry.getKey());i++){                    slaveDataSources.add(entry.getKey());                      slaveCount++;                }             }          } catch (Exception e) {              logger.error("afterPropertiesSet error! ", e);          }      }  
    /**      * 轮询算法实现      *       * @return      */      public Object getSlaveKey() {          if(slaveCount <= 0 || slaveDataSources == null || slaveDataSources.size() <= 0){            return null;        }        Integer index = counter.incrementAndGet() % slaveCount;          if (counter.get() > 9999) { // 以免超出Integer范围              counter.set(-1); // 还原          }          return slaveDataSources.get(index);      }
    public Map<Object, Integer> getSlaveDataSourcesWeight() {        return slaveDataSourcesWeight;    }
    public void setSlaveDataSourcesWeight(Map<Object, Integer> slaveDataSourcesWeight) {        this.slaveDataSourcesWeight = slaveDataSourcesWeight;    }
    public Object getCurrentSlaveKey() {        return currentSlaveKey;    }}

/** * 更新beanFactory * @param properties */public void refreshDataSource(String properties) {    YamlDynamicDataSource dataSource;    try {        dataSource = new YamlDynamicDataSource(properties);    } catch (IOException e) {        throw new RuntimeException("convert datasource config failed!");    }
    // 验证必须字段是否存在    if (dataSource == null && dataSource.getResolvedDataSources() == null            || dataSource.getResolvedDefaultDataSource() == null || dataSource.getSlaveDataSourcesWeight() == null) {        throw new RuntimeException("datasource config error!");    }    ConcurrentHashMap<Object, DataSource> newDataSource = new ConcurrentHashMap<Object, DataSource>(            dataSource.getResolvedDataSources());
    //更新数据源的bean    DynamicDataSource dynamicDataSource = (DynamicDataSource) ((DefaultListableBeanFactory) beanFactory)            .getBean(dataSourceName);    dynamicDataSource.setResolvedDefaultDataSource(dataSource.getResolvedDefaultDataSource());    dynamicDataSource.setResolvedDataSources(new HashMap<Object, DataSource>());//将数据源清空，重新添加    for (Entry<Object, DataSource> element : newDataSource.entrySet()) {        dynamicDataSource.putNewDataSource(element.getKey(), element.getValue());    }    dynamicDataSource.setSlaveDataSourcesWeight(dataSource.getSlaveDataSourcesWeight());    dynamicDataSource.afterPropertiesSet();
}