业务背景

旧的产品线数据：

三大块：

（1）控台

商户、机具、服务商

（2）交易

（3）分润

---

新的 star 系统：

包含上面全部功能。

期望所有的数据从老系统迁移到新系统。

分步走的迁移策略

产品分布

为减低迁移风险，首先迁移一个最小的产品线 rpos。

---

数据继续切分

分步走：

（1）不切交易。只切基本数据。

首先正向迁移迁移基本信息，保障新控台保障所有旧的数据信息。

后续的商户进件、服务商进件、机具等全部走新的系统，数据落在新的系统。

但同时交易没切，为了保证新的用户在老系统正常交易，所以需要把数据反向写到老系统。

（2）分步骤切交易。

按照一定比例，比如 1%=>5%=>10%=>30%=>50%=>100%

逐步切过去，观察没有问题，尽可能降低对于交易的影响。

（3）最后统一把分润切走。

分润涉及到出账，风险最高。

业务梳理

异构

旧的产品线再跑，新的产品线也在跑。

新产品线业务需求在持续推进，旧的系统基本少有维护，除了一些必要的改造，大家的改动是越小越好。

正向：负责把旧系统（Oracle）的数据，变成新系统（mysql）新的设计的格式。

反向：负责把新系统的变动、新的进件，反向写到老系统。保障新用户、数据变化，在旧系统中交易正常、分润正常。

因为两个系统设计等是有差异的；相同的系统，不同的产品设计也存在差异；所以需要梳理两个系统的表设计和业务关系。

困难点

旧的系统，oracle 的表维护的文档很差，缺少字段注释、设计文档甚至是人员变动比较大。

整个链路涉及到的研发人员也比较多，需要同时问新旧两个系统的研发：

（1）商户

（2）机具

（3）渠道/代理

（4）交易

（5）分润

几个人整理了很久，整体非常的枯燥，后续只剩下我一个人在处理反向同步。

接近 100+ 页的梳理文档，几十张表。

编程部分

正向

第一次迁移正向是其他小伙伴的。

整体思路就是通过多线程的方式，把数据迁移到新的系统。

为了保障知道哪些是迁移数据，便于清理。

新的系统，表都会额外新增一个 migrate_batch 迁移批次信息。

反向-设计

反向的话，需要一个触发机制。

最简单直接的当然是让新系统每次变更，同时把变更通知给反向系统。

DTS

但是这个对于每一个系统都有开发量，于是采用了基于 mysql DTS 数据变更的监听方式，每一次表发生变更（insert/update/delete），都会触发一次变化，可以基于 DTS，把这个变化的消息放到 kafka 中，反向程序监听 kafka，直接进行相关的操作处理。

DTS 可以设置关心的具体表，进一步缩小关心的范围。

DTS 数据库变动数据格式：

{
"db": "databaseName", // 数据库名称
"opType": "UPDATE", // 操作类型
"recordTs": 1596006856, // 变化时间
"tb": "tableName",  //表名称
"beforeDataInfo":{}, // 原始的表信息
"dataInfo": {}  // 现在的表信息
}

如何不重复、不遗漏+如何保障数据的实时性

每次变化，都会有一个 recordTS，这个时间是精确到秒的。

然后老的系统，每一张表都新增了一个 recordTs 字段。

用于判断数据是否是最新的，超过 recordTs 则丢弃掉对应的消息。

为了更加准确，每次接收到触发的消息之后，基本采用反查业务只读库的方式。

数据完整性

有的时候，数据是基于多张表的。

可能收到一个变化的时候，此时新系统的数据并不完成，可以加一个校验。后续收到时，才进行反查处理。

数据失败补偿

DTS 可能会挂掉，kafka 不支持指定的消息重发。

所以，开放了一个基于 controller 可以触发的补偿任务，可以把失败的消息 json，重新触发一下。

代码实现层

最简单地方式，是基于 if else 判断，不过不利于代码拓展。

定义了基于 @TransferRoute 的注解，可以指定对应的 databaseName + dbName + prodId 进行路由。

prod 产品信息，从 table 中获取。

给每一个产品，添加了可以开关的配置，便于动态调整。

给每一张表，统一添加了是否消费的开关，便于调整。

加解密

不同系统对于手机号、身份证等敏感信息的加密策略不同，需要先解密，再加密。

测试验证

生产实战

准备工作

数据备份，在迁移前，把旧系统的数据全部备份下来。避免数据被反向污染，无从查问题。

提前准备好条数对比脚本，相关的准备工作，POS 机器等。

灰度

经过多轮生产迁移演练。

发现问题，修改调整。

正式迁移

把老系统的所有功能暂停，提前发公告。

夜里 12:00 开始（业务低峰期）

正向同步数据：商户+机具+代理+渠道

数据的验证，条数对比等。在新的系统，查看验证是否正常。

确认是否正向满足。

迁移完成后，要保证正向的数据 kafka 已经被丢弃的差不多，然后打开反向开关，测试反向的相关逻辑。

事后验证

把遇到的一些问题，尽可能的解决掉。

基本忙到早晨 9 点，等其他同事过来接班。

观察 2 天，如果正常，则不进行回滚操作。

补救刷库

后续使用过程中，发现缺失错误的数据，不断进行刷库处理。

1 年间，基本刷库 600+ 次。

成就

一年时间，顺利把 pos 系列迁移完成，且无重大问题。

保障新老系统的稳定运行。

==》High

踩坑指南

delete 操作，高危

开始的时候，delete 操作也是监听的。

后来做了一次表清空，差点把目标表全部清空掉，浪费了测试的时间。

所以把这个列为高危操作，基本和业务端约定好也没有删除的操作，所以忽略了 delete 操作。

一秒内并发，反向处理方式 & kafka 消息一直没有 ack，多次重复消费

recordTs 是一个精确到秒的时间戳。

有些表的信息，可能一秒内被多次更新，生产就出现了同一秒，数据被丢弃的情况。

核心问题在于时间不够精确，如果精确到毫秒就好了。

但是 recordTs 这个字段又没有办法，所以只能修改了当时表的 update_time 作为时间戳，修改了设置的精度。

测试环境验证过，上线发布之后，发现一个问题，kafka 消费卡主了。当时正值过年，夜里忙到了凌晨 4 点。

如何解决

问题的原因在于 kafka 的配置不合理。

kafka 的触发：生产表比较大，执行完成之后

kafka 有一个配置项，是批量消费，然后 ack 一个消息，当时设置的量比较大，而且 mer 表变动，涉及的流程很长，导致消费很慢，超时了。

然后 kafka 不断重试。

---

解决方式：首先把当前 mer 表的消息快速丢弃掉，保证其他的消息可以正常消费。

上线发布：修正对应的配置项，让每一次批量消费的数据量变少，降低重试次数为 3。

善后

为了尽可能的不影响用户，把 mer 表在事故发生之后，进行重新的 update_time=now() 更新，重新触发一次消费。

然后如果还有用户反应存在问题，让其尝试业务重新操作，或者帮忙刷库处理。

故障-扣费 重复收取

确认字段的时候，新系统和老系统的字段是对的。

但是其实二者的日期格式不同，一个是 yyyyMMdd 的日期，一个是 yyyyMMddHHmmss 的日期。

结果老系统进行交易，通知更新新系统（不是我参与，不知道），导致新系统的表变化，数据反向，覆盖了老系统的数据，导致格式错误。

判断是否已经付费时，格式异常，导致需要重新缴费（老系统的健壮性）。

最核心的问题，还是反向的值错误了。

问题-消费之后机具需要重新签到

和上面类似的原因。

性能问题-时间成本问题

夜里正向迁移的时候，虽然是一次性的。但是耗时非常的重要。

希望尽可能的保留更多的时间，用于容错+验证。

---

以前商户/机具信息在正向同步的时候，是没有问题的，因为数据量还好，大概几十万，一个小时还能接受。

第二次同步的时候，迁移的产品比较多，数据量基本接近原来的十多倍，时间过长，无法接收。

---

2H=>20H 在夜里迁移是不可接受的。

尝试了单机提升配置，加线程，不过效果不是很好。

就把以前的方式否决了。

v1-基于 DTS

当时选择了旧的 oracle 创建一张新的表，和新的系统表结构一致，加上唯一索引。

这样，DTS 就可以把 oracle 的表，直接同步到 mysql，且根据唯一索引进行更新。

所以耗时就变成了两部分：

1）把数据同步到 oracle 中的新表。

2）DTS 把新表的数据，同步到目标表。

第一部分，采用以前的代码方案被否决了。因为商户+机具数据量很大。查出来，处理完，很慢。

当时发现一种方式，oralce 的 merge into 其实很快，不存在则插入，存在则更新。

几分钟，可以把一张大表，直接处理完成，且是一个完整的事务，要么同时成功，要么同时失败。

所以就是往这个方向努力。

痛苦之处

merge into 纯脚本，需要各种 oracle 的 sql 编写。

1）以前的代码方案被推翻，需要大量的时间+精力重新实现+测试

2. oracle 的 sql 能力自然没有代码这么灵活，比如加密机的调用等就无法处理，需要额外跑批补充。

3）对数据的要求更高

代码可以取到结果之后做一些处理，sql 有时候关联存在问题，就不行。

所以当时又做了一些数据的清理。

v2-基于时间

后来考虑到时间的问题。

采用了提前跑全量，然后跑增量的方式。

脚本中加入时间的范围限制。

1）全量，大于一个很小的时间。

2）增量，大于全量之前一点的时间，避免数据遗漏。

3）测试脚本，如果是测试的话，可以指定对应的数据，其他保持一致。

然后全量提前一周跑，分成 3 个晚上，每次处理几千万的数据。

增量在正式迁移的晚上，停止所有流量入口之后。

此时增量的数据量不是很大，且当晚进行了一定的提前。

可以优化的地方

全量+增量的思想

提前把全量迁移过去，然后再根据变更时间，把增量的迁移过去。

这种方式，在做大批量的数据迁移非常有用，可以节约很多时间，更加从容。

如果采用类似的思想，其实先把需要迁移的数据用程序提前跑，然后正式的时候，再跑增量，也未尝不可。

水平扩展

单个机器的性能，总归是存在瓶颈的。

不同的商户、机具之间，其实是互相不影响的。

所以其实可以通过一个生产者+多个消费者。把消息直接均摊到每一个消费者上面，比如 5 台机器，一台 10 个线程，也可以大大的提升同步效率。

只不过当时被单机-多线程的方式限制了。

收获

大型系统的迁移，非常的消耗时间和精力。

需要多个部分的配合协作：

（1）运营

安抚用户，发布公告

（2）产品

功能验收

（3）DBA

提供数据库层的技术支持

（4）研发+测试

配合研发+整体的测试流程

（5）项目带头人

需要有一个人把这些事情做好，流程顺序非常重要。