Spring002--实现读写分离（Mysql实现主从复制）

Posted 2020-11-08 kaixinyufeng

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Spring002--实现读写分离（Mysql实现主从复制）相关的知识，希望对你有一定的参考价值。

Spring AOP实现读写分离（mysql实现主从复制）

本文来自于博客：http://www.cnblogs.com/bjlhx/p/8297460.html

一。背景

一般应用对数据库而言都是“读多写少”，也就说对数据库读取数据的压力比较大
主库，负责写入数据，我们称之为：写库；
从库，负责读取数据，我们称之为：读库；

1、读库和写库的数据一致；
2、写数据必须写到写库；
3、读数据必须到读库；

二。方案

解决读写分离的方案有两种：应用层解决和中间件解决。

2.1。应用层解决

优点：

1、多数据源切换方便，由程序自动完成；
2、不需要引入中间件；
3、理论上支持任何数据库；

缺点：
1、由程序员完成，运维参与不到；
2、不能做到动态增加数据源；

2.2。中间件解决

优点：
1、源程序不需要做任何改动就可以实现读写分离；
2、动态添加数据源不需要重启程序；

缺点：
1、程序依赖于中间件，会导致切换数据库变得困难；
2、由中间件做了中转代理，性能有所下降；

三。使用Spring基于应用层实现

3.1。原理

在进入Service之前，使用AOP来做出判断，是使用写库还是读库，判断依据可以根据方法名判断，比如说以query、find、get等开头的就走读库，其他的走写库。

3.2。代码编写

DynamicDataSource

 1 import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
 2 
 3 /**
 4  * 定义动态数据源，实现通过集成spring提供的AbstractRoutingDataSource，只需要实现determineCurrentLookupKey方法即可
 5  *
 6  * 由于DynamicDataSource是单例的，线程不安全的，所以采用ThreadLocal保证线程安全，由DynamicDataSourceHolder完成。
 7  * @since 2018/1/15 17:16
 8  */
 9 public class DynamicDataSource extends AbstractRoutingDataSource {
10 
11     @Override
12     protected Object determineCurrentLookupKey() {
13         // 使用DynamicDataSourceHolder保证线程安全，并且得到当前线程中的数据源key
14         return DynamicDataSourceHolder.getDataSourceKey();
15     }
16 }

View Code

DynamicDataSourceHolder

 1 /**
 2  *  使用ThreadLocal技术来记录当前线程中的数据源的key
 3  * @since 2018/1/15 17:17
 4  */
 5 public class DynamicDataSourceHolder {
 6     //写库对应的数据源key
 7     private static final String MASTER = "master";
 8 
 9     //读库对应的数据源key
10     private static final String SLAVE = "slave";
11 
12     //使用ThreadLocal记录当前线程的数据源key
13     private static final ThreadLocal<String> holder = new ThreadLocal<String>();
14 
15     /**
16      * 设置数据源key
17      * @param key
18      */
19     public static void putDataSourceKey(String key) {
20         holder.set(key);
21     }
22 
23     /**
24      * 获取数据源key
25      * @return
26      */
27     public static String getDataSourceKey() {
28         return holder.get();
29     }
30 
31     /**
32      * 标记写库
33      */
34     public static void markMaster(){
35         putDataSourceKey(MASTER);
36     }
37 
38     /**
39      * 标记读库
40      */
41     public static void markSlave(){
42         putDataSourceKey(SLAVE);
43     }
44 }

DynamicDataSourceHolder.java

DataSourceAspect

DataSourceAspect.java

3.3。配置2个数据源

3.3.1。jdbc.properties

 1 jdbc.master.driver=com.mysql.jdbc.Driver
 2 jdbc.master.url=jdbc:mysql://127.0.0.1:3306/mybatis_1128?useUnicode=true&characterEncoding=utf8&autoReconnect=true&allowMultiQueries=true
 3 jdbc.master.username=root
 4 jdbc.master.password=123456
 5 
 6 
 7 jdbc.slave01.driver=com.mysql.jdbc.Driver
 8 jdbc.slave01.url=jdbc:mysql://127.0.0.1:3307/mybatis_1128?useUnicode=true&characterEncoding=utf8&autoReconnect=true&allowMultiQueries=true
 9 jdbc.slave01.username=root
10 jdbc.slave01.password=123456

jdbc.properties

3.3.2。定义连接池

 1 <!-- 配置连接池 -->
 2 <bean id="masterDataSource" class="com.mchange.v2.c3p0.ComboPooledDataSource"
 3 destroy-method="close">
 4 <!-- 数据库驱动 -->
 5 <property name="driverClass" value="${jdbc.master.driver}" />
 6 <!-- 相应驱动的jdbcUrl -->
 7 <property name="jdbcUrl" value="${jdbc.master.url}" />
 8 <!-- 数据库的用户名 -->
 9 <property name="username" value="${jdbc.master.username}" />
10 <!-- 数据库的密码 -->
11 <property name="password" value="${jdbc.master.password}" />
12 <!-- 检查数据库连接池中空闲连接的间隔时间，单位是分，默认值：240，如果要取消则设置为0 -->
13 <property name="idleConnectionTestPeriod" value="60" />
14 <!-- 连接池中未使用的链接较大存活时间，单位是分，默认值：60，如果要永远存活设置为0 -->
15 <property name="idleMaxAge" value="30" />
16 <!-- 每个分区较大的连接数 -->
17 <property name="maxConnectionsPerPartition" value="150" />
18 <!-- 每个分区最小的连接数 -->
19 <property name="minConnectionsPerPartition" value="5" />
20 </bean>
21 
22 <!-- 配置连接池 -->
23 <bean id="slave01DataSource" class="com.mchange.v2.c3p0.ComboPooledDataSource"
24 destroy-method="close">
25 <!-- 数据库驱动 -->
26 <property name="driverClass" value="${jdbc.slave01.driver}" />
27 <!-- 相应驱动的jdbcUrl -->
28 <property name="jdbcUrl" value="${jdbc.slave01.url}" />
29 <!-- 数据库的用户名 -->
30 <property name="username" value="${jdbc.slave01.username}" />
31 <!-- 数据库的密码 -->
32 <property name="password" value="${jdbc.slave01.password}" />
33 <!-- 检查数据库连接池中空闲连接的间隔时间，单位是分，默认值：240，如果要取消则设置为0 -->
34 <property name="idleConnectionTestPeriod" value="60" />
35 <!-- 连接池中未使用的链接较大存活时间，单位是分，默认值：60，如果要永远存活设置为0 -->
36 <property name="idleMaxAge" value="30" />
37 <!-- 每个分区较大的连接数 -->
38 <property name="maxConnectionsPerPartition" value="150" />
39 <!-- 每个分区最小的连接数 -->
40 <property name="minConnectionsPerPartition" value="5" />
41 </bean>

View Code

3.3.3。定义DataSource

 1 <!-- 定义数据源，使用自己实现的数据源 -->
 2 <bean id="dataSource" class="cn.lhx.usermanage.spring.DynamicDataSource">
 3 <!-- 设置多个数据源 -->
 4 <property name="targetDataSources">
 5 <map key-type="java.lang.String">
 6 <!-- 这个key需要和程序中的key一致 -->
 7 <entry key="master" value-ref="masterDataSource"/>
 8 <entry key="slave" value-ref="slave01DataSource"/>
 9 </map>
10 </property>
11 <!-- 设置默认的数据源，这里默认走写库 -->
12 <property name="defaultTargetDataSource" ref="masterDataSource"/>
13 </bean>

View Code

3.4。配置事务管理以及动态切换数据源切面

3.4.1。定义事务管理器

1 <!-- 定义事务管理器 -->
2 <bean id="transactionManager"
3 class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
4 <property name="dataSource" ref="dataSource" />
5 </bean>

View Code

3.4.2。定义事务策略

 1 <!-- 定义事务策略 -->
 2 <tx:advice id="txAdvice" transaction-manager="transactionManager">
 3 <tx:attributes>
 4 <!--定义查询方法都是只读的 -->
 5 <tx:method name="query*" read-only="true" />
 6 <tx:method name="find*" read-only="true" />
 7 <tx:method name="get*" read-only="true" />
 8 
 9 <!-- 主库执行操作，事务传播行为定义为默认行为 -->
10 <tx:method name="save*" propagation="REQUIRED" />
11 <tx:method name="update*" propagation="REQUIRED" />
12 <tx:method name="delete*" propagation="REQUIRED" />
13 
14 <!--其他方法使用默认事务策略 -->
15 <tx:method name="*" />
16 </tx:attributes>
17 </tx:advice>

View Code

3.4.3。定义切面

 1 <!-- 定义AOP切面处理器 -->
 2 <bean class="cn.lhx.usermanage.spring.DataSourceAspect" id="dataSourceAspect" />
 3 
 4 <aop:config>
 5 <!-- 定义切面，所有的service的所有方法 -->
 6 <aop:pointcut id="txPointcut" expression="execution(* xx.xxx.xxxxxxx.service.*.*(..))" />
 7 <!-- 应用事务策略到Service切面 -->
 8 <aop:advisor advice-ref="txAdvice" pointcut-ref="txPointcut"/>
 9 
10 <!-- 将切面应用到自定义的切面处理器上，-9999保证该切面优先级较高执行 -->
11 <aop:aspect ref="dataSourceAspect" order="-9999">
12 <aop:before method="before" pointcut-ref="txPointcut" />
13 </aop:aspect>
14 </aop:config>

View Code

四。改进切面实现，使用事务策略规则匹配

之前的实现我们是将通过方法名匹配，而不是使用事务策略中的定义，我们使用事务管理策略中的规则匹配。

4.1。改进后的配置

1 <!-- 定义AOP切面处理器 -->
2 <bean class="cn.lhx.usermanage.spring.DataSourceAspectTx" id="dataSourceAspectTx">
3 <!-- 指定事务策略 -->
4 <property name="txAdvice" ref="txAdvice"/>
5 <!-- 指定slave方法的前缀（非必须） -->
6 <property name="slaveMethodStart" value="query,find,get"/>
7 </bean>

View Code

4.2。改进后的实现

  1 import org.apache.commons.lang3.StringUtils;
  2 import org.aspectj.lang.JoinPoint;
  3 import org.springframework.transaction.interceptor.NameMatchTransactionAttributeSource;
  4 import org.springframework.transaction.interceptor.TransactionAttribute;
  5 import org.springframework.transaction.interceptor.TransactionAttributeSource;
  6 import org.springframework.transaction.interceptor.TransactionInterceptor;
  7 import org.springframework.util.PatternMatchUtils;
  8 import org.springframework.util.ReflectionUtils;
  9 
 10 import java.lang.reflect.Field;
 11 import java.util.ArrayList;
 12 import java.util.List;
 13 import java.util.Map;
 14 
 15 /**
 16  * @since 2018/1/15 18:03
 17  */
 18 public class DataSourceAspectTx {
 19     private List<String> slaveMethodPattern = new ArrayList<String>();
 20 
 21     private static final String[] defaultSlaveMethodStart = new String[]{ "query", "find", "get" };
 22 
 23     private String[] slaveMethodStart;
 24 
 25     /**
 26      * 读取事务管理中的策略
 27      *
 28      * @param txAdvice
 29      * @throws Exception
 30      */
 31     @SuppressWarnings("unchecked")
 32     public void setTxAdvice(TransactionInterceptor txAdvice) throws Exception {
 33         if (txAdvice == null) {
 34             // 没有配置事务管理策略
 35             return;
 36         }
 37         //从txAdvice获取到策略配置信息
 38         TransactionAttributeSource transactionAttributeSource = txAdvice.getTransactionAttributeSource();
 39         if (!(transactionAttributeSource instanceof NameMatchTransactionAttributeSource)) {
 40             return;
 41         }
 42         //使用反射技术获取到NameMatchTransactionAttributeSource对象中的nameMap属性值
 43         NameMatchTransactionAttributeSource matchTransactionAttributeSource = (NameMatchTransactionAttributeSource) transactionAttributeSource;
 44         Field nameMapField = ReflectionUtils.findField(NameMatchTransactionAttributeSource.class, "nameMap");
 45         nameMapField.setAccessible(true); //设置该字段可访问
 46         //获取nameMap的值
 47         Map<String, TransactionAttribute> map = (Map<String, TransactionAttribute>) nameMapField.get(matchTransactionAttributeSource);
 48 
 49         //遍历nameMap
 50         for (Map.Entry<String, TransactionAttribute> entry : map.entrySet()) {
 51             if (!entry.getValue().isReadOnly()) {//判断之后定义了ReadOnly的策略才加入到slaveMethodPattern
 52                 continue;
 53             }
 54             slaveMethodPattern.add(entry.getKey());
 55         }
 56     }
 57 
 58     /**
 59      * 在进入Service方法之前执行
 60      *
 61      * @param point 切面对象
 62      */
 63     public void before(JoinPoint point) {
 64         // 获取到当前执行的方法名
 65         String methodName = point.getSignature().getName();
 66 
 67         boolean isSlave = false;
 68 
 69         if (slaveMethodPattern.isEmpty()) {
 70             // 当前Spring容器中没有配置事务策略，采用方法名匹配方式
 71             isSlave = isSlave(methodName);
 72         } else {
 73             // 使用策略规则匹配
 74             for (String mappedName : slaveMethodPattern) {
 75                 if (isMatch(methodName, mappedName)) {
 76                     isSlave = true;
 77                     break;
 78                 }
 79             }
 80         }
 81 
 82         if (isSlave) {
 83             // 标记为读库
 84             DynamicDataSourceHolder.markSlave();
 85         } else {
 86             // 标记为写库
 87             DynamicDataSourceHolder.markMaster();
 88         }
 89     }
 90 
 91     /**
 92      * 判断是否为读库
 93      *
 94      * @param methodName
 95      * @return
 96      */
 97     private Boolean isSlave(String methodName) {
 98         // 方法名以query、find、get开头的方法名走从库
 99         return StringUtils.startsWithAny(methodName, getSlaveMethodStart());
100     }
101 
102     /**
103      * 通配符匹配
104      *
105      * Return if the given method name matches the mapped name.
106      * <p>
107      * The default implementation checks for "xxx*", "*xxx" and "*xxx*" matches, as well as direct
108      * equality. Can be overridden in subclasses.
109      *
110      * @param methodName the method name of the class
111      * @param mappedName the name in the descriptor
112      * @return if the names match
113      * @see org.springframework.util.PatternMatchUtils#simpleMatch(String, String)
114      */
115     protected boolean isMatch(String methodName, String mappedName) {
116         return PatternMatchUtils.simpleMatch(mappedName, methodName);
117     }
118 
119     /**
120      * 用户指定slave的方法名前缀
121      * @param slaveMethodStart
122      */
123     public void setSlaveMethodStart(String[] slaveMethodStart) {
124         this.slaveMethodStart = slaveMethodStart;
125     }
126 
127     public String[] getSlaveMethodStart() {
128         if(this.slaveMethodStart == null){
129             // 没有指定，使用默认
130             return defaultSlaveMethodStart;
131         }
132         return slaveMethodStart;
133     }
134 }

View Code

五。一主多从的实现

很多实际使用场景下都是采用“一主多从”的架构的，所有我们现在对这种架构做支持，目前只需要修改DynamicDataSource即可。

实现没有校验：

 1 package com.jd.ofc.trace.common.datasource;
 2 
 3 import org.slf4j.Logger;
 4 import org.slf4j.LoggerFactory;
 5 import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
 6 import org.springframework.util.ReflectionUtils;
 7 
 8 import javax.sql.DataSource;
 9 import java.lang.reflect.Field;
10 import java.util.ArrayList;
11 import java.util.List;
12 import java.util.Map;
13 import java.util.concurrent.atomic.AtomicInteger;
14 
15 /**
16  * @author lihongxu6
17  * @since 2018/1/16 15:07
18  */
19 public class DynamicDataSourceMutliSlave extends AbstractRoutingDataSource {
20     private static final Logger LOGGER = LoggerFactory.getLogger(DynamicDataSource.class);
21 
22     private Integer slaveCount;
23 
24     // 轮询计数,初始为-1,AtomicInteger是线程安全的
25     private AtomicInteger counter = new AtomicInteger(-1);
26 
27     // 记录读库的key
28     private List<Object> slaveDataSources = new ArrayList<Object>(0);
29 
30     @Override
31     protected Object determineCurrentLookupKey() {
32         // 使用DynamicDataSourceHolder保证线程安全，并且得到当前线程中的数据源key
33         if (DynamicDataSourceHolder.isMaster()) {
34             Object key = DynamicDataSourceHolder.getDataSourceKey();
35             if (LOGGER.isDebugEnabled()) {
36                 LOGGER.debug("当前DataSource的key为: " + key);
37             }
38             return key;
39         }
40         Object key = getSlaveKey();
41         if (LOGGER.isDebugEnabled()) {
42             LOGGER.debug("当前DataSource的key为: " + key);
43         }
44         return key;
45 
46     }
47 
48     @SuppressWarnings("unchecked")
49     @Override
50     public void afterPropertiesSet() {
51         super.afterPropertiesSet();
52 
53         // 由于父类的resolvedDataSources属性是私有的子类获取不到，需要使用反射获取
54         Field field = ReflectionUtils.findField(AbstractRoutingDataSource.class, "resolvedDataSources");
55         field.setAccessible(true); // 设置可访问
56 
57         try {
58             Map<Object, DataSource> resolvedDataSources = (Map<Object, DataSource>) field.get(this);
59             // 读库的数据量等于数据源总数减去写库的数量
60             this.slaveCount = resolvedDataSources.size() - 1;
61             for (Map.Entry<Object, DataSource> entry : resolvedDataSources.entrySet()) {
62                 if (DynamicDataSourceHolder.MASTER.equals(entry.getKey())) {
63                     continue;
64                 }
65                 slaveDataSources.add(entry.getKey());
66             }
67         } catch (Exception e) {
68             LOGGER.error("afterPropertiesSet error! ", e);
69         }
70     }
71 
72     /**
73      * 轮询算法实现
74      *
75      * @return
76      */
77     public Object getSlaveKey() {
78         // 得到的下标为：0、1、2、3……
79         Integer index = counter.incrementAndGet() % slaveCount;
80         if (counter.get() > 9999) { // 以免超出Integer范围
81             counter.set(-1); // 还原
82         }
83         return slaveDataSources.get(index);以上是关于Spring002--实现读写分离（Mysql实现主从复制）的主要内容，如果未能解决你的问题，请参考以下文章 
 Mybatis+Spring实现Mysql读写分离
 使用Spring AOP实现MySQL读写分离
 使用Spring实现读写分离（ MySQL实现主从复制）
 使用Spring实现MySQL读写分离
 Spring AOP 实现读写分离
 使用Spring+MySql实现读写分离主从复制