- VisualStudio2022插件的安装及使用-编程手把手系列文章
- pprof-在现网场景怎么用
- C#实现的下拉多选框,下拉多选树,多级节点
- 【学习笔记】基础数据结构:猫树
作者:来自 vivo 互联网服务器团队- Li Gang 。
本文介绍了一次 MySQL 数据迁移的流程,通过方案选型、业务改造、双写迁移最终实现了亿级数据的迁移.
预约业务是 vivo 游戏中心的重要业务之一。由于历史原因,预约业务数据表与其他业务数据表存储在同一个数据库中。当其他业务出现慢 SQL 等异常情况时,可能会直接影响到预约业务,从而降低系统整体的可靠性和稳定性。为了尽可能提高系统的稳定性和数据隔离性,我们迫切需要将预约相关数据表从原来的数据库中迁移出来,单独建立一个预约业务的数据库.
常见的迁移方案大致可以分为以下几类:
而预约业务有以下特点:
结合这些特点,我们再评估下上面的方案:
停机迁移方案需要停机,不适用于预约场景。预约场景存在不活跃的用户数据,如果用渐进式迁移方案的话很难迁移干净,可能还需要再写一个迁移任务来辅助完成迁移。而双写方案最大的优势在于每一步操作都可向上回滚,能尽可能的保证业务不出问题.
因此,最终选择的是双写方案。预约业务涉及到的读写场景多,每一个场景单独进行改造的成本大,采用 Mybatis 插件来实现迁移所需的双写等功能,可以有效降低改造成本.
这几步使用了公司提供的数据同步工具。全量同步基于 MySQLDump 实现;增量同步基于 binlog 实现;一致性校验通过在新老库各选一个分块,然后聚合列数据计算并对比其特征值实现.
引入了新库,那自然就需要在项目里新建数据源,并创建表对应的 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.
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<Object> paramValues = new ArrayList<>();
if (parameter instanceof Map) {
Map<String, Object> paramMap = (Map<String, Object>) 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 插件拦截老库的执行过程,实现迁移所需的读写数据源切换/新老库查询结果对比/先写老库再异步写新库等功能.
此时业务仍只读写老库.
第1步和第2步并没有严格的顺序要求,只要在切换为双写前做完第1步和第2步就好。 条件允许的情况下,全量+增量同步时应选择不对外提供服务的离线从库作为数据源,避免主从延迟等问题对线上业务造成影响.
老库追上新库后,对数据做一次全量校验,避免出现数据不一致的情况。此外还需要开启新老库查询结果对比开关,通过日志监控观察新老库的查询结果是否一致.
停止数据同步和切换双写之间必然有时间差,如果先开启双写再停止数据同步,则可能出现插入重复数据或数据被覆盖的情况。因此需要对数据同步工具和迁移插件进行改造,以处理数据异常的情况,但是这样改造需要处理的情况较多,改造成本较高。所以这里选择先停止同步,再切换到双写,中间丢失的数据使用对比&补偿任务恢复,由于此时仍然全量读老库,所以对业务不会有影响。需要注意的是,双写阶段的时间不应太长,只要确保新老库数据一致就应该前进到下一步.
这一步在实际操作过程中需要注意以下情况:
预约业务新库的主键 id 需要和老库保持一致,因此在迁移前检查了一遍业务代码,确保插入语句都用了 Mybatis 的 useGeneratedKeys 功能来返回 id ,这样插入新库时可以直接用设置好 id 的对象。但是这里有一个问题,批量插入时 Mybatis 自动设置的 id 和数据库生成的自增主键不一定完全一致,比如批量 insert ignore 和 on duplicate key update 语句.
这个问题和 useGeneratedKeys 的实现有关,代码可参考com.mysql.jdbc.StatementImpl#getGeneratedKeysInternal(long) 函数,以下是其执行逻辑:
举例来说,假设老库的某张表里有数据 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() 函数或者其中的对象发生了变更,由于是异步写新库,所以实际的执行流程可能如下:
这会导致新库执行操作时,传入的对象和老库执行操作时不一样,导致新老库数据不一致。建议在迁移前人为的确认业务逻辑,避免异步写入导致新老库数据不一致.
该对比&补偿任务有一个缺陷,其不能处理数据被删除的情况,如果老库里的数据被删除但是新库的数据删除失败,那使用更新时间区间就无法从老库查出这条数据,自然也无法进行对比&补偿.
双写期间,如果出现删老库成功但是删新库失败的情况会有日志告警,所以不会有问题。但是停止数据同步工具 → 开启双写开关这一过程中删除的数据无法补偿。不过大部分业务用的都是逻辑删除,只有一处用了物理删除,笔者在这一处添加了日志,如果切换过程中出现删除数据的日志,就需要手动进行补偿操作。实际操作过程中,开关的切换的耗时较短,只花了30秒左右,在这过程中没有打印删除数据的日志.
双写时,由于数据先写入老库再异步写入新库,因此新库的数据肯定会滞后于老库。如果将一部分读流量切换到新库上,就可能会在一些对延迟要求较高的业务场景中出现问题。对于这种场景,我们不能采用逐步切量的策略,只能同时切换读写开关,将其修改为只写老库+只读新库.
反向补偿是从新库补偿数据到老库,由于该任务是定时执行,开启后,新库和老库的数据会有 1~2 分钟的延迟,万一写新库的逻辑有问题,可以切回老库。至于为什么用反向补偿任务而不是使用先写新库再异步写老库的策略,是因为双写是用 MyBatis 插件实现的,插件代理的是 excutor 的 update 和 query 方法,如果异步写入老库,有可能会发生以下情况:
因此无法使用 Mybatis 插件来实现异步写老库.
停止反向补偿前,需要关注是否还有业务在读老库。观察一段时间,确认老库没有补偿任务以外的读写流量后,可以关闭补偿任务,清理迁移过程中产生的代码,清理老库数据.
在进行数据表迁移的过程中,虽然遇到了一些问题,但是制定的方案中每一步都有回退措施,即使出现问题也不会影响业务的正常运行。此外,笔者在迁移过程中对各种异常情况进行了监控,能及时发现并解决问题。如果其他业务需要进行类似的迁移,需要关注以下几个方面:
最后此篇关于MySQL亿级数据平滑迁移实战的文章就讲到这里了,如果你想了解更多关于MySQL亿级数据平滑迁移实战的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。
实战-行业攻防应急响应 简介: 服务器场景操作系统 Ubuntu 服务器账号密码:root/security123 分析流量包在/home/security/security.pcap 相
背景 最近公司将我们之前使用的链路工具切换为了 OpenTelemetry. 我们的技术栈是: OTLP C
一 同一类的方法都用 synchronized 修饰 1 代码 package concurrent; import java.util.concurrent.TimeUnit; public c
一 简单例子 1 代码 package concurrent.threadlocal; /** * ThreadLocal测试 * * @author cakin */ public class T
1. 问题背景 问题发生在快递分拣的流程中,我尽可能将业务背景简化,让大家只关注并发问题本身。 分拣业务针对每个快递包裹都会生成一个任务,我们称它为 task。task 中有两个字段需要
实战环境 elastic search 8.5.0 + kibna 8.5.0 + springboot 3.0.2 + spring data elasticsearch 5.0.2 +
Win10下yolov8 tensorrt模型加速部署【实战】 TensorRT-Alpha 基于tensorrt+cuda c++实现模型end2end的gpu加速,支持win10、
yolov8 tensorrt模型加速部署【实战】 TensorRT-Alpha 基于tensorrt+cuda c++实现模型end2end的gpu加速,支持win10、linux,
目录如下: 为什么需要自定义授权类型? 前面介绍OAuth2.0的基础知识点时介绍过支持的4种授权类型,分别如下: 授权码模式 简化模式 客户端模式 密码模式
今天这篇文章介绍一下如何在修改密码、修改权限、注销等场景下使JWT失效。 文章的目录如下: 解决方案 JWT最大的一个优势在于它是无状态的,自身包含了认证鉴权所需要的所有信息,服务器端
前言 大家好,我是捡田螺的小男孩。(求个星标置顶) 我们日常做分页需求时,一般会用limit实现,但是当偏移量特别大的时候,查询效率就变得低下。本文将分四个方案,讨论如何优化MySQL百万数
前言 大家好,我是捡田螺的小男孩。 平时我们写代码呢,多数情况都是流水线式写代码,基本就可以实现业务逻辑了。如何在写代码中找到乐趣呢,我觉得,最好的方式就是:使用设计模式优化自己
我们先讲一些arm汇编的基础知识。(我们以armv7为例,最新iphone5s上的64位暂不讨论) 基础知识部分: 首先你介绍一下寄存器: r0-r3:用于函数参数及返回值的传递 r4-r6
一 同一类的静态方法都用 synchronized 修饰 1 代码 package concurrent; import java.util.concurrent.TimeUnit; public
DRF快速写五个接口,比你用手也快··· 实战-DRF快速写接口 开发环境 Python3.6 Pycharm专业版2021.2.3 Sqlite3 Django 2.2 djangorestfram
一 添加依赖 org.apache.thrift libthrift 0.11.0 二 编写 IDL 通过 IDL(.thrift 文件)定义数据结构、异常和接口等数据,供各种编程语言使用 nam
我正在阅读 Redis in action e-book关于semaphores的章节.这是使用redis实现信号量的python代码 def acquire_semaphore(conn, semn
自定义控件在WPF开发中是很常见的,有时候某些控件需要契合业务或者美化统一样式,这时候就需要对控件做出一些改造。 目录 按钮设置圆角
师父布置的任务,让我写一个服务练练手,搞清楚socket的原理和过程后跑了一个小demo,很有成就感,代码内容也比较清晰易懂,很有教育启发意义。 代码 ?
? 1 2
我是一名优秀的程序员,十分优秀!