使用Spring实现MySQL读写分离

/**
* 如果在spring配置了事务的策略，则标记了ReadOnly的方法用从库Slave, 其它使用主库Master。
* 如果没有配置事务策略, 则采用方法名匹配, 以query、find、get开头的方法用Slave，其它用Master。
*/
public class DataSourceAspect { private List<String> slaveMethodPattern = new ArrayList<String>(); //保存有readonly属性的带通配符方法名
private static final String[] defaultSlaveMethodStartWith = new String[]{"query", "find", "get" };
private String[] slaveMethodStartWith; //保存有slaveMethodStartWith属性的方法名头部 //注入
public void setTxAdvice(TransactionInterceptor txAdvice) throws Exception {
if (txAdvice == null) {
// 没有配置事务策略
return;
}
//从txAdvice获取策略配置信息
TransactionAttributeSource transactionAttributeSource = txAdvice.getTransactionAttributeSource();
if (!(transactionAttributeSource instanceof NameMatchTransactionAttributeSource)) {
return;
}
//使用反射技术获取到NameMatchTransactionAttributeSource对象中的nameMap属性值
NameMatchTransactionAttributeSource matchTransactionAttributeSource = (NameMatchTransactionAttributeSource) transactionAttributeSource;
Field nameMapField = ReflectionUtils.findField(NameMatchTransactionAttributeSource.class, "nameMap");
nameMapField.setAccessible(true); //设置该字段可访问
//获取nameMap的值
Map<String, TransactionAttribute> map = (Map<String, TransactionAttribute>) nameMapField.get(matchTransactionAttributeSource);
//遍历nameMap
for (Map.Entry<String, TransactionAttribute> entry : map.entrySet()) {
if (!entry.getValue().isReadOnly()) { // 定义了ReadOnly的策略才加入到slaveMethodPattern
continue;
}
slaveMethodPattern.add(entry.getKey());
}
} // 切面 before方法
public void before(JoinPoint point) {
// 获取到当前执行的方法名
String methodName = point.getSignature().getName(); boolean isSlave = false; if (slaveMethodPattern.isEmpty()) {
// 没有配置read-only属性，采用方法名匹配方式
isSlave = isSlaveByMethodName(methodName);
} else {
// 配置read-only属性, 采用通配符匹配
for (String mappedName : slaveMethodPattern) {
if (isSlaveByConfigWildcard(methodName, mappedName)) {
isSlave = true;
break;
}
}
}
if (isSlave) {
// 标记为读库
DynamicDataSource.markMaster(true);
} else {
// 标记为写库
DynamicDataSource.markMaster(false);
}
} // 匹配以指定名称开头的方法名, 配置了slaveMethodStartWith属性, 或使用默认
private Boolean isSlaveByMethodName(String methodName) {
return StringUtils.startsWithAny(methodName, getSlaveMethodStartWith());
} // 匹配带通配符"xxx*", "*xxx" 和 "*xxx*"的方法名, 源自配置了readonly属性的方法名
protected boolean isSlaveByConfigWildcard(String methodName, String mappedName) {
return PatternMatchUtils.simpleMatch(mappedName, methodName);
} // 注入
public void setSlaveMethodStartWith(String[] slaveMethodStartWith) {
this.slaveMethodStartWith = slaveMethodStartWith;
} public String[] getSlaveMethodStartWith() {
if(this.slaveMethodStartWith == null){
// 没有配置slaveMethodStartWith属性，使用默认
return defaultSlaveMethodStartWith;
}
return slaveMethodStartWith;
}
}

/**
* 使用Spring的动态数据源，需要实现AbstractRoutingDataSource
* 通过determineCurrentLookupKey方法拿到识别key来判断选择读/写数据源
* token显然是多例的, 所以引入ThreadLocal保存
*/
public class DynamicDataSource extends AbstractRoutingDataSource {
// 读库总数
private Integer slaveCount;
// 读库轮询计数, 初始为-1, 本类为单例, AtomicInteger线程安全
private AtomicInteger counter = new AtomicInteger(-1);
// 存储读库的识别key sl1ve01, slave02... 写库识别key为master
private List<Object> slaveDataSources = new ArrayList<Object>(); //当前线程的写库/读库token
private static final ThreadLocal<Boolean> tokenHolder = new ThreadLocal<>(); public static void markMaster(boolean isMaster){
tokenHolder.set(isMaster);
} @Override
protected Object determineCurrentLookupKey() {
if (tokenHolder.get()) {
return "master"; // 写库
}
// 轮询读库, 得到的下标为：0、1、2...
Integer index = counter.incrementAndGet() % slaveCount;
if (counter.get() > 99999) { // 以免超出Integer范围
counter.set(-1);
}
return slaveDataSources.get(index);
} @Override
public void afterPropertiesSet() {
super.afterPropertiesSet();
// 父类的resolvedDataSources属性是private, 需要使用反射获取
Field field = ReflectionUtils.findField(AbstractRoutingDataSource.class, "resolvedDataSources");
field.setAccessible(true); // 设置可访问
try {
Map<Object, DataSource> resolvedDataSources = (Map<Object, DataSource>) field.get(this);
// 读库数等于dataSource总数减写库数
this.slaveCount = resolvedDataSources.size() - 1;
for (Map.Entry<Object, DataSource> entry : resolvedDataSources.entrySet()) {
if ("master".equals(entry.getKey())) {
continue;
}
slaveDataSources.add(entry.getKey());
}
} catch (Exception e) {
e.printStackTrace();
}
}
}

<!-- 定义事务策略 -->
<tx:advice id="txAdvice" transaction-manager="transactionManager">
<tx:attributes>
<!--所有以query开头的方法都是只读的 -->
<tx:method name="query*" read-only="true" /> <!-- readonly属性 -->
<!--其他方法使用默认事务策略 -->
<tx:method name="*" />
</tx:attributes>
</tx:advice> <!-- 定义AOP切面处理器 -->
<bean class="com.zx.DataSourceAspect" id="dataSourceAspect">
<!-- 注入事务策略 -->
<property name="txAdvice" ref="txAdvice"/>
<!-- 指定slave方法的前缀（非必须） -->
<property name="slaveMethodStartWith" value="query,find,get"/>
</bean> <aop:config>
<aop:pointcut id="myPointcut" expression="execution(* com.zx.service.*.*(..))" />
<!-- 将切面应用到自定义的切面处理器上，-9999保证该切面优先级最高执行 -->
<aop:aspect ref="dataSourceAspect" order="-9999">
<aop:before method="before" pointcut-ref="myPointcut" />
</aop:aspect>
</aop:config> <!-- 定义数据源，继承了spring的动态数据源 -->
<bean id="dataSource" class="com.zx.DynamicDataSource">
<!-- 设置多个数据源 -->
<property name="targetDataSources">
<map key-type="java.lang.String">
<!-- 这些设置的key和determineCurrentLookupKey方法拿到的key相比对, 根据匹配选择数据源 -->
<entry key="master" value-ref="masterDataSource"/> <!-- value-ref指向数据源 -->
<entry key="slave01" value-ref="slave01DataSource"/>
<entry key="slave02" value-ref="slave02DataSource"/>
<entry key="slave03" value-ref="slave03DataSource"/>
</map>
</property>
<!-- 设置默认的数据源，这里默认走写库 -->
<property name="defaultTargetDataSource" ref="masterDataSource"/>
</bean>