MySQL数据迁移的流程介绍

本文介绍了一次 MySQL 数据迁移的流程，通过方案选型、业务改造、双写迁移最终实现了亿级数据的迁移。

一、背景

预约业务是 vivo 游戏中心的重要业务之一。由于历史原因，预约业务数据表与其他业务数据表存储在同一个数据库中。当其他业务出现慢 SQL 等异常情况时，可能会直接影响到预约业务，从而降低系统整体的可靠性和稳定性。为了尽可能提高系统的稳定性和数据隔离性，我们迫切需要将预约相关数据表从原来的数据库中迁移出来，单独建立一个预约业务的数据库。

二、方案选型

常见的迁移方案大致可以分为以下几类：

而预约业务有以下特点：

读写场景多，频率高，在用户预约/取消预约/福利发放等场景均涉及到大量的读写。

不可接受停机，停机不可避免的会造成经济损失，在有其他方案的情况下不适合选择此方案。

大部分的场景能接受秒级的数据不一致，少部分不能。

结合这些特点，我们再评估下上面的方案：

停机迁移方案需要停机，不适用于预约场景。预约场景存在不活跃的用户数据，如果用渐进式迁移方案的话很难迁移干净，可能还需要再写一个迁移任务来辅助完成迁移。而双写方案最大的优势在于每一步操作都可向上回滚，能尽可能的保证业务不出问题。

因此，最终选择的是双写方案。预约业务涉及到的读写场景多，每一个场景单独进行改造的成本大，采用 Mybatis 插件来实现迁移所需的双写等功能，可以有效降低改造成本。

三、前期准备

3.1 全量同步&增量同步&一致性校验

这几步使用了公司提供的数据同步工具。全量同步基于 MySQLDump 实现；增量同步基于 binlog 实现；一致性校验通过在新老库各选一个分块，然后聚合列数据计算并对比其特征值实现。

3.2 代码改造

引入了新库，那自然就需要在项目里新建数据源，并创建表对应的 Mybatis Mapper 类。这里有一个小细节需要注意，Mybatis 默认的 BeanNameGenerator 是

AnnotationBeanNameGenerator，它会使用类名作为 BeanName 注册到 Spring 的 ioc 容器中，Spring 启动时如果发现有了两个重名 Bean 就会启动失败，笔者这里给 Mybatis 设置了一个新的 BeanNameGenerator ，使用类的全路径名作为 BeanName 解决了问题。

public class FullPathBeanNameGenerator implements BeanNameGenerator {
    @Override
    public String generateBeanName(BeanDefinition definition, BeanDefinitionRegistry registry) {
        return definition.getBeanClassName();
    }
}

还有一点是主键 id，本次预约迁移需要保证新老库主键 id 一致，预约业务没做分库分表，id 都是直接用 MySQL 的自增 id，没有用 id 生成器之类的中间件。因此插入新表时只需要使用插入老表后 Mybatis 自动设置好的 id 即可，这次迁移前先检查了一遍业务代码，确保插入语句都用了 Mybatis 的 useGeneratedKeys 功能来自动设置 id。

3.3 插件实现

Mybatis 插件可以拦截 SQL 语句执行过程中的某一点进行干预和处理，而 Executor 是 Mybatis 中负责执行 SQL 语句的核心组件。我们可以对 Executor 的 update 和 query 方法进行代理以实现迁移所需的功能。

插件需要为读写场景分别实现以下功能：

考虑到开关切换部分的代码逻辑较为简单，因此在下文中，笔者将不再过多介绍该部分的具体实现，而是着重介绍如何在插件中使用老库的执行语句来访问新的数据库。此外，代码里会涉及到 Mybatis 相关的一些概念，由于网上已经有较多详尽的资料，这里就不再赘述。

迁移插件代理了 Executor 的 query 和 update 方法，首先在插件里获取到当前执行的 SQL 语句所在的 Mapper 路径。

@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}),
                @Signature(type = Executor.class, method = "query", args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class, CacheKey.class, BoundSql.class}),
        }
)
public class AppointMigrateInterceptor implements Interceptor {
 
    @Override
    public Object intercept(Invocation invocation) throws Throwable {
 
        Object[] args = invocation.getArgs();
        // Mybatis插件代理的Executor的update或者query方法，第一个参数就是MappedStatement
        MappedStatement ms = (MappedStatement) args[0];
        SqlCommandType sqlCommandType = ms.getSqlCommandType();
        String id = ms.getId();
        // 从MappedStatement id中获取对应的Mapper接口文件全路径
        String sourceMapper = id.substring(0, id.lastIndexOf("."));
 
        // ...
    }
     
    // ...
}

得到老库 Mapper 路径后，将其转换为新库 Mapper 路径，再使用 Class.forName 获取到新库 Mapper 类，然后用新库的 sqlSessionFactory 开启 sqlSession，再获取反射调用所需的方法、对象、参数，在新库上执行语句。

protected Object invoke(Invocation invocation, TableConfiguration tableConfiguration) throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
    // 获取 MappedStatement
    MappedStatement ms = (MappedStatement) invocation.getArgs()[0];
 
    // 获取 Mybatis 封装好的入参，封装函数 MapperMethod.convertArgsToSqlCommandParam(Object[] args)
    Object parameter = invocation.getArgs()[1];
 
    // 使用 Class.forName 获取到的新库 Mapper
    Class targetMapperClass = tableConfiguration.getTargetMapperClazz();
 
    // 使用新库的 sqlSessionFactory 创建 sqlSession
    SqlSession sqlSession = sqlSessionFactory.openSession();
    Object result = null;
    try{
        // 使用新库的 Mapper 路径获取对应的 MapperProxy 对象
        Object mapper = sqlSession.getMapper(targetMapperClass);
 
        // 将 Mybatis 封装好的参数转换为原始参数
        Object[] paramValues = getParamValue(parameter);
 
        // 使用 mappedStatement Id 从新库对应的 Mapper 里获取对应的方法
        Method method = getMethod(ms.getId(), targetMapperClass, paramValues);
        paramValues = fixNullParam(method, paramValues);
 
        // 反射调用新库 Mapper 的方法,本质上执行的是 MapperProxy.invoke
        result = method.invoke(mapper, paramValues);
    } finally {
        sqlSession.close();
    }
    return result;
}
 
private Object[] fixNullParam(Method method, Object[] paramValues) {
    if (method.getParameterTypes().length > 0 && paramValues.length == 0) {
        return new Object[]{null};
    }
    return paramValues;
}

 

 

（左右滑动查看更多）

上述代码里，getMethod 方法负责从新库 Mapper 类里找到对应的方法，以用于后续的反射调用。

 

 

private Method getMethod(String id, Class mapperClass) throws NoSuchMethodException {
    //获取参数对应的 class
    String methodName = id.substring(id.lastIndexOf(".") + 1);
    String key = id;
    // methodCache 用来缓存 MappedStatement 和对应的 Method，避免每次都从 Mapper 里查找
    Method method = methodCache.get(key);
    if (method == null){
        method = findMethodByMethodSignature(mapperClass, methodName);
        if (method == null){
            throw new NoSuchMethodException("No such method " + methodName + " in class " + mapperClass.getName());
        }
        methodCache.put(key,method);
    }
    return method;
}
 
private Method findMethodByMethodSignature(Class mapperClass,String methodName) throws NoSuchMethodException {
    // mybatis 的 Mapper 内的方法不支持重载，所以这里只要方法名匹配到了就行，不用进行参数的匹配
    Method method = null;
    for (Method m : mapperClass.getMethods()) {
        if (m.getName().equals(methodName)) {
            method = m;
            break;
        }
    }
    return method;
}

得到方法后，还需要得到反射调用所需的参数。Mybatis 执行到 Executor.update/query 方法时，参数已经经过   MapperMethod.convertArgsToSqlCommandParam(Object[] args) 方法封装，不能直接用来执行  MapperProxy.invoke ，需要转换后才可用。下图是MapperMethod.convertArgsToSqlCommandParam(Object[] args) 的封装过程，而下面的 getParamValue 是这个函数的逆过程。

 

 

private Object[] getParamValue(Object parameter) {
    List