Spring 如何实现多数据源读写分离？

秒速五厘米 2021-09-03 04:30 410阅读 0赞

点击上方 [Java后端][Java]，选择 设为星标

优质文章，及时送达

--------------------

很多小伙伴私聊我说：最近他们公司的业务涉及到多个数据源的问题，问我 Spring 如何实现多数据源的问题。回答这个问题之前，首先需要弄懂什么是多数据源：多数据源就是在同一个项目中，会连接两个甚至多个数据存储，这里的数据存储可以是关系型数据库（比如：MySQL、SQL Server、Oracle），也可以非关系型数据库，比如：HBase、MongoDB、ES等。

那么，问题来了，Spring能够实现多数据源吗？并且还要实现读者分离？答案是：必须的，这么强大的Spring，肯定能实现啊！别急，我们就一点点剖析、解决这些问题！

## **背景** ##

我们一般应用对数据库而言都是“读多写少”，也就说对数据库读取数据的压力比较大，有一个思路就是说采用数据库集群的方案，其中一个是主库，负责写入数据，我们称之为：写库；其它都是从库，负责读取数据，我们称之为：读库；

那么，对我们的要求是：

*  读库和写库的数据一致；
 *  写数据必须写到写库；
 *  读数据必须到读库；

## **方案** ##

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

### 应用层解决 ###

![format_png][]

优点：

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

缺点：

*  由程序员完成，运维参与不到；
 *  不能做到动态增加数据源；

### 中间件解决 ###

![format_png 1][]

优点：

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

缺点：

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

## **Spring方案** ##

### 原理 ###

![format_png 2][]

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

### **DynamicDataSource**    ###

import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
    /**
     * 定义动态数据源，实现通过集成Spring提供的AbstractRoutingDataSource，只需要实现determineCurrentLookupKey方法即可
     * 由于DynamicDataSource是单例的，线程不安全的，所以采用ThreadLocal保证线程安全，由DynamicDataSourceHolder完成。
     * @author binghe
     */
    public class DynamicDataSource extends AbstractRoutingDataSource{
        @Override
        protected Object determineCurrentLookupKey() {
            // 使用DynamicDataSourceHolder保证线程安全，并且得到当前线程中的数据源key
            return DynamicDataSourceHolder.getDataSourceKey();
        }
    }

**DynamicDataSourceHolder**

/**
     * 使用ThreadLocal技术来记录当前线程中的数据源的key
     * @author binghe
     */
    public class DynamicDataSourceHolder {
        //写库对应的数据源key
        private static final String MASTER = "master";
        //读库对应的数据源key
        private static final String SLAVE = "slave";
        //使用ThreadLocal记录当前线程的数据源key
        private static final ThreadLocal<String> holder = new ThreadLocal<String>();
        /**
         * 设置数据源key
         * @param key
         */
        public static void putDataSourceKey(String key) {
            holder.set(key);
        }
        /**
         * 获取数据源key
         * @return
         */
        public static String getDataSourceKey() {
            return holder.get();
        }
        /**
         * 标记写库
         */
        public static void markMaster(){
            putDataSourceKey(MASTER);
        }
        /**
         * 标记读库
         */
        public static void markSlave(){
            putDataSourceKey(SLAVE);
        }
    }

### **DataSourceAspect**    ###

import org.apache.commons.lang3.StringUtils;
    import org.aspectj.lang.JoinPoint;
    /**
     * 定义数据源的AOP切面，通过该Service的方法名判断是应该走读库还是写库
     * @author binghe
     */
    public class DataSourceAspect {
        /**
         * 在进入Service方法之前执行
         * @param point 切面对象
         */
        public void before(JoinPoint point) {
            // 获取到当前执行的方法名
            String methodName = point.getSignature().getName();
            if (isSlave(methodName)) {
                // 标记为读库
                DynamicDataSourceHolder.markSlave();
            } else {
                // 标记为写库
                DynamicDataSourceHolder.markMaster();
            }
        }
        /**
         * 判断是否为读库
         *
         * @param methodName
         * @return
         */
        private Boolean isSlave(String methodName) {
            // 方法名以query、find、get开头的方法名走从库
            return StringUtils.startsWithAny(methodName, "query", "find", "get");
        }
    }

### **配置2个数据源** ###

###     ###

jdbc.master.driver=com.mysql.jdbc.Driver
    jdbc.master.url=jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=utf8&autoReconnect=true&allowMultiQueries=true
    jdbc.master.username=root
    jdbc.master.password=123456
    jdbc.slave01.driver=com.mysql.jdbc.Driver
    jdbc.slave01.url=jdbc:mysql://127.0.0.1:3307/test?useUnicode=true&characterEncoding=utf8&autoReconnect=true&allowMultiQueries=true
    jdbc.slave01.username=root
    jdbc.slave01.password=123456

### **定义连接池** ###

<bean id="masterDataSource" class="com.jolbox.bonecp.BoneCPDataSource"
      destroy-method="close">
      
      <property name="driverClass" value="${jdbc.master.driver}" />
      
      <property name="jdbcUrl" value="${jdbc.master.url}" />
      
      <property name="username" value="${jdbc.master.username}" />
      
      <property name="password" value="${jdbc.master.password}" />
      
      <property name="idleConnectionTestPeriod" value="60" />
      
      <property name="idleMaxAge" value="30" />
      
      <property name="maxConnectionsPerPartition" value="150" />
      
      <property name="minConnectionsPerPartition" value="5" />
    </bean>
    
    <bean id="slave01DataSource" class="com.jolbox.bonecp.BoneCPDataSource"
      destroy-method="close">
      
      <property name="driverClass" value="${jdbc.slave01.driver}" />
      
      <property name="jdbcUrl" value="${jdbc.slave01.url}" />
      
      <property name="username" value="${jdbc.slave01.username}" />
      
      <property name="password" value="${jdbc.slave01.password}" />
      
      <property name="idleConnectionTestPeriod" value="60" />
      
      <property name="idleMaxAge" value="30" />
      
      <property name="maxConnectionsPerPartition" value="150" />
      
      <property name="minConnectionsPerPartition" value="5" />
    </bean>

### **定义DataSource** ###

###     ###

<bean id="dataSource" class="cn.itcast.usermanage.spring.DynamicDataSource">
      
      <property name="targetDataSources">
        <map key-type="java.lang.String">
          
          <entry key="master" value-ref="masterDataSource"/>
          <entry key="slave" value-ref="slave01DataSource"/>
        </map>
      </property>
      
      <property name="defaultTargetDataSource" ref="masterDataSource"/>
    </bean>

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

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

<bean id="transactionManager"
      class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
      <property name="dataSource" ref="dataSource" />
    </bean>

### 定义事务策略 ###

<tx:advice id="txAdvice" transaction-manager="transactionManager">
      <tx:attributes>
        
        <tx:method name="query*" read-only="true" />
        <tx:method name="find*" read-only="true" />
        <tx:method name="get*" read-only="true" />
        
        <tx:method name="save*" propagation="REQUIRED" />
        <tx:method name="update*" propagation="REQUIRED" />
        <tx:method name="delete*" propagation="REQUIRED" />
        
        <tx:method name="*" />
      </tx:attributes>
    </tx:advice>

### 定义切面 ###

###     ###

<bean class="cn.itcast.usermanage.spring.DataSourceAspect" id="dataSourceAspect" />
    <aop:config>
      
      <aop:pointcut id="txPointcut" expression="execution(* xx.xxx.xxxxxxx.service.*.*(..))" />
      
      <aop:advisor advice-ref="txAdvice" pointcut-ref="txPointcut"/>
      
      <aop:aspect ref="dataSourceAspect" order="-9999">
        <aop:before method="before" pointcut-ref="txPointcut" />
      </aop:aspect>
    </aop:config>

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

### 改进后的配置 ###

<bean class="cn.itcast.usermanage.spring.DataSourceAspect" id="dataSourceAspect">
      
      <property name="txAdvice" ref="txAdvice"/>
      
      <property name="slaveMethodStart" value="query,find,get"/>
    </bean>

### 改进后的实现 ###

###     ###

/**
     * 定义数据源的AOP切面，该类控制了使用Master还是Slave。
     * 如果事务管理中配置了事务策略，则采用配置的事务策略中的标记了ReadOnly的方法是用Slave，其它使用Master。
     * 如果没有配置事务管理的策略，则采用方法名匹配的原则，以query、find、get开头方法用Slave，其它用Master。
     * @author binghe
     *
     */
    public class DataSourceAspect {
        private List<String> slaveMethodPattern = new ArrayList<String>();
        private static final String[] defaultSlaveMethodStart = new String[]{ "query", "find", "get" };
        private String[] slaveMethodStart;
        /**
         * 读取事务管理中的策略
         * @param txAdvice
         * @throws Exception
         */
        @SuppressWarnings("unchecked")
        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());
            }
        }
        /**
         * 在进入Service方法之前执行
         *
         * @param point 切面对象
         */
        public void before(JoinPoint point) {
            // 获取到当前执行的方法名
            String methodName = point.getSignature().getName();
            boolean isSlave = false;
            if (slaveMethodPattern.isEmpty()) {
                // 当前Spring容器中没有配置事务策略，采用方法名匹配方式
                isSlave = isSlave(methodName);
            } else {
                // 使用策略规则匹配
                for (String mappedName : slaveMethodPattern) {
                    if (isMatch(methodName, mappedName)) {
                        isSlave = true;
                        break;
                    }
                }
            }
            if (isSlave) {
                // 标记为读库
                DynamicDataSourceHolder.markSlave();
            } else {
                // 标记为写库
                DynamicDataSourceHolder.markMaster();
            }
        }
        /**
         * 判断是否为读库
         *
         * @param methodName
         * @return
         */
        private Boolean isSlave(String methodName) {
            // 方法名以query、find、get开头的方法名走从库
            return StringUtils.startsWithAny(methodName, getSlaveMethodStart());
        }
        /**
         * 通配符匹配
         *
         * Return if the given method name matches the mapped name.
         * <p>
         * The default implementation checks for "xxx*", "*xxx" and "*xxx*" matches, as well as direct
         * equality. Can be overridden in subclasses.
         *
         * @param methodName the method name of the class
         * @param mappedName the name in the descriptor
         * @return if the names match
         * @see org.springframework.util.PatternMatchUtils#simpleMatch(String, String)
         */
        protected boolean isMatch(String methodName, String mappedName) {
            return PatternMatchUtils.simpleMatch(mappedName, methodName);
        }
        /**
         * 用户指定slave的方法名前缀
         * @param slaveMethodStart
         */
        public void setSlaveMethodStart(String[] slaveMethodStart) {
            this.slaveMethodStart = slaveMethodStart;
        }
        public String[] getSlaveMethodStart() {
            if(this.slaveMethodStart == null){
                // 没有指定，使用默认
                return defaultSlaveMethodStart;
            }
            return slaveMethodStart;
        }
    }

## **一主多从的实现**    ##

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

![format_png 3][]

## **实现** ##

##     ##

/**
     * 定义动态数据源，实现通过集成Spring提供的AbstractRoutingDataSource，只需要实现determineCurrentLookupKey方法即可
     * 由于DynamicDataSource是单例的，线程不安全的，所以采用ThreadLocal保证线程安全，由DynamicDataSourceHolder完成。
     * @author binghe
     *
     */
    public class DynamicDataSource extends AbstractRoutingDataSource {
        private static final Logger LOGGER = LoggerFactory.getLogger(DynamicDataSource.class);
        private Integer slaveCount;
        // 轮询计数,初始为-1,AtomicInteger是线程安全的
        private AtomicInteger counter = new AtomicInteger(-1);
        // 记录读库的key
        private List<Object> slaveDataSources = new ArrayList<Object>(0);
        @Override
        protected Object determineCurrentLookupKey() {
            // 使用DynamicDataSourceHolder保证线程安全，并且得到当前线程中的数据源key
            if (DynamicDataSourceHolder.isMaster()) {
                Object key = DynamicDataSourceHolder.getDataSourceKey();
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("当前DataSource的key为: " + key);
                }
                return key;
            }
            Object key = getSlaveKey();
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("当前DataSource的key为: " + key);
            }
            return key;
        }
        @SuppressWarnings("unchecked")
        @Override
        public void afterPropertiesSet() {
            super.afterPropertiesSet();
            // 由于父类的resolvedDataSources属性是私有的子类获取不到，需要使用反射获取
            Field field = ReflectionUtils.findField(AbstractRoutingDataSource.class, "resolvedDataSources");
            field.setAccessible(true); // 设置可访问
            try {
                Map<Object, DataSource> resolvedDataSources = (Map<Object, DataSource>) field.get(this);
                // 读库的数据量等于数据源总数减去写库的数量
                this.slaveCount = resolvedDataSources.size() - 1;
                for (Map.Entry<Object, DataSource> entry : resolvedDataSources.entrySet()) {
                    if (DynamicDataSourceHolder.MASTER.equals(entry.getKey())) {
                        continue;
                    }
                    slaveDataSources.add(entry.getKey());
                }
            } catch (Exception e) {
                LOGGER.error("afterPropertiesSet error! ", e);
            }
        }
        /**
         * 轮询算法实现
         *
         * @return
         */
        public Object getSlaveKey() {
            // 得到的下标为：0、1、2、3……
            Integer index = counter.incrementAndGet() % slaveCount;
            if (counter.get() > 9999) { // 以免超出Integer范围
                counter.set(-1); // 还原
            }
            return slaveDataSources.get(index);
        }
    }

## **MySQL主从复制**    ##

### 原理 ###

![format_png 4][]

MySQL主(master)从(slave)复制的原理：

*  master将数据改变记录到二进制日志(binarylog)中,也即是配置文件log-bin指定的文件(这些记录叫做二进制日志事件，binary log events)
 *  slave将master的binary logevents拷贝到它的中继日志(relay log)
 *  slave重做中继日志中的事件,将改变反映它自己的数据(数据重演)

### 主从配置需要注意的地方 ###

*  主DB server和从DB server数据库的版本一致
 *  主DB server和从DB server数据库数据一致\[ 这里就会可以把主的备份在从上还原，也可以直接将主的数据目录拷贝到从的相应数据目录\]
 *  主DB server开启二进制日志,主DB server和从DB server的server\_id都必须唯一

### 主库配置（windows，Linux下也类似） ###

在my.ini修改：

#开启主从复制，主库的配置
    log-bin = mysql3306-bin
    #指定主库serverid
    server-id=101
    #指定同步的数据库，如果不指定则同步全部数据库
    binlog-do-db=mybatis_1128
    执行SQL语句查询状态：
    SHOW MASTER STATUS

![format_png 5][]

需要记录下Position值，需要在从库中设置同步起始值。

### **在主库创建同步用户** ###

#授权用户slave01使用123456密码登录mysql
    grant replication slave on *.* to 'slave01'@'127.0.0.1' identified by '123456';
    flush privileges;

### 从库配置 ###

在my.ini修改

#指定serverid，只要不重复即可，从库也只有这一个配置，其他都在SQL语句中操作
    server-id=102

接下来，从从库命令行执行如下SQL语句。

CHANGE MASTER TO
     master_host='127.0.0.1',
     master_user='slave01',
     master_password='123456',
     master_port=3306,
     master_log_file='mysql3306-bin.000006',
     master_log_pos=1120;
    #启动slave同步
    START SLAVE;
    #查看同步状态
    SHOW SLAVE STATUS;

![format_png 6][]

作者：冰河团队

链接：www.bbsmax.com/A/LPdonDwN53/

Java后端交流群已成立
    公众号运营至今，离不开小伙伴们的支持。为了给小伙伴们提供一个互相交流的平台，特地开通了官方交流群。扫描下方二维码备注 进群 或者关注公众号 Java后端 后获取进群通道。

推荐阅读 1. GET 和 POST 两种基本请求方法的区别2. 牛逼！IntelliJ IDEA 从入门到上瘾！3. REST API URI 设计的七准则4. Java 项目实战天天酷跑5. 基于 Spring Boot 的 QQ 登陆实战

[Java]: http://mp.weixin.qq.com/s?__biz=Mzg2MjEwMjI1Mg%3D%3D&chksm=ce0da17bf97a286df119047a25b7de51621fe542816673bb4aa3ebd8ab97b09a86ed7db2e2ee&idx=3&mid=2247489272&scene=21&sn=ce74938072fc73a808d83ccb474db46d#wechat_redirect
[format_png]: /images/20210730/6b9e37501f7847cca5c73e88eafc2e55.png
[format_png 1]: /images/20210730/7f9bc998336840b18fae2ee915715d4d.png
[format_png 2]: /images/20210730/ed204a27390e47cb9db6afe18ffd96b4.png
[format_png 3]: /images/20210730/0a6dc1e636f44cb496ba8f009256e63c.png
[format_png 4]: /images/20210730/443a6f6b995b47c39cf0a34761ad66e3.png
[format_png 5]: /images/20210730/66f8e23f58d64cbc8900df25c3f0e838.png
[format_png 6]: /images/20210730/dc0817c5ea33491fa3c9dde9c2d6c093.png