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

逃离我推掉我的手 2021-09-29 03:10 269阅读 0赞

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

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

## 一。背景 ##

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

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

## 二。方案 ##

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

### 2.1。应用层解决 ###

![1271038-20180526102653332-989686501.png][]

优点：

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

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

### 2.2。中间件解决 ###

![1271038-20180526102838925-630295632.png][]

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

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

相关中间件产品使用：

mysql-proxy：[http://hi.baidu.com/geshuai2008/item/0ded5389c685645f850fab07][http_hi.baidu.com_geshuai2008_item_0ded5389c685645f850fab07]

Amoeba for MySQL：[http://www.iteye.com/topic/188598][http_www.iteye.com_topic_188598]和[http://www.iteye.com/topic/1113437][http_www.iteye.com_topic_1113437]

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

### 3.1。原理 ###

![1271038-20180526103221953-578819556.png][]

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

### 3.2。代码编写 ###

DynamicDataSource

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

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 }

DynamicDataSourceHolder

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

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

![ContractedBlock.gif][]DataSourceAspect.java

### 3.3。配置2个数据源 ###

#### 3.3.1。jdbc.properties ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

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。定义连接池 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

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 
     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 
    13 <property name="idleConnectionTestPeriod" value="60" />
    14 
    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 
    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 
    34 <property name="idleConnectionTestPeriod" value="60" />
    35 
    36 <property name="idleMaxAge" value="30" />
    37 
    38 <property name="maxConnectionsPerPartition" value="150" />
    39 
    40 <property name="minConnectionsPerPartition" value="5" />
    41 </bean>

#### 3.3.3。定义DataSource ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 
     2 <bean id="dataSource" class="cn.lhx.usermanage.spring.DynamicDataSource">
     3 
     4 <property name="targetDataSources">
     5 <map key-type="java.lang.String">
     6 
     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>

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

#### 3.4.1。定义事务管理器 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 
    2 <bean id="transactionManager"
    3 class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
    4 <property name="dataSource" ref="dataSource" />
    5 </bean>

#### 3.4.2。定义事务策略 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

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>

#### 3.4.3。定义切面 ####

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 
     2 <bean class="cn.lhx.usermanage.spring.DataSourceAspect" id="dataSourceAspect" />
     3 
     4 <aop:config>
     5 
     6 <aop:pointcut id="txPointcut" expression="execution(* xx.xxx.xxxxxxx.service.*.*(..))" />
     7 
     8 <aop:advisor advice-ref="txAdvice" pointcut-ref="txPointcut"/>
     9 
    10 
    11 <aop:aspect ref="dataSourceAspect" order="-9999">
    12 <aop:before method="before" pointcut-ref="txPointcut" />
    13 </aop:aspect>
    14 </aop:config>

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

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

### 4.1。改进后的配置 ###

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 
    2 <bean class="cn.lhx.usermanage.spring.DataSourceAspectTx" id="dataSourceAspectTx">
    3 
    4 <property name="txAdvice" ref="txAdvice"/>
    5 
    6 <property name="slaveMethodStart" value="query,find,get"/>
    7 </bean>

### 4.2。改进后的实现 ###

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

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 }

## 五。一主多从的实现 ##

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

![1271038-20180526105351384-748686003.png][]

**实现没有校验：**

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

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);
    84     }
    85 }

## 六。Mysql主从复制 ##

### 6.1。原理 ###

![1271038-20180526105606860-717889292.png][]

转载于:https://www.cnblogs.com/kaixinyufeng/p/9092158.html

[1271038-20180526102653332-989686501.png]: /images/20210726/f6201c2e18434d48ab94740a5be7e311.png
[1271038-20180526102838925-630295632.png]: https://images2018.cnblogs.com/blog/1271038/201805/1271038-20180526102838925-630295632.png
[http_hi.baidu.com_geshuai2008_item_0ded5389c685645f850fab07]: http://hi.baidu.com/geshuai2008/item/0ded5389c685645f850fab07
[http_www.iteye.com_topic_188598]: http://www.iteye.com/topic/188598
[http_www.iteye.com_topic_1113437]: http://www.iteye.com/topic/1113437
[1271038-20180526103221953-578819556.png]: https://images2018.cnblogs.com/blog/1271038/201805/1271038-20180526103221953-578819556.png
[ContractedBlock.gif]: https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif
[ExpandedBlockStart.gif]: https://images.cnblogs.com/OutliningIndicators/ExpandedBlockStart.gif
[1271038-20180526105351384-748686003.png]: https://images2018.cnblogs.com/blog/1271038/201805/1271038-20180526105351384-748686003.png
[1271038-20180526105606860-717889292.png]: https://images2018.cnblogs.com/blog/1271038/201805/1271038-20180526105606860-717889292.png