作者：来自 vivo 互联网服务器团队- Li Gang
↓推荐关注↓
本文介绍了一次 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(); MappedStatement ms = (MappedStatement) args[0 ]; SqlCommandType sqlCommandType = ms.getSqlCommandType(); String id = ms.getId(); 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 ms = (MappedStatement) invocation.getArgs()[0 ]; Object parameter = invocation.getArgs()[1 ]; Class> targetMapperClass = tableConfiguration.getTargetMapperClazz(); SqlSession sqlSession = sqlSessionFactory.openSession(); Object result = null; try { Object mapper = sqlSession.getMapper(targetMapperClass); Object[] paramValues = getParamValue(parameter); Method method = getMethod(ms.getId(), targetMapperClass, paramValues); paramValues = fixNullParam(method, paramValues); 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 String methodName = id.substring(id.lastIndexOf("." ) + 1 ); String key = id; 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 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 paramValues = new ArrayList<>(); if (parameter instanceof Map) { Map paramMap = (Map) parameter; if (paramMap.containsKey("collection" )) { paramValues.add(paramMap.get("collection" )); } else if (paramMap.containsKey("array" )) { paramValues.add(paramMap.get("array" )); } else { int count = 1 ; while (count <= paramMap.size() / 2 ){ try { paramValues.add(paramMap.get("param" +(count++))); }catch (BindingException e){ break ; } } } } else if (parameter != null){ paramValues.add(parameter); } return paramValues.toArray(); } （左右滑动查看更多）
通过上述流程，我们就能使用 Mybatis 插件拦截老库的执行过程，实现迁移所需的读写数据源切换/新老库查询结果对比/先写老库再异步写新库等功能。
4.1 上线双写改造后的业务代码，上线时只读写老库
读开关：只读老库
写开关：只写老库
新老库查询结果对比开关：关
此时业务仍只读写老库。
4.2 使用公司中间件平台提供的数据工具同步老库数据到新库
读开关：只读老库
写开关：只写老库
新老库查询结果对比开关：关
第1步和第2步并没有严格的顺序要求，只要在切换为双写前做完第1步和第2步就好。
条件允许的情况下，全量+增量同步时应选择不对外提供服务的离线从库作为数据源，避免主从延迟等问题对线上业务造成影响。
4.3 停止同步程序，然后开启双写
读开关：只读老库（开启查询结果对比开关）
写开关：双写
新老库查询结果对比开关：开
老库追上新库后，对数据做一次全量校验，避免出现数据不一致的情况。此外还需要开启新老库查询结果对比开关，通过日志监控观察新老库的查询结果是否一致。
停止数据同步和切换双写之间必然有时间差，如果先开启双写再停止数据同步，则可能出现插入重复数据或数据被覆盖的情况。因此需要对数据同步工具和迁移插件进行改造，以处理数据异常的情况，但是这样改造需要处理的情况较多，改造成本较高。所以这里选择先停止同步，再切换到双写，中间丢失的数据使用对比&补偿任务恢复，由于此时仍然全量读老库，所以对业务不会有影响。需要注意的是，双写阶段的时间不应太长，只要确保新老库数据一致就应该前进到下一步。
这一步在实际操作过程中需要注意以下情况：
预约业务新库的主键 id 需要和老库保持一致，因此在迁移前检查了一遍业务代码，确保插入语句都用了 Mybatis 的 useGeneratedKeys 功能来返回 id ，这样插入新库时可以直接用设置好 id 的对象。但是这里有一个问题，批量插入时 Mybatis 自动设置的 id 和数据库生成的自增主键不一定完全一致，比如批量 insert ignore 和 on duplicate key update 语句。
这个问题和 useGeneratedKeys 的实现有关，代码可参考
com.mysql.jdbc.StatementImpl#getGeneratedKeysInternal(long) 函数，以下是其执行逻辑：
Mybatis 执行完插入语句后，MySQL 会返回这次插入影响的数据行数，注意，使用 insert ignore 插入时，忽略的那部分数据不会加到影响的行数上。
Mybatis 使用 SELECT LAST_INSERT_ID() 查询这次插入的最小 id 。
Mybatis 循环遍历插入时用的对象列表，循环的最大次数为第1步里获取的这次插入影响的行数，使用 n 代表当前的循环次数，列表中的每个对象的 id 被赋值为 LAST_INSERT_ID() + n*AUTO_INCREMENT 。
举例来说，假设老库的某张表里有数据 b ，其 id=1，此时往该表使用 insert ignore 批量插入三条数据 a,b,c，其在表内的 id 为 a:2、b:1、c:3，返回的影响行数为2，SELECT LAST_INSERT_ID() 返回的是2，因此 Mybatis 往对象里设置的主键分别为 a:2、b:3、c:null，再使用这个设置好 id 的对象列表插入新库时会导致新老库 id 不一致。
解决方案：由于直接删除 ignore 会改变这条 SQL 的语义，无法通过修改语句来解决问题。所以我们只能在迁移插件里跳过这条语句，使其固定写入老库。然后在业务层单独对其进行迁移改造，将插入新库的流程修改为先使用 id 以外的唯一键查询一次老库的数据，获取到 id 以后设置到对象列表里，再插入新库。
预约业务有部分逻辑用到了事务，但这部分逻辑在双写期间均可以暂停功能，因此迁移插件没有实现事务的支持。如果需要支持业务的话可以不依赖插件，在业务层单独对那部分代码进行改造。
双写过程中是异步写新库，需要重点关注是否会有线程安全问题。举例来说，假设有个业务需要往表里插入一个列表，插入完列表后又对列表进行了修改，比如执行了 List.clear() 函数或者其中的对象发生了变更，由于是异步写新库，所以实际的执行流程可能如下：
老库 insert(list)
list.clear()
新库 insert(list)
