Sharding-Jdbc之讀寫分離導讀

前言

      Sharding-JDBC是一個開源的分佈式數據庫中間件，它無需額外部署和依賴，徹底兼容JDBC和各類ORM框架。Sharding-JDBC做爲面向開發的微服務雲原生基礎類庫，完整的實現了分庫分表、讀寫分離和分佈式主鍵功能，並初步實現了柔性事務。

以2.0.3版本爲例maven包依賴以下

<project xmlns="http://maven.apache.org/POM/4.0.0"
	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>
</project>

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<Connection> 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();
        }
    }
}