Spring+Ibatis數據庫水平分庫

時間 2019-12-05

標籤 spring+ibatis spring ibatis 數據庫水平分庫欄目 MyBatis 简体版

原文原文鏈接

1.引言
   筆者最近在作一個互聯網的「類SNS」應用，應用中用戶數量巨大（約4000萬）左右，所以，簡單的使用傳統單一數據庫存儲確定是不行的。

   參考了業內普遍使用的分庫分表，以及使用DAL數據訪問層等的作法，筆者決定使用一種最簡單的數據源路由選擇方式來解決問題。

   嚴格的說，目前的實現不能算是一個解決方案，只能是一種思路的簡易實現，筆者也僅花了2天時間來完成（其中1.5天是在看資料和Spring/ibatis的源碼）。這裏也只是爲各位看官提供一個思路參考，順便給本身留個筆記

2.系統的設計前提
   咱們的系統使用了16個數據庫實例（目前分佈在2臺物理機器上，後期將根據系統負荷的增長，逐步移庫到16臺物理機器上）。16個庫是根據用戶的UserID進行簡單的hash分配。這裏值得一說的是，咱們既然作了這樣的橫向切分設計，就已經考慮了系統需求的特性，html

1.不會發生常常性的跨庫訪問。java
2.主要的業務邏輯都是圍繞UserID爲核心的，在一個單庫事務內便可完成。mysql

   在系統中，咱們使用Spring和iBatis。Spring負責數據庫的事務管理AOP，以及Bean間的IOC。選擇iBatis的最大緣由是對Sql的性能優化，以及後期若是有分表要求的時，能夠很容易實現對sql表名替換。

3.設計思路
   首先，要說明一下筆者的思路，其實很簡單，即「在每次數據庫操做前，肯定當前要選擇的數據庫對象」然後就如同訪問單庫同樣的訪問當前選中的數據庫便可。

   其次，要在每次DB訪問前選擇數據庫，須要明確幾個問題，1.iBatis在何時從DataSource中取得具體的數據庫Connection 的，2.對取得的Connection，iBatis是否進行緩存，由於在多庫狀況下Connection被緩存就意味着沒法及時改變數據庫連接選擇。 3.因爲咱們使用了Spring來管理DB事務，所以必須搞清Spring對DB Connction的開關攔截過程是否會影響多DataSource的狀況。

   幸運的是，研究源碼的結果發現，iBatis和Spring都是經過標準的DataSource接口來控制
Connection的，這就爲咱們省去了不少的麻煩，只須要實現一個可以支持多個數據庫的DataSource，就能達到咱們的目標。

4.代碼與實現
多數據庫的DataSource實現：MultiDataSource.classweb

Java代碼算法

[java] view plain copyspring

import java.io.PrintWriter; sql
import java.sql.Connection; 數據庫
import java.sql.SQLException; apache
import java.util.ArrayList; 緩存
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import javax.sql.DataSource;
import org.apache.log4j.Logger;
import com.xxx.sql.DataSourceRouter.RouterStrategy;
/**
* 複合多數據源（Alpha）
* @author linliangyi2005@gmail.com
* Jul 15, 2010
*/
public class MultiDataSource implements DataSource {
static Logger logger = Logger.getLogger(MultiDataSource.class);
//當前線程對應的實際DataSource
private ThreadLocal<DataSource> currentDataSourceHolder = new ThreadLocal<DataSource>();
//使用Key-Value映射的DataSource
private Map<String , DataSource> mappedDataSources;
//使用橫向切分的分佈式DataSource
private ArrayList<DataSource> clusterDataSources;
public MultiDataSource(){
mappedDataSources = new HashMap<String , DataSource>(4);
clusterDataSources = new ArrayList<DataSource>(4);
}
/**
* 數據庫鏈接池初始化
* 該方法一般在web 應用啓動時調用
*/
public void initialMultiDataSource(){
for(DataSource ds : clusterDataSources){
if(ds != null){
Connection conn = null;
try {
conn = ds.getConnection();
} catch (SQLException e) {
e.printStackTrace();
} finally{
if(conn != null){
try {
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
conn = null;
}
}
}
}
Collection<DataSource> dsCollection = mappedDataSources.values();
for(DataSource ds : dsCollection){
if(ds != null){
Connection conn = null;
try {
conn = ds.getConnection();
} catch (SQLException e) {
e.printStackTrace();
} finally{
if(conn != null){
try {
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
conn = null;
}
}
}
}
}
/**
* 獲取當前線程綁定的DataSource
* @return
*/
public DataSource getCurrentDataSource() {
//若是路由策略存在，且更新過，則根據路由算法選擇新的DataSource
RouterStrategy strategy = DataSourceRouter.currentRouterStrategy.get();
if(strategy == null){
throw new IllegalArgumentException("DataSource RouterStrategy No found.");
}
if(strategy != null && strategy.isRefresh()){
if(RouterStrategy.SRATEGY_TYPE_MAP.equals(strategy.getType())){
this.choiceMappedDataSources(strategy.getKey());
}else if(RouterStrategy.SRATEGY_TYPE_CLUSTER.equals(strategy.getType())){
this.routeClusterDataSources(strategy.getRouteFactor());
}
strategy.setRefresh(false);
}
return currentDataSourceHolder.get();
}
public Map<String, DataSource> getMappedDataSources() {
return mappedDataSources;
}
public void setMappedDataSources(Map<String, DataSource> mappedDataSources) {
this.mappedDataSources = mappedDataSources;
}
public ArrayList<DataSource> getClusterDataSources() {
return clusterDataSources;
}
public void setClusterDataSources(ArrayList<DataSource> clusterDataSources) {
this.clusterDataSources = clusterDataSources;
}
/**
* 使用Key選擇當前的數據源
* @param key
*/
public void choiceMappedDataSources(String key){
DataSource ds = this.mappedDataSources.get(key);
if(ds == null){
throw new IllegalStateException("No Mapped DataSources Exist!");
}
this.currentDataSourceHolder.set(ds);
}
/**
* 使用取模算法，在羣集數據源中作路由選擇
* @param routeFactor
*/
public void routeClusterDataSources(int routeFactor){
int size = this.clusterDataSources.size();
if(size == 0){
throw new IllegalStateException("No Cluster DataSources Exist!");
}
int choosen = routeFactor % size;
DataSource ds = this.clusterDataSources.get(choosen);
if(ds == null){
throw new IllegalStateException("Choosen DataSources is null!");
}
logger.debug("Choosen DataSource No." + choosen+ " : " + ds.toString());
this.currentDataSourceHolder.set(ds);
}
/* (non-Javadoc)
* @see javax.sql.DataSource#getConnection()
*/
public Connection getConnection() throws SQLException {
if(getCurrentDataSource() != null){
return getCurrentDataSource().getConnection();
}
return null;
}
/* (non-Javadoc)
* @see javax.sql.DataSource#getConnection(java.lang.String, java.lang.String)
*/
public Connection getConnection(String username, String password)
throws SQLException {
if(getCurrentDataSource() != null){
return getCurrentDataSource().getConnection(username , password);
}
return null;
}
/* (non-Javadoc)
* @see javax.sql.CommonDataSource#getLogWriter()
*/
public PrintWriter getLogWriter() throws SQLException {
if(getCurrentDataSource() != null){
return getCurrentDataSource().getLogWriter();
}
return null;
}
/* (non-Javadoc)
* @see javax.sql.CommonDataSource#getLoginTimeout()
*/
public int getLoginTimeout() throws SQLException {
if(getCurrentDataSource() != null){
return getCurrentDataSource().getLoginTimeout();
}
return 0;
}
/* (non-Javadoc)
* @see javax.sql.CommonDataSource#setLogWriter(java.io.PrintWriter)
*/
public void setLogWriter(PrintWriter out) throws SQLException {
if(getCurrentDataSource() != null){
getCurrentDataSource().setLogWriter(out);
}
}
/* (non-Javadoc)
* @see javax.sql.CommonDataSource#setLoginTimeout(int)
*/
public void setLoginTimeout(int seconds) throws SQLException {
if(getCurrentDataSource() != null){
getCurrentDataSource().setLoginTimeout(seconds);
}
}
/* (non-Javadoc)
* 該接口方法since 1.6
* 不是全部的DataSource都實現有這個方法
* @see java.sql.Wrapper#isWrapperFor(java.lang.Class)
*/
public boolean isWrapperFor(Class<?> iface) throws SQLException {
// if(getCurrentDataSource() != null){
// return getCurrentDataSource().isWrapperFor(iface);
// }
return false;
}
/* (non-Javadoc)
* 該接口方法since 1.6
* 不是全部的DataSource都實現有這個方法
* @see java.sql.Wrapper#unwrap(java.lang.Class)
*/
public <T> T unwrap(Class<T> iface) throws SQLException {
// if(getCurrentDataSource() != null){
// return getCurrentDataSource().unwrap(iface);
// }
return null;
}
import java.io.PrintWriter;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import javax.sql.DataSource;
import org.apache.log4j.Logger;
import com.xxx.sql.DataSourceRouter.RouterStrategy;
/**
* 複合多數據源（Alpha）
* @author linliangyi2005@gmail.com
* Jul 15, 2010
*/
public class MultiDataSource implements DataSource {
static Logger logger = Logger.getLogger(MultiDataSource.class);
//當前線程對應的實際DataSource
private ThreadLocal<DataSource> currentDataSourceHolder = new ThreadLocal<DataSource>();
//使用Key-Value映射的DataSource
private Map<String , DataSource> mappedDataSources;
//使用橫向切分的分佈式DataSource
private ArrayList<DataSource> clusterDataSources;
public MultiDataSource(){
mappedDataSources = new HashMap<String , DataSource>(4);
clusterDataSources = new ArrayList<DataSource>(4);
}
/**
* 數據庫鏈接池初始化
* 該方法一般在web 應用啓動時調用
*/
public void initialMultiDataSource(){
for(DataSource ds : clusterDataSources){
if(ds != null){
Connection conn = null;
try {
conn = ds.getConnection();
} catch (SQLException e) {
e.printStackTrace();
} finally{
if(conn != null){
try {
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
conn = null;
}
}
}
}
Collection<DataSource> dsCollection = mappedDataSources.values();
for(DataSource ds : dsCollection){
if(ds != null){
Connection conn = null;
try {
conn = ds.getConnection();
} catch (SQLException e) {
e.printStackTrace();
} finally{
if(conn != null){
try {
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
conn = null;
}
}
}
}
}
/**
* 獲取當前線程綁定的DataSource
* @return
*/
public DataSource getCurrentDataSource() {
//若是路由策略存在，且更新過，則根據路由算法選擇新的DataSource
RouterStrategy strategy = DataSourceRouter.currentRouterStrategy.get();
if(strategy == null){
throw new IllegalArgumentException("DataSource RouterStrategy No found.");
}
if(strategy != null && strategy.isRefresh()){
if(RouterStrategy.SRATEGY_TYPE_MAP.equals(strategy.getType())){
this.choiceMappedDataSources(strategy.getKey());
}else if(RouterStrategy.SRATEGY_TYPE_CLUSTER.equals(strategy.getType())){
this.routeClusterDataSources(strategy.getRouteFactor());
}
strategy.setRefresh(false);
}
return currentDataSourceHolder.get();
}
public Map<String, DataSource> getMappedDataSources() {
return mappedDataSources;
}
public void setMappedDataSources(Map<String, DataSource> mappedDataSources) {
this.mappedDataSources = mappedDataSources;
}
public ArrayList<DataSource> getClusterDataSources() {
return clusterDataSources;
}
public void setClusterDataSources(ArrayList<DataSource> clusterDataSources) {
this.clusterDataSources = clusterDataSources;
}
/**
* 使用Key選擇當前的數據源
* @param key
*/
public void choiceMappedDataSources(String key){
DataSource ds = this.mappedDataSources.get(key);
if(ds == null){
throw new IllegalStateException("No Mapped DataSources Exist!");
}
this.currentDataSourceHolder.set(ds);
}
/**
* 使用取模算法，在羣集數據源中作路由選擇
* @param routeFactor
*/
public void routeClusterDataSources(int routeFactor){
int size = this.clusterDataSources.size();
if(size == 0){
throw new IllegalStateException("No Cluster DataSources Exist!");
}
int choosen = routeFactor % size;
DataSource ds = this.clusterDataSources.get(choosen);
if(ds == null){
throw new IllegalStateException("Choosen DataSources is null!");
}
logger.debug("Choosen DataSource No." + choosen+ " : " + ds.toString());
this.currentDataSourceHolder.set(ds);
}
/* (non-Javadoc)
* @see javax.sql.DataSource#getConnection()
*/
public Connection getConnection() throws SQLException {
if(getCurrentDataSource() != null){
return getCurrentDataSource().getConnection();
}
return null;
}
/* (non-Javadoc)
* @see javax.sql.DataSource#getConnection(java.lang.String, java.lang.String)
*/
public Connection getConnection(String username, String password)
throws SQLException {
if(getCurrentDataSource() != null){
return getCurrentDataSource().getConnection(username , password);
}
return null;
}
/* (non-Javadoc)
* @see javax.sql.CommonDataSource#getLogWriter()
*/
public PrintWriter getLogWriter() throws SQLException {
if(getCurrentDataSource() != null){
return getCurrentDataSource().getLogWriter();
}
return null;
}
/* (non-Javadoc)
* @see javax.sql.CommonDataSource#getLoginTimeout()
*/
public int getLoginTimeout() throws SQLException {
if(getCurrentDataSource() != null){
return getCurrentDataSource().getLoginTimeout();
}
return 0;
}
/* (non-Javadoc)
* @see javax.sql.CommonDataSource#setLogWriter(java.io.PrintWriter)
*/
public void setLogWriter(PrintWriter out) throws SQLException {
if(getCurrentDataSource() != null){
getCurrentDataSource().setLogWriter(out);
}
}
/* (non-Javadoc)
* @see javax.sql.CommonDataSource#setLoginTimeout(int)
*/
public void setLoginTimeout(int seconds) throws SQLException {
if(getCurrentDataSource() != null){
getCurrentDataSource().setLoginTimeout(seconds);
}
}
/* (non-Javadoc)
* 該接口方法since 1.6
* 不是全部的DataSource都實現有這個方法
* @see java.sql.Wrapper#isWrapperFor(java.lang.Class)
*/
public boolean isWrapperFor(Class<?> iface) throws SQLException {
// if(getCurrentDataSource() != null){
// return getCurrentDataSource().isWrapperFor(iface);
// }
return false;
}
/* (non-Javadoc)
* 該接口方法since 1.6
* 不是全部的DataSource都實現有這個方法
* @see java.sql.Wrapper#unwrap(java.lang.Class)
*/
public <T> T unwrap(Class<T> iface) throws SQLException {
// if(getCurrentDataSource() != null){
// return getCurrentDataSource().unwrap(iface);
// }
return null;
}

這個類實現了DataSource的標準接口，而最核心的部分是getConnection()方法的重載。下面具體闡述：

1.實例變量 clusterDataSources 是一個DataSource 的 ArrayList它存儲了多個數據庫的DataSource實例，咱們使用Spring的IOC功能，將多個DataSource注入到這個list中。
2.實例變量 mappedDataSources 是一個DataSource 的Map，它與clusterDataSources 同樣用來存儲多個數據庫的DataSource實例，不一樣的是，它可使用key直接獲取DataSource。咱們同樣會使用Spring的IOC功能，將多個DataSource注入到這個Map中。
3.實例變量currentDataSourceHolder ，他是一個ThreadLocal變量，保存與當前線程相關的且已經取得的DataSource實例。這是爲了在同一線程中，屢次訪問同一數據庫時，不須要再從新作路由選擇。
4.當外部類調用getConnection()方法時，方法將根據上下文的路由規則，從clusterDataSources 或者 mappedDataSources 選擇對應DataSource，並返回其中的Connection。

（PS：關於DataSource的路由選擇規則，能夠根據應用場景的不一樣，自行設計。筆者這裏提供兩種簡單的思路，1.根據 HashCode,在上述例子中能夠是UserId，進行取模運算，來定位數據庫。2.根據上下文設置的關鍵字key，從map中選擇映射的 DataSource）

5.將MultiDataSource與Spring，iBatis結合
在完成了上述的編碼過程後，就是將這個MultiDataSource與現有Spring和iBatis結合起來配置。

STEP 1。配置多個數據源
筆者這裏使用了C3P0做爲數據庫鏈接池，這一步和標準的Spring配置同樣，惟一不一樣的是，之前只配置一個，如今要配置多個

Xml代碼

[xml] view plain copy

<textarea cols="84" rows="15" name="code" class="xhtml">
<bean id="c3p0_dataSource_1" class="com.mchange.v2.c3p0.ComboPooledDataSource" destroy-method="close">
<property name="driverClass">
<value>${jdbc.driverClass}</value>
</property>
<property name="jdbcUrl">
<value>${mysql.url_1}</value>
</property>
<property name="user">
value>${jdbc.username}</value>
</property>
<property name="password">
<value>${jdbc.password}</value>
</property>
<property name="minPoolSize">
<value>${c3p0.minPoolSize}</value>
</property>
<property name="maxPoolSize">
<value>${c3p0.maxPoolSize}</value>
</property>
<property name="initialPoolSize">
<value>${c3p0.initialPoolSize}</value>
</property>
<property name="idleConnectionTestPeriod">
<value>${c3p0.idleConnectionTestPeriod}</value>
</property>
</bean>
<bean id="c3p0_dataSource_2" class="com.mchange.v2.c3p0.ComboPooledDataSource" destroy-method="close">
<property name="driverClass">
<value>${jdbc.driverClass}</value>
</property>
<property name="jdbcUrl">
<value>${mysql.url_2}</value>
</property>
<property name="user">
<value>${jdbc.username}</value>
</property>
<property name="password">
<value>${jdbc.password}</value>
</property>
<property name="minPoolSize">
<value>${c3p0.minPoolSize}</value>
</property>
<property name="maxPoolSize">
<value>${c3p0.maxPoolSize}</value>
</property>
<property name="initialPoolSize">
<value>${c3p0.initialPoolSize}</value>
</property>
<property name="idleConnectionTestPeriod">
<value>${c3p0.idleConnectionTestPeriod}</value>
</property>
</bean>
......
<bean id="c3p0_dataSource_1" class="com.mchange.v2.c3p0.ComboPooledDataSource" destroy-method="close">
<property name="driverClass">
<value>${jdbc.driverClass}</value>
</property>
<property name="jdbcUrl">
<value>${mysql.url_1}</value>
</property>
<property name="user">
<value>${jdbc.username}</value>
</property>
<property name="password">
<value>${jdbc.password}</value>
</property>
<property name="minPoolSize">
<value>${c3p0.minPoolSize}</value>
</property>
<property name="maxPoolSize">
<value>${c3p0.maxPoolSize}</value>
</property>
<property name="initialPoolSize">
<value>${c3p0.initialPoolSize}</value>
</property>
<property name="idleConnectionTestPeriod">
<value>${c3p0.idleConnectionTestPeriod}</value>
</property>
</bean>
<bean id="c3p0_dataSource_2" class="com.mchange.v2.c3p0.ComboPooledDataSource" destroy-method="close">
<property name="driverClass">
<value>${jdbc.driverClass}</value>
</property>
<property name="jdbcUrl">
<value>${mysql.url_2}</value>
</property>
<property name="user">
<value>${jdbc.username}</value>
</property>
<property name="password">
<value>${jdbc.password}</value>
</property>
<property name="minPoolSize">
<value>${c3p0.minPoolSize}</value>
</property>
<property name="maxPoolSize">
<value>${c3p0.maxPoolSize}</value>
</property>
<property name="initialPoolSize">
<value>${c3p0.initialPoolSize}</value>
</property>
<property name="idleConnectionTestPeriod">
<value>${c3p0.idleConnectionTestPeriod}</value>
</property>
</bean>
......
</textarea>

STEP 2。將多個數據源都注入到MultiDataSource中

Xml代碼

[xhtml] view plain copy

<bean id="multiDataSource" class="com.xxx.sql.MultiDataSource">
<property name="clusterDataSources">
<list>
<ref bean="c3p0_dataSource_1" />
<ref bean="c3p0_dataSource_2" />
<ref bean="c3p0_dataSource_3" />
<ref bean="c3p0_dataSource_4" />
<ref bean="c3p0_dataSource_5" />
<ref bean="c3p0_dataSource_6" />
<ref bean="c3p0_dataSource_7" />
<ref bean="c3p0_dataSource_8" />
</list>
</property>
<property name="mappedDataSources">
<map>
<entry key="system" value-ref="c3p0_dataSource_system" />
</map>
</property>
</bean>
<bean id="multiDataSource" class="com.xxx.sql.MultiDataSource">
<property name="clusterDataSources">
<list>
<ref bean="c3p0_dataSource_1" />
<ref bean="c3p0_dataSource_2" />
<ref bean="c3p0_dataSource_3" />
<ref bean="c3p0_dataSource_4" />
<ref bean="c3p0_dataSource_5" />
<ref bean="c3p0_dataSource_6" />
<ref bean="c3p0_dataSource_7" />
<ref bean="c3p0_dataSource_8" />
</list>
</property>
<property name="mappedDataSources">
<map>
<entry key="system" value-ref="c3p0_dataSource_system" />
</map>
</property>
</bean>

[xml] view plain copy

STEP 3。像使用標準的DataSource同樣，使用MultiDataSource

Xml代碼

[xml] view plain copy

<textarea cols="84" rows="15" name="code" class="xhtml">
<bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean">
<property name="configLocation" value="classpath:SqlMapConfig.xml"/>
<property name="dataSource" ref="multiDataSource"></property>
</bean>
<bean id="jdbc_TransactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="multiDataSource"></property>
</bean>
<bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean">
<property name="configLocation" value="classpath:SqlMapConfig.xml"/>
<property name="dataSource" ref="multiDataSource"></property>
</bean>
<bean id="jdbc_TransactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="multiDataSource"></property>
</bean>
</textarea>

至此，咱們的程序就可讓Spring來管理多庫訪問了，但請注意，數據庫事務仍然限於單庫範圍（以前已經說過，這裏的應用場景不存在跨庫的事務）。

6.Java代碼使用例子
首先要說明的是，這裏咱們只是提供了一個簡單的使用範例，在範例中，咱們還必須手動的調用API，以肯定DataSource的路由規則，在實際的應用中，您能夠針對本身的業務特色，對此進行封裝，以實現相對透明的路由選擇

Java代碼

[java] view plain copy

<textarea cols="81" rows="15" name="code" class="java">public boolean addUserGameInfo(UserGameInfo userGameInfo){
//1.根據UserGameInfo.uid 進行數據源路由選擇 DataSourceRouter.setRouterStrategy(
RouterStrategy.SRATEGY_TYPE_CLUSTER ,
null,
userGameInfo.getUid());
//2.數據庫存儲
try {
userGameInfoDAO.insert(userGameInfo);
return true;
} catch (SQLException e) {
e.printStackTrace();
logger.debug("Insert UserGameInfo failed. " + userGameInfo.toString());
}
return false;
}
public boolean addUserGameInfo(UserGameInfo userGameInfo){
//1.根據UserGameInfo.uid 進行數據源路由選擇
DataSourceRouter.setRouterStrategy(
RouterStrategy.SRATEGY_TYPE_CLUSTER ,
null,
userGameInfo.getUid());
//2.數據庫存儲
try {
userGameInfoDAO.insert(userGameInfo);
return true;
} catch (SQLException e) {
e.printStackTrace();
logger.debug("Insert UserGameInfo failed. " + userGameInfo.toString());
}
return false;
}
</textarea>