基于MySql主从分离的代码层实现

程序员波特

发布于 2024-01-19 09:47:26

1200

发布于 2024-01-19 09:47:26

文章被收录于专栏：魔法书

前言

该文是基于上篇《MySQL主从分离的实现》的代码层实现，所以本文配置的主数据库和从数据库的数据源都是在上篇博文中已经介绍了的。

动态选择数据源的配置

由于我们在写数据的时候需要使用主库的数据源，读的时候需要从库的数据源，我们可以在Spring源码中，通过DataSource可以找到AbstractDataSource抽象类，由于我们需要动态的选择数据源，我们可以通过AbstractDataSource发现他的一个子类是AbstractRoutingDataSource的抽象类，通过类名我们可以知道该类是具有路由功能的，可以路由到不同的数据源，这个类中有一个方法determineTargetDateSource()，该方法就是决定目标数据源的，该方法会调用determineCurrentLookupKey()，就是决定数据源的名字了，该方法是一个抽象的，所以我们需要去继承AbstractRoutingDataSource这个类，并实现determineCurrentLookupKey()这个方法，来动态选择数据源，读数据的时候选择从库的数据源，写操作的时候选择主库的数据源

public class DynamicDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return DynamicDataSourceHolder.getDBType();
    }
}

编写DynamicDataSourceHolder

public class DynamicDataSourceHolder {
    private static Logger logger = LoggerFactory.getLogger(DynamicDataSourceHolder.class);

    //ThreadLocal是线程安全的
    private static ThreadLocal<String> contextHolder = new ThreadLocal<>();
    public static final String DB_MASTER = "master";
    public static final String DB_SLAVE = "slave";

    public static String getDBType() {
        String db = contextHolder.get();
        if (db == null){
            db = DB_MASTER; // 默认为master,因为master即支持读也支持写
        }
        return db;
    }

    /**
     * 设置线程的dbType
     * @param str
     */
    public static void setDBType(String str) {
        logger.debug("所使用的数据源："+ str);
        contextHolder.set(str);
    }

    /**
     * 清理连接类型
     */
    public static void clearDBType(){
        contextHolder.remove();
    }
}

设置mybatis的拦截器

完成路由后，我们需要依靠拦截器对传递进来的SQL信息来选择数据源，例如传进来的是insert,update,delete语句，就使用主库的数据源，如果是select就选择从库的数据源。

@Intercepts({@Signature(type = Executor.class,method = "update",args = {MappedStatement.class,Object.class}),
        @Signature(type = Executor.class,method = "query",
                args = {MappedStatement.class,Object.class, RowBounds.class, ResultHandler.class})})
public class DynamicDataSourceInterceptor implements Interceptor {
    private static Logger logger = LoggerFactory.getLogger(DynamicDataSourceInterceptor.class);
    //使用正则表达式匹配增删改
    private static final String REGEX = ".*insert\\u0020.*|.*delete\\u0020.*|.*update\\u0020.*";

    /**
     * 拦截方法
     * @param invocation
     * @return
     * @throws Throwable
     */
    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        // 判断当前操作是否是事务的
        // 使用@Transactional来处理，则会返回true
        boolean transactionActive = TransactionSynchronizationManager.isActualTransactionActive();
        Object[] args = invocation.getArgs();
        MappedStatement ms = (MappedStatement) args[0];
        String lookupKey = DynamicDataSourceHolder.DB_MASTER;
        if ( !transactionActive ) {
            // 如果是查询操作
            if (ms.getSqlCommandType().equals(SqlCommandType.SELECT)) {
                //selectKey 为自增id查询主键SELECT_KEY_SUFFIX()方法，使用主库
                if (ms.getId().contains(SelectKeyGenerator.SELECT_KEY_SUFFIX)){
                    lookupKey = DynamicDataSourceHolder.DB_MASTER;
                } else {
                    BoundSql boundSql = ms.getSqlSource().getBoundSql(args[1]);
                    // 对制表符，换行符，空格符就行替换
                    String sql = boundSql.getSql().toLowerCase(Locale.CHINA).replaceAll("[\\t\\n\\r]", " ");
                    //增删改使用主库，查使用从库
                    if (sql.matches(REGEX)) {
                        lookupKey = DynamicDataSourceHolder.DB_MASTER;
                    } else {
                        lookupKey = DynamicDataSourceHolder.DB_SLAVE;
                    }
                }
            }

        } else {
            lookupKey = DynamicDataSourceHolder.DB_MASTER;
        }
        logger.debug("设置方法[{}] use [{}] Strategy,SqlCommandType [{}] ...",
                ms.getId(), lookupKey, ms.getSqlCommandType().name());
        DynamicDataSourceHolder.setDBType(lookupKey);
        return invocation.proceed();
    }

    /**
     * 决定返回封装好的对象还是代理对象
     * 增删改查得操作
     * @param target
     * @return
     */
    @Override
    public Object plugin(Object target) {
        //当我们拦截的对象是Executor时，就拦截，通过intercept()方法 决定所使用的数据源
        //为什么要拦截Executor类型呢？因为在我们的mybatis中，Executor是用来支持一系列增删改查操作的
        //只要我们检测到拦截的对象包含增删改查操作，就拦截下来，使用intercept()方法，决定所使用的数据源
        if (target instanceof Executor) {
            return Plugin.wrap(target,this);
        } else {
            return target;
        }
    }

    /**
     * 在类初始化的时候，去做一些相关的设置
     * @param properties
     */
    @Override
    public void setProperties(Properties properties) {

    }
}

只是编写完这个方法是没用的，我们还需要在mybaties-config.xml配置文件中，配置上我们实现的拦截器，如下：

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration
  PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
  "http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>
	<!-- 配置全局属性 -->
	<settings>
		<!-- 使用jdbc的getGeneratedKeys获取数据库自增主键值 -->
		<setting name="useGeneratedKeys" value="true" />

		<!-- 使用列别名替换列名 默认:true -->
		<setting name="useColumnLabel" value="true" />

		<!-- 开启驼峰命名转换:Table{create_time} -> Entity{createTime} -->
		<setting name="mapUnderscoreToCamelCase" value="true" />
		<!-- 打印查询语句 -->
		<setting name="logImpl" value="STDOUT_LOGGING" />
	</settings>
	<plugins>
		<plugin interceptor="cn.reminis.o2o.dao.split.DynamicDataSourceInterceptor" />
	</plugins>
</configuration>

配置多数据源

将原来配置dataSource的bean，改为abstractDatasource，并增加主库数据源和从库数据源的配置，如下：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/context
    http://www.springframework.org/schema/context/spring-context.xsd">
    <!-- 配置整合mybatis过程 -->
    <context:property-placeholder location="classpath:jdbc.properties"/>

    <!-- 2.数据库连接池 -->
    <bean id="abstractDataSource" abstract="true" class="com.mchange.v2.c3p0.ComboPooledDataSource" destroy-method="close">
        <!-- 配置连接池属性 -->
        <!--
        <property name="driverClass" value="${jdbc.driver}" />
        <property name="jdbcUrl" value="${jdbc.url}" />
        <property name="user" value="${jdbc.username}" />
        <property name="password" value="${jdbc.password}" />
        -->

        <!-- c3p0连接池的私有属性 -->
        <property name="maxPoolSize" value="30" />
        <property name="minPoolSize" value="10" />
        <!-- 关闭连接后不自动commit -->
        <property name="autoCommitOnClose" value="false" />
        <!-- 获取连接超时时间 -->
        <property name="checkoutTimeout" value="10000" />
        <!-- 当获取连接失败重试次数 -->
        <property name="acquireRetryAttempts" value="2" />
    </bean>

    <!--主库的数据源配置-->
    <bean id="master" parent="abstractDataSource">
        <property name="driverClass" value="${jdbc.master.driver}" />
        <property name="jdbcUrl" value="${jdbc.master.url}" />
        <property name="user" value="${jdbc.master.username}" />
        <property name="password" value="${jdbc.master.password}" />
    </bean>

    <!--从库的数据源配置-->
    <bean id="slave" parent="abstractDataSource">
        <property name="driverClass" value="${jdbc.slave.driver}" />
        <property name="jdbcUrl" value="${jdbc.slave.url}" />
        <property name="user" value="${jdbc.slave.username}" />
        <property name="password" value="${jdbc.slave.password}" />
    </bean>

    <!--配置动态数据源。这里targetDataSource就是路由数据源的名称-->
    <bean id="dynamicDataSource" class="cn.reminis.o2o.dao.split.DynamicDataSource">
        <property name="targetDataSources">
            <map>
                <entry value-ref="master" key="master"></entry>
                <entry value-ref="slave" key="slave"></entry>
            </map>
        </property>
    </bean>

    <!--懒加载，因为数据源是程序运行时决定的-->
    <bean id="dataSource" class="org.springframework.jdbc.datasource.LazyConnectionDataSourceProxy">
        <property name="targetDataSource">
            <ref bean="dynamicDataSource" />
        </property>
    </bean>

    <!-- 3.配置SqlSessionFactory对象 -->
    <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
        <!-- 注入数据库连接池 -->
        <property name="dataSource" ref="dataSource" />
        <!-- 配置MyBaties全局配置文件:mybatis-config.xml -->
        <property name="configLocation" value="classpath:mybatis-config.xml" />
        <!-- 扫描entity包 使用别名 -->
        <property name="typeAliasesPackage" value="cn.reminis.o2o.entity" />
        <!-- 扫描sql配置文件:mapper需要的xml文件 -->
        <property name="mapperLocations" value="classpath:mapper/*.xml" />
    </bean>

    <!-- 4.配置扫描Dao接口包，动态实现Dao接口，注入到spring容器中 -->
    <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
        <!-- 注入sqlSessionFactory -->
        <property name="sqlSessionFactoryBeanName" value="sqlSessionFactory" />
        <!-- 给出需要扫描Dao接口包 -->
        <property name="basePackage" value="cn.reminis.o2o.dao" />
    </bean>
</beans>

我们在jdbc.properties配置文件中，配置主从库数据源的地址：

## 主库数据源配置
jdbc.master.driver=com.mysql.jdbc.Driver
jdbc.master.url=jdbc:mysql://192.168.0.188:3306/o2o?useUnicode=true&characterEncoding=utf8
jdbc.master.username=root
jdbc.master.password=123456

## 从库数据源配置
jdbc.slave.driver=com.mysql.jdbc.Driver
jdbc.slave.url=jdbc:mysql://192.168.0.152:3306/o2o?useUnicode=true&characterEncoding=utf8
jdbc.slave.username=root
jdbc.slave.password=root