|
精华帖 (0) :: 良好帖 (12) :: 新手帖 (2) :: 隐藏帖 (0)
|
|
|---|---|
| 作者 | 正文 |
|
发表时间：2010-10-11
1.引言
笔者最近在做一个互联网的“类SNS”应用，应用中用户数量巨大（约4000万）左右，因此，简单的使用传统单一数据库存储肯定是不行的。 参考了业内广泛使用的分库分表，以及使用DAL数据访问层等的做法，笔者决定使用一种最简单的数据源路由选择方式来解决问题。 严格的说，目前的实现不能算是一个解决方案，只能是一种思路的简易实现，笔者也仅花了2天时间来完成（其中1.5天是在看资料和Spring/ibatis的源码）。这里也只是为各位看官提供一个思路参考，顺便给自己留个笔记 
2.系统的设计前提 我们的系统使用了16个数据库实例（目前分布在2台物理机器上，后期将根据系统负荷的增加，逐步移库到16台物理机器上）。16个库是根据用户的UserID进行简单的hash分配。这里值得一说的是，我们既然做了这样的横向切分设计，就已经考虑了系统需求的特性，
在系统中，我们使用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.class
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()方法的重载。下面具体阐述：
（PS：关于DataSource的路由选择规则，可以根据应用场景的不同，自行设计。笔者这里提供两种简单的思路，1.根据HashCode,在上述例子中可以是UserId，进行取模运算，来定位数据库。2.根据上下文设置的关键字key，从map中选择映射的DataSource） 5.将MultiDataSource与Spring，iBatis结合 在完成了上述的编码过程后，就是将这个MultiDataSource与现有Spring和iBatis结合起来配置。 STEP 1。配置多个数据源 笔者这里使用了C3P0作为数据库连接池，这一步和标准的Spring配置一样，唯一不同的是，以前只配置一个，现在要配置多个
<!-- jdbc连接池-1-->
<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>
<!--连接池中保留的最大连接数。Default: 15 -->
<property name="maxPoolSize">
<value>${c3p0.maxPoolSize}</value>
</property>
<!--初始化时获取的连接数，取值应在minPoolSize与maxPoolSize之间。Default: 3 -->
<property name="initialPoolSize">
<value>${c3p0.initialPoolSize}</value>
</property>
<!--每60秒检查所有连接池中的空闲连接。Default: 0 -->
<property name="idleConnectionTestPeriod">
<value>${c3p0.idleConnectionTestPeriod}</value>
</property>
</bean>
<!------------- jdbc连接池-2------------------->
<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>
<!--连接池中保留的最大连接数。Default: 15 -->
<property name="maxPoolSize">
<value>${c3p0.maxPoolSize}</value>
</property>
<!--初始化时获取的连接数，取值应在minPoolSize与maxPoolSize之间。Default: 3 -->
<property name="initialPoolSize">
<value>${c3p0.initialPoolSize}</value>
</property>
<!--每60秒检查所有连接池中的空闲连接。Default: 0 -->
<property name="idleConnectionTestPeriod">
<value>${c3p0.idleConnectionTestPeriod}</value>
</property>
</bean>
<!------------- 更多的链接池配置------------------->
......
STEP 2。将多个数据源都注入到MultiDataSource中
<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>
STEP 3。像使用标准的DataSource一样，使用MultiDataSource <!-- iBatis Client配置 将 MultiDataSource 与iBatis Client 绑定--> <bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean"> <property name="configLocation" value="classpath:SqlMapConfig.xml"/> <property name="dataSource" ref="multiDataSource"></property> </bean> <!-- jdbc事务管理配置 将 MultiDataSource 与事务管理器绑定--> <bean id="jdbc_TransactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager"> <property name="dataSource" ref="multiDataSource"></property> </bean> 至此，我们的程序就可以让Spring来管理多库访问了，但请注意，数据库事务仍然限于单库范围（之前已经说过，这里的应用场景不存在跨库的事务）。 6.Java代码使用例子 首先要说明的是，这里我们只是提供了一个简单的使用范例，在范例中，我们还必须手动的调用API，以确定DataSource的路由规则，在实际的应用中，您可以针对自己的业务特点，对此进行封装，以实现相对透明的路由选择
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;
}
OK，我们的多库横向切分的实验可以暂告一个段落。实际上，要实现一个完整的DAL是非常庞大的工程，而对我们推动巨大的，可能只是很小的一个部分，到处都存在着8-2法则，要如何选择，就看各位看官了！！ 声明：ITeye文章版权属于作者，受法律保护。没有作者书面许可不得转载。
推荐链接
|
| 返回顶楼 | |
|
发表时间：2010-10-11
对DataSourceRouter里面的东西比较感兴趣，不知能否贴点出来看看？
|
| 返回顶楼 | |
|
发表时间：2010-10-11
最后修改：2010-10-11
SNS里大部分信息都是面向文档的而不是面向数据的，事务要求又不严格，用MongoDB来解决才是王道。
|
| 返回顶楼 | |
|
发表时间：2010-10-11
ray_linn 写道 SNS里大部分信息都是面向文档的而不是面向数据的，事务要求又不严格，用MongoDB来解决才是王道。
MongoDB怎么用？一个user一个collection？MongoDB允许的最多那几万个Collection个数根本就不够。 都说ibatis是做大系统的，根本就不是，连原生的水平垂直切分数据库都不支持，这么多年还是那样，放弃它吧！ |
| 返回顶楼 | |
|
发表时间：2010-10-11
最后修改：2010-10-12
myreligion 写道 ray_linn 写道 SNS里大部分信息都是面向文档的而不是面向数据的，事务要求又不严格，用MongoDB来解决才是王道。
MongoDB怎么用？一个user一个collection？MongoDB允许的最多那几万个Collection个数根本就不够。 都说ibatis是做大系统的，根本就不是，连原生的水平垂直切分数据库都不支持，这么多年还是那样，放弃它吧！ 每个user怎么也只是一份document而不是一个collection。如果需要切分，可以每数万个user放入一个collection中。这种问题只需要做个规划就成： 每100个id 存在一个collection里，名字叫 user100,user200,user300....（假设100是个很大的数） 每500个collection 存在一个db中，名字叫 db50000,db100000...... 现在只要有id，比如说701，我们可以推测出，这个id存放在user800 collection中，db应该是db50000。这个算法很简单吧。 object: { $ref: "user800", $id: ObjectID("701"),$db: "db50000" } 搞定。 |
| 返回顶楼 | |
|
发表时间：2010-10-11
你是为了解决Master-Slave问题，还是多个数据库群组之前的选择？
|
| 返回顶楼 | |
|
发表时间：2010-10-12
downpour 写道 你是为了解决Master-Slave问题，还是多个数据库群组之前的选择？
为多数据库的横向切分，提供一个思路
|
| 返回顶楼 | |
|
发表时间：2010-10-12
numen_wlm 写道 对DataSourceRouter里面的东西比较感兴趣，不知能否贴点出来看看？
其实没啥看头的，呵呵
/**
* @author linliangyi2005@gmail.com
* Jul 15, 2010
*/
public class DataSourceRouter {
public static ThreadLocal<RouterStrategy> currentRouterStrategy =
new ThreadLocal<RouterStrategy>();
/**
* 设置MultiDataSource的路由策略
* @param type
* @param key
* @param routeFactor
*/
public static void setRouterStrategy(String type , String key , int routeFactor){
if(type == null){
throw new IllegalArgumentException("RouterStrategy Type must not be null");
}
RouterStrategy rs = currentRouterStrategy.get();
if(rs == null){
rs = new RouterStrategy();
currentRouterStrategy.set(rs);
}
rs.setType(type);
rs.setKey(key);
rs.setRouteFactor(routeFactor);
}
/**
* 数据源路由策略
* @author linliangyi2005@gmail.com
* Jul 15, 2010
*/
public static class RouterStrategy{
public static final String SRATEGY_TYPE_MAP = "MAP";
public static final String SRATEGY_TYPE_CLUSTER = "CLUSTER";
/*
* 可选值 “MAP” ， “CLUSTER”
* MAP ： 根据key从DataSourceMap中选中DS
* CLUSTER : 根据routeFactor参数，通过算法获取群集
*/
private String type;
/*
* “MAP” ROUTE 中的key
*
*/
private String key;
/*
* "CLUSTER" ROUTE时的参数
*/
private int routeFactor;
/*
* True表示RouterStrategy更新过
* False表示没有更新
*/
private boolean refresh;
public String getType() {
return type;
}
public void setType(String type) {
if(this.type != null && !this.type.equals(type)){
this.type = type;
this.refresh = true;
}else if(this.type == null && type != null){
this.type = type;
this.refresh = true;
}
}
public String getKey() {
return key;
}
public void setKey(String key) {
if(this.key != null && !this.key.equals(key)){
this.key = key;
this.refresh = true;
}else if(this.key == null && key != null){
this.key = key;
this.refresh = true;
}
}
public int getRouteFactor() {
return routeFactor;
}
public void setRouteFactor(int routeFactor) {
if(this.routeFactor != routeFactor){
this.routeFactor = routeFactor;
this.refresh = true;
}
}
public boolean isRefresh() {
return refresh;
}
public void setRefresh(boolean refresh) {
this.refresh = refresh;
}
}
}
|
| 返回顶楼 | |
|
发表时间：2010-10-12
感谢楼主分享，这是这几天看过的最好的文章了，支持下！
|
| 返回顶楼 | |
|
发表时间：2010-10-12
我也做了1个数据库(spring+ibatis)水平切分的功能，原理跟楼主差不多，不过我们会有一些跨库的查询。
建议楼主不要直接根据用户id去关联数据库，这样写的太死，比如以后数据量大，数据库有16个变成32个，需要把以前的用户再平均分配到其他数据库的时候就比较麻烦。 我们是直接在用户表记录1个dbid，记录当前用户在哪个库，这样有1个好处就是用户的数据非常方便进行迁移，迁移到其他数据库只足要该下他的dbid就行。 个人浅见，^_^ |
| 返回顶楼 | |
