Sharding-Jdbc之读写分离导读

前言

      Sharding-JDBC是一个开源的分布式数据库中间件，它无需额外部署和依赖，完全兼容JDBC和各种ORM框架。Sharding-JDBC作为面向开发的微服务云原生基础类库，完整的实现了分库分表、读写分离和分布式主键功能，并初步实现了柔性事务。

以2.0.3版本为例maven包依赖如下

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>

<groupId>com.dongpeng</groupId>
<artifactId>sharding-jdbc</artifactId>
<version>1.0.0</version>
<packaging>jar</packaging>

<name>sharding-jdbc</name>
<url>http://maven.apache.org</url>

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>

<dependencies>
    <dependency>
        <groupId>io.shardingjdbc</groupId>
        <artifactId>sharding-jdbc-core</artifactId>
        <version>2.0.3</version>
    </dependency> 
    <dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>23.0</version>
    </dependency>
    <dependency>
        <groupId>com.mchange</groupId>
        <artifactId>c3p0</artifactId> 
        <version>0.9.5.2</version>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.46</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.21</version>
    </dependency> 
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.21</version>
    </dependency>

</dependencies>
<build>
    <resources>
        <resource>
            <directory>src/main/java</directory>
        </resource>
        <resource>
            <directory>src/main/resources</directory>
        </resource>
    </resources>
    <plugins>

        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
                <encoding>UTF-8</encoding>
            </configuration>
        </plugin>
    </plugins>
</build>

demo实现如下

package com.dongpeng.sharding.jdbc;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;

import javax.sql.DataSource;

import com.mchange.v2.c3p0.ComboPooledDataSource;

import io.shardingjdbc.core.api.MasterSlaveDataSourceFactory;
import io.shardingjdbc.core.api.config.MasterSlaveRuleConfiguration;
/**
 * sharding-jdbc读写分离
 * @author Admin
 *
 */
public class MasterSalveDemo {
    public static void main(String[] args) throws Exception {
        Map<String, DataSource> dataSourceMap = new HashMap<String, DataSource>();
        ComboPooledDataSource dataSource1 = new ComboPooledDataSource();
        dataSource1.setDriverClass("com.mysql.jdbc.Driver"); // loads the jdbc driver
        dataSource1.setJdbcUrl("jdbc:mysql://localhost:3306/db_0?useSSL=false");
        dataSource1.setUser("root");
        dataSource1.setPassword("root");
        dataSourceMap.put("ds_0", dataSource1);

        ComboPooledDataSource dataSource2 = new ComboPooledDataSource();
        dataSource2.setDriverClass("com.mysql.jdbc.Driver"); // loads the jdbc driver
        dataSource2.setJdbcUrl("jdbc:mysql://localhost:3306/db_1?useSSL=false");
        dataSource2.setUser("root");
        dataSource2.setPassword("root");
        dataSourceMap.put("ds_1", dataSource2);

        MasterSlaveRuleConfiguration masterSlaveRuleConfig = new MasterSlaveRuleConfiguration();
        masterSlaveRuleConfig.setName("ms_ds");
        masterSlaveRuleConfig.setMasterDataSourceName("ds_0");
        masterSlaveRuleConfig.getSlaveDataSourceNames().add("ds_1");

        DataSource dataSource = MasterSlaveDataSourceFactory.createDataSource(dataSourceMap, masterSlaveRuleConfig,
                new HashMap<String, Object>());
        Connection connection = dataSource.getConnection();
        Statement statement = connection.createStatement();
        ResultSet rs = statement.executeQuery("select * from t_order_0 where user_id=1");
        while (rs.next()) {
            System.out.println(rs.getString("order_id"));
        }

        statement.close();
        connection.close();
    }
}

源码解析

   主要的几个类

MasterSlaveRuleConfiguration

    MasterSlaveDataSourceFactory

    MasterSlaveConnection

    MasterSlaveDataSource

    MasterSlaveStatement

    MasterSlavePreparedStatement

   各个类的说明如下

    MasterSalveRuleConfiguration

  是一个配置类用于配置主从的一些参数，主要的配置参数如下

private String name;

private String masterDataSourceName;

private Collection<String> slaveDataSourceNames = new LinkedList<>();

private MasterSlaveLoadBalanceAlgorithmType loadBalanceAlgorithmType;

private String loadBalanceAlgorithmClassName;

name在独立的读写分离模式中，目前在策略接口中有使用

masterDataSourceName配置主节点dataSource

slaveDataSourceNames 提供从节点dataSource配置

loadBalanceAlgorithmType配置默认的从节点选取策略，默认支持两种一是轮询()，二是随机(由类            
RandomMasterSlaveLoadBalanceAlgorithm实现)

loadBanceAlgorithmClassName 配置自定义实现的从节点选取策略，可根据自己的需求自定义实现，
loadBalanceAlgorithmType配置优于自定义配置要这个类启作用，不要配置loadBalanceAlgorithmType

MasterSlaveDataSourceFactory

这个是个DataSource的工厂类实现，用于提供各种终端的配置载入方式,支持文件，自配置等等

public static DataSource createDataSource(final Map<String, DataSource> dataSourceMap, final MasterSlaveRuleConfiguration masterSlaveRuleConfig, 
                                              final Map<String, Object> configMap) throws SQLException {
        return new MasterSlaveDataSource(masterSlaveRuleConfig.build(dataSourceMap), configMap);
    }

  public static DataSource createDataSource(final File yamlFile) throws SQLException, IOException {
        YamlMasterSlaveConfiguration config = unmarshal(yamlFile);
        return new MasterSlaveDataSource(config.getMasterSlaveRule(Collections.<String, DataSource>emptyMap()), config.getMasterSlaveRule().getConfigMap());
    }

MasterSlaveDataSource

主要用于构建dataSource的信息，同时提供一些dataSource的选取方法，主要的先取方法如下

public NamedDataSource getDataSource(final SQLType sqlType) {
       if (isMasterRoute(sqlType)) {
           DML_FLAG.set(true);
           return new NamedDataSource(masterSlaveRule.getMasterDataSourceName(), masterSlaveRule.getMasterDataSource());
       }
       String selectedSourceName = masterSlaveRule.getStrategy().getDataSource(masterSlaveRule.getName(), 
               masterSlaveRule.getMasterDataSourceName(), new ArrayList<>(masterSlaveRule.getSlaveDataSourceMap().keySet()));
       DataSource selectedSource = selectedSourceName.equals(masterSlaveRule.getMasterDataSourceName())
               ? masterSlaveRule.getMasterDataSource() : masterSlaveRule.getSlaveDataSourceMap().get(selectedSourceName);
       Preconditions.checkNotNull(selectedSource, "");
       return new NamedDataSource(selectedSourceName, selectedSource);
   }

MasterSlaveConnection

MasterSlaveConnection是继承自AbstractConnectionAdapter的类，实现了connection接口，主要提供MasterSlaveStatement和MasterSlavePreparedStatement构建方式

如下几个方法

@Override
    public Statement createStatement() throws SQLException {
        return new MasterSlaveStatement(this);
    }

  @Override
    public PreparedStatement prepareStatement(final String sql) throws SQLException {
        return new MasterSlavePreparedStatement(this, sql);
    }

同是提供connection的获取方式方法如下，在MasterSlaveStatement,MasterSlavePreparedStatement中有使用到

public Collection<Connection> getConnections(final SQLType sqlType) throws SQLException {
     cachedSQLType = sqlType;
     Map<String, DataSource> dataSources = SQLType.DDL == sqlType ? masterSlaveDataSource.getMasterDataSource() : masterSlaveDataSource.getDataSource(sqlType).toMap();
     Collection<Connection> result = new LinkedList<>();
     for (Entry<String, DataSource> each : dataSources.entrySet()) {
         String dataSourceName = each.getKey();
         if (getCachedConnections().containsKey(dataSourceName)) {
             result.add(getCachedConnections().get(dataSourceName));
             continue;
         }
         Connection connection = each.getValue().getConnection();
         getCachedConnections().put(dataSourceName, connection);
         result.add(connection);
         replayMethodsInvocation(connection);
     }
     return result;
 }

MasterSlavePreparedStatement和MasterSlaveStatement

这两类都是对PreparedStatement和SlaveStatement的封装,提供了他们对应的sql执行方法，两类执行的方法都会调用Connection的获取方式如下两行代码，最终执行jdbc本身的实现，详情可以查看源码

SQLStatement sqlStatement = new SQLJudgeEngine(sql).judge();
Collection connections = connection.getConnections(sqlStatement.getType());

SQLStatement是一个sql类型解析类，sharding-jdbc实现了自己的一套sql的解析规则代码如下

/*
 * Copyright 1999-2015 dangdang.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * </p>
 */

package io.shardingjdbc.core.parsing;

import io.shardingjdbc.core.parsing.lexer.Lexer;
import io.shardingjdbc.core.parsing.lexer.analyzer.Dictionary;
import io.shardingjdbc.core.parsing.lexer.token.Assist;
import io.shardingjdbc.core.parsing.lexer.token.DefaultKeyword;
import io.shardingjdbc.core.parsing.lexer.token.Keyword;
import io.shardingjdbc.core.parsing.lexer.token.TokenType;
import io.shardingjdbc.core.parsing.parser.exception.SQLParsingException;
import io.shardingjdbc.core.parsing.parser.sql.SQLStatement;
import io.shardingjdbc.core.parsing.parser.sql.ddl.DDLStatement;
import io.shardingjdbc.core.parsing.parser.sql.dml.DMLStatement;
import io.shardingjdbc.core.parsing.parser.sql.dql.select.SelectStatement;
import lombok.RequiredArgsConstructor;

/**
 * SQL judge engine.
 *
 * @author zhangliang
 */
@RequiredArgsConstructor
public final class SQLJudgeEngine {
    
    private final String sql;
    
    /**
     * judge SQL Type only.
     *
     * @return SQL statement
     */
    public SQLStatement judge() {
        Lexer lexer = new Lexer(sql, new Dictionary());
        lexer.nextToken();
        while (true) {
            TokenType tokenType = lexer.getCurrentToken().getType();
            if (tokenType instanceof Keyword) {
                if (DefaultKeyword.SELECT == tokenType) {
                    return new SelectStatement();
                }
                if (DefaultKeyword.INSERT == tokenType || DefaultKeyword.UPDATE == tokenType || DefaultKeyword.DELETE == tokenType) {
                    return new DMLStatement();
                }
                if (DefaultKeyword.CREATE == tokenType || DefaultKeyword.ALTER == tokenType || DefaultKeyword.DROP == tokenType || DefaultKeyword.TRUNCATE == tokenType) {
                    return new DDLStatement();
                }
            }
            if (tokenType instanceof Assist && Assist.END == tokenType) {
                throw new SQLParsingException("Unsupported SQL statement: [%s]", sql);
            }
            lexer.nextToken();
        }
    }
}