Mysql主从Java端实现

继昨天的Mysql主从的概述及基本的配置。今天趁着不是很忙的时候整理一下主从JAVA端的代码实现。下面开始贴代码：

Spring MVC 的datasorce配置，这里我们使用的阿里的druid（德鲁伊）数据库连接池，先配置两个数据库链接池。分别连接主、从两个库。然后使用spring 的一个主要特性AOP切面编程来根据方法的前缀命名来选择是使用主数据库还是使用从数据库。（我选择切的是controller层，当用户访问方法的 时候，我就已经决定了是选择哪个数据源进行处理）

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:aop="http://www.springframework.org/schema/aop"
xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
">
<!-- spring-datasource.xml 主要进行应用程序数据源的分配。根据请求的方法名称规范，自动分配主/从数据库 -->

<!-- 配置数据源 -->
<bean name="dataSourceMaster" class="com.alibaba.druid.pool.DruidDataSource"
init-method="init" destroy-method="close">
<property name="url" value="${master_jdbc_url}" />
<property name="username" value="${master_jdbc_username}" />
<property name="password" value="${master_jdbc_password}" />

<property name="initialSize" value="${initialSize}" />
<property name="maxActive" value="${maxActive}" />
<property name="maxIdle" value="${maxIdle}" />
<property name="minIdle" value="${minIdle}" />
<property name="maxWait" value="${maxWait}" />

<property name="validationQuery" value="${validationQuery}" />
<property name="testOnBorrow" value="${testOnBorrow}" />
<property name="testOnReturn" value="${testOnReturn}" />
<property name="testWhileIdle" value="${testWhileIdle}" />

<property name="timeBetweenEvictionRunsMillis" value="${timeBetweenEvictionRunsMillis}" />
<property name="minEvictableIdleTimeMillis" value="${minEvictableIdleTimeMillis}" />

<property name="removeAbandoned" value="${removeAbandoned}" />
<property name="removeAbandonedTimeout" value="${removeAbandonedTimeout}" />
<property name="logAbandoned" value="${logAbandoned}" />

<property name="filters" value="stat" />
</bean>

<!-- 配置数据源 -->
<bean name="dataSourceSlave" class="com.alibaba.druid.pool.DruidDataSource"
init-method="init" destroy-method="close">
<property name="url" value="${slave_jdbc_url}" />
<property name="username" value="${slave_jdbc_username}" />
<property name="password" value="${slave_jdbc_password}" />

<property name="initialSize" value="${initialSize}" />
<property name="maxActive" value="${maxActive}" />
<property name="maxIdle" value="${maxIdle}" />
<property name="minIdle" value="${minIdle}" />
<property name="maxWait" value="${maxWait}" />

<property name="validationQuery" value="${validationQuery}" />
<property name="testOnBorrow" value="${testOnBorrow}" />
<property name="testOnReturn" value="${testOnReturn}" />
<property name="testWhileIdle" value="${testWhileIdle}" />

<property name="timeBetweenEvictionRunsMillis" value="${timeBetweenEvictionRunsMillis}" />
<property name="minEvictableIdleTimeMillis" value="${minEvictableIdleTimeMillis}" />

<property name="removeAbandoned" value="${removeAbandoned}" />
<property name="removeAbandonedTimeout" value="${removeAbandonedTimeout}" />
<property name="logAbandoned" value="${logAbandoned}" />

<property name="filters" value="stat" />

</bean>

<!-- 自定义数据源 -->
<bean id="dynamicDataSourceHolder" class="com.changba.idc.datasource.DynamicDataSourceHolder">
<property name="slavetDataSources">
<map key-type="java.lang.String">
<entry key="slaveDataSource1" value-ref="dataSourceSlave"></entry>
</map>
</property>
<property name="masterDataSources">
<map key-type="java.lang.String">
<entry key="masterDataSource1" value-ref="dataSourceMaster"></entry>
</map>
</property>
<property name="defaultTargetDataSource" ref="dataSourceMaster" />
</bean>

<!-- mybatis session factory -->
<bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
<property name="dataSource" ref="dynamicDataSourceHolder" />
<property name="mapperLocations" value="classpath:com/changba/idc/mapper/imp/*.xml" />
<property name="configLocation" value="classpath:mybatis-config.xml" />
</bean>

<bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
<property name="basePackage" value="com.changba.idc.mapper" />
<property name="sqlSessionFactoryBeanName" value="sqlSessionFactory" />
</bean>

<!--事务管理DataSourceTransactionManager -->
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dynamicDataSourceHolder" />
</bean>

<!--最后一步定义拦截-->
<tx:advice id="transactionAdvice" transaction-manager="transactionManager">
<tx:attributes>
<tx:method name="add*" propagation="REQUIRED" />
<tx:method name="append*" propagation="REQUIRED" />
<tx:method name="insert*" propagation="REQUIRED" />
<tx:method name="save*" propagation="REQUIRED" />
<tx:method name="update*" propagation="REQUIRED" />
<tx:method name="modify*" propagation="REQUIRED" />
<tx:method name="edit*" propagation="REQUIRED" />
<tx:method name="delete*" propagation="REQUIRED" />
<tx:method name="remove*" propagation="REQUIRED" />
<tx:method name="repair" propagation="REQUIRED" />
<tx:method name="delAndRepair" propagation="REQUIRED" />

<tx:method name="get*" propagation="SUPPORTS" read-only="true" />
<tx:method name="find*" propagation="SUPPORTS" read-only="true" />
<tx:method name="load*" propagation="SUPPORTS" read-only="true" />
<tx:method name="search*" propagation="SUPPORTS" read-only="true" />
<tx:method name="datagrid*" propagation="SUPPORTS" read-only="true" />

<tx:method name="*" propagation="SUPPORTS" />
</tx:attributes>
</tx:advice>

<!-- 定义声明式事务 -->
<aop:config>
<aop:pointcut id="transactionPointcut" expression="execution(* com.changba.idc.service..*.*(..))" />
<!-- 用来定义只有一个通知和一个切入点的切面 -->
<aop:advisor pointcut-ref="transactionPointcut" advice-ref="transactionAdvice" />
</aop:config>

<!-- 系统服务组件的切面Bean -->
<bean id="dynamicDataSourceAop" class="com.changba.idc.datasource.DynamicDataSourceAop"></bean>

<!-- expose-proxy="true" 解决“自我调用”而导致的不能设置正确的事务属性 -->
<aop:config expose-proxy="true">
<!-- 通过AOP切面实现读/写库选择 -->
<!-- <aop:aspect>：用来定义切面，该切面可以包含多个切入点和通知，而且标签内部的通知和切入点定义是无序的 -->
<!-- order 用来控制AOP通知的优先级，值越小，优先级越高。从而保证在操作事务之前已经决定了使用读/写库 -->
<aop:aspect id="dataSourceAspect" ref="dynamicDataSourceAop">
<!-- 只对业务逻辑层实施事务 -->
<aop:pointcut id="dataSourcePointcut" expression="execution(* com.changba.idc.controller.*.*(..))" />
<aop:around method="doAroundMethod" pointcut-ref="dataSourcePointcut"/>
</aop:aspect>
</aop:config>

</beans>

上边的配置是实现读写分离基本的配置，具体的JAVA代码实现如下：
package com.changba.idc.datasource;

import org.aspectj.lang.ProceedingJoinPoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

/**
* 数据源的选取,在操作controller方法前先选取数据源
*
* @author An Dong
*
*/
public class DynamicDataSourceAop {
Logger logger = LoggerFactory.getLogger(DynamicDataSourceHolder.class);

@Autowired
private DynamicDataSourceHolder dataSourceHolder;

//AOP环绕切点
public Object doAroundMethod(ProceedingJoinPoint pjp) throws Throwable {
Object response = null;
// method为方法名。并不是URL访问的名称
String method = pjp.getSignature().getName();
boolean hasBinded = false;
try {
// hasBinded 是否已经绑定datasource
hasBinded = dataSourceHolder.hasBindedDataSourse();
logger.info("判断是否已经绑定数据源 ******* "+hasBinded+" 当前要执行的方法名称--------"+method);
if (!hasBinded) {
if (method.startsWith("query") || method.startsWith("select") || method.startsWith("find")
|| method.startsWith("get") || method.startsWith("load") || method.startsWith("search")) {
dataSourceHolder.markSlave();
} else {
dataSourceHolder.markMaster();
}
}
//调用proceed()方法作为环绕通知执行方法前后的分水岭
response = pjp.proceed();
} finally {
if (!hasBinded) {
logger.info("请求的方法执行完毕，清除已标记选取的数据源");
dataSourceHolder.markRemove();
}
}
return response;
}

}
其中dataSourceHolder 类中具体的方法，这里也是真正实现读写分离的主要逻辑处理的地方。
package com.changba.idc.datasource;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;

import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
/**
* 向spring datasource 提供数据源选取key，对外提供从slave选取数据源，或从master选取数据源方法。 默认从master选取数据源
* InitializingBean接口为bean提供了初始化方法的方式，它只包括afterPropertiesSet方法，凡是继承该接口的类，在初始化bean的时候会执行该方法。
*
* @author An Dong
*
*/
public class DynamicDataSourceHolder extends AbstractRoutingDataSource implements InitializingBean{
// 线程本地环境
private static final ThreadLocal<String> dataSourceHolder = new ThreadLocal<String>();
// 可选取slave的keys
private List<String> slaveDataSourcesKeys;
// 可选取master的keys
private List<String> masterDataSourcesKeys;
//从库数据源
private Map<String, DataSource> slavetDataSources;
//主库数据源
private Map<String, DataSource> masterDataSources;
private Logger logger = LoggerFactory.getLogger(DynamicDataSourceHolder.class);

@Override
//spring 执行数据源切换的核心方法。
protected Object determineCurrentLookupKey() {
return dataSourceHolder.get();
}

@Override
public void afterPropertiesSet() {
// 数据检验和合并
logger.info("开始向Spring DataSource 提供数据源选取");
Map<Object, Object> allDataSources = new HashMap<Object, Object>();
allDataSources.putAll(masterDataSources);
if (slavetDataSources != null) {
allDataSources.putAll(slavetDataSources);
}
super.setTargetDataSources(allDataSources);
super.afterPropertiesSet();
logger.info("主库数据源：："+masterDataSources.keySet());
logger.info("从库数据源：："+slavetDataSources.keySet());
logger.info("已经完提供数据源选取");
}
/**
* 注册slave datasource
*
* @param slavetDataSources
*/
public void setSlavetDataSources(Map<String, DataSource> slavetDataSources) {
if (slavetDataSources == null || slavetDataSources.size() == 0) {
return;
}
logger.info("提供可选取从库数据源：{}", slavetDataSources.keySet());
this.slavetDataSources = slavetDataSources;
slaveDataSourcesKeys = new ArrayList<String>();
for (Entry<String, DataSource> entry : slavetDataSources.entrySet()) {
slaveDataSourcesKeys.add(entry.getKey());
}
}
/**
* 注册master datasource
*
* @param masterDataSources
*/
public void setMasterDataSources(Map<String, DataSource> masterDataSources) {
if (masterDataSources == null) {
throw new IllegalArgumentException("Property 'masterDataSources' is required");
}
logger.info("提供可选取主库数据源：{}", masterDataSources.keySet());
this.masterDataSources = masterDataSources;
this.masterDataSourcesKeys = new ArrayList<String>();
for (Entry<String, DataSource> entry : masterDataSources.entrySet()) {
masterDataSourcesKeys.add(entry.getKey());
}
}

/**
* 标记选取从库数据源
*/
public void markSlave() {
if (dataSourceHolder.get() != null) {
// 从现在的策略来看,不允许标记两次,直接抛异常,优于早发现问题
throw new IllegalArgumentException("当前已有选取数据源,不允许覆盖,已选数据源key:" + dataSourceHolder.get());
}
logger.info("查询的是从库");
String dataSourceKey = selectFromSlave();
setDataSource(dataSourceKey);
}

/**
* 标记选取主库数据源
*/
public void markMaster() {
if (dataSourceHolder.get() != null) {
// 从现在的策略来看,不允许标记两次,直接抛异常,优于早发现问题
throw new IllegalArgumentException("当前已有选取数据源,不允许覆盖,已选数据源key:" + dataSourceHolder.get());
}
logger.info("查询的是主库");
String dataSourceKey = selectFromMaster();
setDataSource(dataSourceKey);
}

/**
* 删除己标记选取的数据源
*/
public void markRemove() {
dataSourceHolder.remove();
}
/**
* 是否已经绑定datasource
* 绑定：true
* 没绑定：false
* @return
*/
public boolean hasBindedDataSourse(){
boolean hasBinded= dataSourceHolder.get()!=null;
return hasBinded;
}

private String selectFromSlave() {
if (slavetDataSources == null) {
logger.info("提供可选取slave数据源：{},将自动切换从主master选取数据源", slavetDataSources);
return selectFromMaster();
} else {
Random random = new Random();
return slaveDataSourcesKeys.get(random.nextInt(slaveDataSourcesKeys.size()));
}
}

private String selectFromMaster() {
Random random = new Random();
String dataSourceKey = masterDataSourcesKeys.get(random.nextInt(masterDataSourcesKeys.size()));
return dataSourceKey;
}

private void setDataSource(String dataSourceKey) {
logger.info("数据源 key *-*-*-*-*-* "+dataSourceKey);
dataSourceHolder.set(dataSourceKey);
}
}



上边的两个JAVA类就不过多解释了。代码注释也比较多，能够看明白的。好了，主从的JAVA代码方面也已经介绍完了。有一个说的不是很准的地方，是这个主从的代码不仅仅应用于MySQL数据库的。其他数据库应该也是通用的，只是我没有尝试。有小伙伴们尝试后也可以给博主进行反馈呢。