Spring整合Mybatis下多数据源的读写分离实现(一主多仆，配合事务)

你的名字 2022-06-01 04:42 97阅读 0赞

为了提高并发量，降低数据库访问压力，配置多个数据库，一主多仆，实现写入主库，读取从库，主从间复制。  
当然数据库之间的复制实现不是我们关注的重点，我们关注的是在应用代码层面如何实现读写分离，以及确保读写分离的准确性。

常用的确保数据准确性方式的自然是事务，Spring里也是如此。在单个数据库的情形下使用事务很简，那么如果是由多个数据库的情况下是否有什么变化？在说明这个问题之前我先讲下事务的实现原理。

我们使用事务很多时候是看中其一致性的特性。当一个Service方法里只有一次对数据库的操作时，要么操作成功要么操作失败，这种情形下事务没有发挥什么作用。当Service方法里有多次对数据库的操作时，此时事务就很有必要性了。比如两次写操作，如果一次成功一次失败，他们原本应该是一个原子操作，要么一起成功要么一起失败，如果使用了事务就可以实现回滚；如果一次读一次写操作，原先是希望先看看要插入的数据存不存在，读取了结果是不存在的，那么在插入，发现插入失败，因为其他用户在你读取了之后插入前已经插入了相同的数据，这种情况下如果配置了事务，也是可以避免的，在读取要插入的数据后，给这个数据区间加个锁(Mysql用间隙锁机制)，可以在事务提交前防止其他用户插入此区间内的数据。

事务是如何让多次数据库操作一起成功或一起回滚的？**主要是事务里会绑定一个Connection，也就是数据库连接对象，多次数据库操作共用同一个Connection对象**，Connection的提交或者回滚就能将其操作的所有记录统一提交会回滚，这一点就是关键。我们打开数据库操作软件，比如navicat，打开一个操作窗口，其实就是打开了一个Connection，只不过这个Connection是默认提交的，也就是说执行完一个sql就自动提交，也就是事务自动开启和提交，可以设置为非自动提交。如果你在当前开启事务，窗口插入一条数据，但是不提交，再打开另一个窗口，也插入这条数据，会发现另一个窗口一直在执行中，当第一个窗口提交后，第二个 窗口才会报错。

Spring事务是如何让多个数据库操作共用同一个Connection的呢？当事务配置在Servic层时，Spring在开启一个事务时，就预先从配置的数据源DataSource里生成一个Connection，然后存入一个ThreadLocal类型的变量里，放在 “TransactionSynchronizationManager” 的 “resources” 变量里，key就是DataSource，这样Mybatis从SqlSession在操作数据库时，如果当前操作是配置了事务，也会用当前DataSource从此ThreadLocal的变量里获取Connection，具体代码可以查看**SpringManagedTransaction**的**getConnection**方法。

当然以上情形在单数据源的情况下自然不会有什么问题，获取Connection的key DataSource不变，每次获取到的都是事务里的Connection，使用的都是这个Connection，可以共同提交或回滚。那么如果有多个数据源怎么办 ？ 如果两个数据源各自生成一个Connection，读从库，写主库，如果不配置事务是好实现的，但在事务中肯定不适用，两个Connection肯定实现不了操作的统一提交和回滚，也不能保证操作的原子性；其次事务在提交或回滚时会发现ThreadLocal内的变量已经被改变，会抛出异常的。

当存在多个数据源的情况下，**在事务内要确保Connection不会改变，也就是说这个事务内连接的库不变**。当有读有写的Service方法，那么必须要指定其连接的是主库，这样才能确保事务的原子性和一致性。

事务在执行Service方法前会从数据源DataSource内生成一个Connection，存入ThreadLocal类型变量中，也就是说我们**要在Service方法执行前就确定这个Service方法是连接主库还是从库**，也就是说我们要在方法上或方法之外能对当前Service方法做一个识别，我这里使用的是方法的名称标识，当方法名称含有写操作的关键字，比如”insert”, “create”, “delete”, “remove”, “modify”, “update”, “change”等时，指定其内的操作连接主库，否则一律连接从库。

因为需要切换数据源，所以使用的是**RoutingDataSource**，因为需要在事务开始初始化Connection时就能从方法信息上提取连接库标识，所以对事务管理器**PlatformTransactionManager**做了自定义拓展，同时对事务的配置形式做了自定义形式。

先看看我的实现的类结构：

![这里写图片描述][SouthEast]

先看数据库相关配置类DataAccessContextConfig：

import java.util.Map;
    
    import javax.sql.DataSource;
    
    import com.bob.config.root.mapper.BaseMapper;
    import com.bob.config.root.mybatis.readsepwrite.DataSourceTransactionManagerAdapter;
    import com.bob.config.root.mybatis.readsepwrite.DynamicDataSource;
    import org.apache.commons.dbcp2.BasicDataSource;
    import org.apache.ibatis.session.ExecutorType;
    import org.apache.ibatis.session.SqlSessionFactory;
    import org.mybatis.spring.SqlSessionFactoryBean;
    import org.mybatis.spring.annotation.MapperScan;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
    import org.springframework.jdbc.datasource.DataSourceTransactionManager;
    
    /** * 数据库配置类 * * @version $Id$ * @since 2016年12月5日 下午5:24:24 */
    @Configuration
    @MapperScan(value = "com.bob.mvc.mapper", markerInterface = BaseMapper.class)
    public class DataAccessContextConfig { 
    
        private static final String DRIVER_CLASS_NAME = "com.mysql.jdbc.Driver";
        private static final String USERNAME = "root";
        private static final String PASSWORD = "lanboal";
    
        /** * MySQL的JDBC URL编写方式：jdbc:mysql://主机名称：连接端口/数据库的名称?参数=值 * 避免中文乱码要指定useUnicode和characterEncoding */
        private static final String READ_URL = "jdbc:mysql://localhost:3306/demo?useUnicode=true&characterEncoding=UTF8&useSSL=false";
        private static final String WRITE_URL = "jdbc:mysql://localhost:3306/project?useUnicode=true&characterEncoding=UTF8&useSSL=false";
    
        /** * 读数据源 * * @return */
        @Bean(destroyMethod = "close")
        public DataSource readDataSource() {
            return generateDataSource(READ_URL);
        }
    
        /** * 写数据源 * * @return */
        @Bean(destroyMethod = "close")
        public DataSource writeDataSource() {
            return generateDataSource(WRITE_URL);
        }
    
        /** * 动态数据源 * * @param dataSources * @return */
        @Bean
        public DataSource dataSource(Map<String, DataSource> dataSources) {
            return new DynamicDataSource(dataSources);
        }
    
        /** * 事务管理器 * * @param dataSource * @return */
        @Bean
        public DataSourceTransactionManager txManager(DataSource dataSource) {
            return new DataSourceTransactionManagerAdapter(dataSource);
        }
    
        /** * SqlSession工厂 * * @param dataSource * @return * @throws Exception */
        @Bean
        public SqlSessionFactory sqlSessionFactory(DataSource dataSource) throws Exception {
            // 配置MapperConfig
            org.apache.ibatis.session.Configuration configuration = new org.apache.ibatis.session.Configuration();
    
            //当数据库集群时,配置多个数据源,通过设置不同的DatebaseId来区分数据源,同时sql语句中通过DatabaseId来指定匹配哪个数据源
            //configuration.setDatabaseId("Mysql-1");
    
            // 这个配置使全局的映射器启用或禁用缓存
            configuration.setCacheEnabled(true);
    
            // 允许 JDBC 支持生成的键，需要适合的驱动（如MySQL，SQL Server，Sybase ASE）。
            // 如果设置为 true 则这个设置强制生成的键被使用，尽管一些驱动拒绝兼容但仍然有效（比如 Derby）。
            // 但是在 Oracle 中一般不需要它，而且容易带来其它问题，比如对创建同义词DBLINK表插入时发生以下错误：
            // "ORA-22816: unsupported feature with RETURNING clause" 在 Oracle
            // 中应明确使用 selectKey 方法
            //configuration.setUseGeneratedKeys(false);
    
            // 配置默认的执行器:
            // SIMPLE :> SimpleExecutor 执行器没有什么特别之处;
            // REUSE :> ReuseExecutor 执行器重用预处理语句,在一个Service方法中多次执行SQL字符串一致的操作时,会复用Statement及Connection,
            // 也就是说不需要再预编译Statement,不需要重新通过DataSource生成Connection及释放Connection,能大大提高操纵数据库效率;
            // BATCH :> BatchExecutor 执行器重用语句和批量更新
            configuration.setDefaultExecutorType(ExecutorType.REUSE);
            // 全局启用或禁用延迟加载，禁用时所有关联对象都会即时加载
            configuration.setLazyLoadingEnabled(false);
            // 设置SQL语句执行超时时间缺省值，具体SQL语句仍可以单独设置
            configuration.setDefaultStatementTimeout(5000);
    
            SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
            sqlSessionFactoryBean.setDataSource(dataSource);
            sqlSessionFactoryBean.setConfiguration(configuration);
            // 匹配多个 MapperConfig.xml, 使用mappingLocation属性
            sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath*:com/bob/mvc/mapper/*Mapper.xml"));
            return sqlSessionFactoryBean.getObject();
        }
    
        private DataSource generateDataSource(String url) {
            BasicDataSource dataSource = new BasicDataSource();
            dataSource.setDriverClassName(DRIVER_CLASS_NAME);
            //针对mysql获取字段注释
            dataSource.addConnectionProperty("useInformationSchema", "true");
            //dataSource.addConnectionProperty("remarksReporting","true"); 针对oracle获取字段注释
            dataSource.setUrl(url);
            dataSource.setUsername(USERNAME);
            dataSource.setPassword(PASSWORD);
            dataSource.setMaxTotal(50);
            dataSource.setMinIdle(5);
            dataSource.setMaxIdle(10);
            return dataSource;
        }
    
    }

这个配置类中，我定义了两个连接数据库的DataSource，一个读一个写，若有多个从库可以定义多个读数据源，数据源使用的是dbcp2，还定义了一个动态数据源，这个数据源就是包装了所有的实际数据源，在连接的时候动态的选择读或者写，看DynamicDataSource的代码：

import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.Map.Entry;
    import java.util.concurrent.atomic.AtomicLong;
    
    import javax.sql.DataSource;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
    import org.springframework.util.Assert;
    
    /** * 动态数据源配置 * * @create 2018-01-16 9:56 */
    public class DynamicDataSource extends AbstractRoutingDataSource { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(DynamicDataSource.class);
    
        private final AtomicLong readRequestedTime = new AtomicLong(0);
    
        private DataSource writeDataSource;
        private List<DataSource> readDataSources = new ArrayList<DataSource>();
    
        public DynamicDataSource(Map<String, DataSource> dataSources) {
            Assert.notEmpty(dataSources, "至少需要定义一个数据源");
            //若只有一个数据源，则指定其为写数据源
            if (dataSources.size() == 1) {
                LOGGER.info("当前容器内只定义了一个数据源，指定其为写数据源");
                writeDataSource = dataSources.values().iterator().next();
                return;
            }
            for (Entry<String, DataSource> entry : dataSources.entrySet()) {
                if (entry.getKey().toLowerCase().contains("write")) {
                    if (writeDataSource != null) {
                        throw new IllegalStateException("写数据源只能定义一个");
                    }
                    this.writeDataSource = entry.getValue();
                    continue;
                }
                //若数据源名称不含write,则默认为读数据源
                readDataSources.add(entry.getValue());
            }
            if (writeDataSource == null) {
                throw new IllegalStateException("至少需要定义一个写数据源");
            }
        }
    
        /** * 当前数据操作类型时写时或只有一个写数据源时,直接返回此数据源; * 当有多个读数据源时,轮询读数据源 * * @return */
        @Override
        protected Object determineCurrentLookupKey() {
            DataManipulationType manipulationType = DataSourceTransactionManagerAdapter.getCurrentManipulationType();
            if (manipulationType == DataManipulationType.WRITE || readDataSources.isEmpty()) {
                return (long)-1;
            }
            long times = readRequestedTime.getAndIncrement();
            return times % readDataSources.size();
        }
    
        @Override
        public void afterPropertiesSet() {
            HashMap<Integer, DataSource> dataSources = new HashMap<Integer, DataSource>();
            //按序号设置数据源，0为写数据源,之后为读数据源,方便多个读数据源时负载均衡
            dataSources.put(0, writeDataSource);
            for (int i = 0; i < readDataSources.size(); i++) {
                dataSources.put(i + 1, readDataSources.get(i));
            }
            setDefaultTargetDataSource(writeDataSource);
            setTargetDataSources(new HashMap<Object, Object>(dataSources));
            super.afterPropertiesSet();
        }
    
        @Override
        protected DataSource determineTargetDataSource() {
            DataSource dataSource;
            int key = ((Long)determineCurrentLookupKey()).intValue();
            if (key == -1) {
                LOGGER.info("获取写数据源");
                dataSource = writeDataSource;
            } else {
                LOGGER.info("获取读数据源");
                dataSource = readDataSources.get(key);
            }
            return dataSource;
        }
    
        /** * 如果从库宕机,则从主库读取 * * @return * @throws SQLException */
        @Override
        public Connection getConnection() throws SQLException {
            try {
                return super.getConnection();
            } catch (SQLException e) {
                if (DataSourceTransactionManagerAdapter.getCurrentManipulationType() == DataManipulationType.READ) {
                    LOGGER.warn("尝试从读数据源生成连接失败", e);
                    return writeDataSource.getConnection();
                }
                throw e;
            }
        }
    }

DataSource最重要的方法就是getConnection，DynamicDataSource 继承自AbstractRoutingDataSource ，它的作用就是能够自定义择取数据源的逻辑，然后通过选择的数据源生成Connection对象，我在此处加了多个从库的轮询以实现负载均衡。

动态数据源主要是通过DataSourceTransactionManagerAdapter.getCurrentManipulationType()方法来判断当前应该连接读数据源还是写数据源，DataSourceTransactionManagerAdapter是我自对Spring事务管理器的一个拓展，看如下代码：

import javax.sql.DataSource;
    
    import org.springframework.jdbc.datasource.DataSourceTransactionManager;
    import org.springframework.transaction.TransactionDefinition;
    import org.springframework.transaction.support.DefaultTransactionStatus;
    
    /** * 数据源事务管理器拓展 * * @create 2018-01-16 19:33 */
    public class DataSourceTransactionManagerAdapter extends DataSourceTransactionManager { 
    
        public static final ThreadLocal<DataManipulationType> DATA_MANIPULATION_TYPE = new ThreadLocal<>();
    
        public DataSourceTransactionManagerAdapter(DataSource dataSource) {
            super(dataSource);
        }
    
        @Override
        protected void doBegin(Object transaction, TransactionDefinition definition) {
            if (definition.isReadOnly()) {
                DATA_MANIPULATION_TYPE.set(DataManipulationType.READ);
            } else {
                DATA_MANIPULATION_TYPE.set(DataManipulationType.WRITE);
            }
            super.doBegin(transaction, definition);
        }
    
        @Override
        protected void prepareForCommit(DefaultTransactionStatus status) {
            DATA_MANIPULATION_TYPE.remove();
            super.prepareForCommit(status);
        }
    
        @Override
        protected void doRollback(DefaultTransactionStatus status) {
            DATA_MANIPULATION_TYPE.remove();
            super.doRollback(status);
        }
    
        /** * 查看当前线程的数据操作类型 * * @return */
        public static DataManipulationType getCurrentManipulationType() {
            return DATA_MANIPULATION_TYPE.get();
        }
    
    }
    
    /** * 数据操作类型 * * @author Administrator * @create 2018-01-17 19:54 */
    enum DataManipulationType {
    
        READ,
        WRITE;
    
    }

DataSourceTransactionManagerAdapter 重写了DataSourceTransactionManager 的doBegin方法，在执行父类执行doBegin前插入自己的逻辑。我在此处先判断当前事务信息，如果当前事务是只读事务，则将枚举DataManipulationType.READ实例存入ThreadLocal类型变量DATA\_MANIPULATION\_TYPE 中，之后动态数据源就可以通过DATA\_MANIPULATION\_TYPE 中的线程绑定值来选择相应的数据源了，在事务提交或者回滚前清空变量值。DataSourceTransactionManager 的doBegin方法里有通过dataSource生成Connection的逻辑，这个dataSource就是我们定义的DynamicDataSource 。

以上步骤的**关键就是事务信息里能够明确的指出当前业务应该连接的是读操作还是写操作**，使用传统的事务配置是实现不了的，所以需要自定义事务的实现方式，先看TransactionAttributeGenerator相关代码：

import java.lang.reflect.Method;
    import java.util.Arrays;
    import java.util.List;
    
    import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
    import org.springframework.transaction.interceptor.TransactionAttribute;
    import org.springframework.transaction.interceptor.TransactionAttributeSource;
    
    /** * 事务属性生成器 * * @create 2018-01-17 19:45 */
    public class TransactionAttributeGenerator implements TransactionAttributeSource { 
    
        private static final List<String> WRITE_KEY_WORDS = Arrays.asList("insert", "create", "delete", "remove", "modify", "update", "change");
    
        @Override
        public TransactionAttribute getTransactionAttribute(Method method, Class<?> targetClass) {
            DefaultTransactionAttribute transactionAttribute = new DefaultTransactionAttribute();
            //默认当前方法是读操作
            transactionAttribute.setReadOnly(true);
            //不设置事务的隔离级别,默认使用数据库的设置
            //transactionAttribute.setIsolationLevel(Connection.TRANSACTION_REPEATABLE_READ);
            //设置事务的超时时间,单位秒,注意是事务不是连接的超时设置
            //transactionAttribute.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
            transactionAttribute.setTimeout(300);
    
            //若方法或者类上标识了@WriteManipulation，则指定是写操作
            if (method.isAnnotationPresent(WriteManipulation.class) || targetClass.isAnnotationPresent(WriteManipulation.class)) {
                transactionAttribute.setReadOnly(false);
                return transactionAttribute;
            }
            //若方法上还有写操作的关键字,则指定是写操作
            String methodName = method.getName().toLowerCase();
            for (String keyword : WRITE_KEY_WORDS) {
                if (methodName.contains(keyword)) {
                    transactionAttribute.setReadOnly(false);
                    break;
                }
            }
            return transactionAttribute;
        }
    }
    
    /** * 标识数据写操作 * * @author Administrator * @create 2018-01-17 20:05 */
    @Documented
    @Retention(RetentionPolicy.RUNTIME)
    @Target({ElementType.METHOD,ElementType.TYPE})
    public @interface WriteManipulation { 
    }

TransactionAttributeSource 接口的作用就是从被代理方法及类上生成事务属性TransactionAttribute ，事务管理器通过TransactionAttribute 生成TransactionInfo，就是当前线程的事务信息了。我们定义了自己的事务信息源，若当前方法含有写操作的关键字或者方法及类上标识了@WriteManipulation注解，则生成的事务属性不是只读的，否则一律视为只读事务。

事务信息源设置好了， 接下来需要将其应用到事务上，看TransactionAspectInvoker相关代码：

import org.aspectj.lang.ProceedingJoinPoint;
    import org.aspectj.lang.reflect.MethodSignature;
    import org.springframework.aop.support.AopUtils;
    import org.springframework.transaction.interceptor.TransactionAspectSupport;
    
    /** * 事务切面执行器 * * @create 2017-12-27 11:17 */
    public class TransactionAspectInvoker extends TransactionAspectSupport { 
    
        public TransactionAspectInvoker() {
            setTransactionAttributeSource(new TransactionAttributeGenerator());
            //可以配置多个事务属性源,对不同的方法生成不同的事务属性
            //ssetTransactionAttributeSources(new TransactionAttributeSource[] {});
        }
    
        /** * @param joinPoint * @return * @throws Throwable */
        public Object invoke(final ProceedingJoinPoint joinPoint) throws Throwable {
            // Work out the target class: may be {@code null}.
            // The TransactionAttributeSource should be passed the target class
            // as well as the method, which may be from an interface.
            Class<?> targetClass = (joinPoint.getThis() != null ? AopUtils.getTargetClass(joinPoint.getThis()) : null);
    
            // Adapt to TransactionAspectSupport's invokeWithinTransaction...
            return invokeWithinTransaction(((MethodSignature)joinPoint.getSignature()).getMethod(), targetClass, () -> joinPoint.proceed());
        }
    }

TransactionAspectSupport 这个类就是事务执行的关键，invokeWithinTransaction方法，顾名思义，在这个方法内执行Service方法时，事务就能发挥作用。我们对其做了一些拓展，将自定义的事务信息源注入到其中，同时将其适配为一个AOP的Advice，切入方法，准备将去切入到Service层做事务的切面定义。

下面看自定义事务切面配置：

/** * 事务配置类 * * @create 2017-12-27 9:50 */
    @Configuration
    public class TransactionConfiguration { 
    
        @Bean
        public TransactionAspectInvoker transactionAspectInvoker() {
            return new TransactionAspectInvoker();
        }
    
    }
    
    import org.aspectj.lang.ProceedingJoinPoint;
    import org.aspectj.lang.annotation.Around;
    import org.aspectj.lang.annotation.Aspect;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    /** * 事务切面 * * @create 2017-12-27 13:53 */
    @Aspect
    @Component
    public class TransactionAspectJAdvisor { 
    
        @Autowired
        private TransactionAspectInvoker transactionAspectInvoker;
    
        @Around("execution(public * com.bob.mvc.service..*(..))")
        public Object invokeWithTransaction(ProceedingJoinPoint joinPoint) throws Throwable {
            return transactionAspectInvoker.invoke(joinPoint);
        }
    
    }

定义一个AOP切面，使用TransactionAspectInvoker 来环绕Service层以实现事务配置。

整篇博客的主流程已经讲解完了，每个类的作用以及整体架构和原理也都说明了，下面最一个总结吧。

1.  使用自定义事务的形式，嵌套自定义事务信息源，在生成事务信息时，通过方法名称或者注解来指定当前事务是否是只读事务。
2.  对事务管理器做了拓展，在事务执行前期，通过对事务信息是否只读的判断，将标识放入ThreadLocal类型的变量中。
3.  使用RoutingDataSource，其内选择数据源时，以ThreadLocal变量中的是否只读标识来判断是连接读数据源还是写数据源。

[SouthEast]: /images/20220601/35335dc2dda24daca524841177792b71.png