跳至主要內容

MySQL单表有10亿数据如何做迁移

Java大神2026/4/14大约 20 分钟场景设计题MySQL单表有10亿数据如何做迁移

MySQL单表有10亿数据如何做迁移

面试官您好，针对这个问题我会从核心原则、分阶段执行、工具选型、风险控制四个维度展开，整体思路是先全量后增量，双写兜底，灰度切换，可回滚，这也是互联网大厂处理 TB 级数据迁移的标准方案

迁移核心三原则（面试官最看重的底线）

  • ✅ 数据零丢失：所有变更必须可追溯，最终一致性保证
  • ✅ 最小停机时间：业务无感知或秒级停机
  • ✅ 可快速回滚：任何阶段出问题都能 1 分钟切回旧库

整体迁移流程图

分阶段详细执行（核心技术点）

3.1 迁移前：准备与评估（决定成败的阶段 ⚠️）

1.数据体检

  • 统计：行数、平均行长度、大字段 (TEXT/BLOB) 占比、索引总大小
  • 分析：热点数据分布、主键连续性、是否有唯一键冲突风险

2.目标库预配置

  • 硬件：CPU / 内存 / 磁盘 IO 不低于源库，预留 1.5 倍源库空间
  • 参数临时调优：innodb_flush_log_at_trx_commit=2、sync_binlog=0
  • 关键操作：预建表结构只建主键索引（全量完成后再建二级索引，速度提升 5-10 倍）

3.监控与预案

  • 监控大盘：迁移速度、延迟、QPS、磁盘使用率、连接数
  • 回滚方案：保留旧库 7 天数据，双写机制兜底

3.2 迁移中：全量 + 增量双阶段执行

全量迁移工具选型对比表

工具适用场景优点缺点10 亿数据推荐指数
mysqldump小表 (<1000 万)简单易用锁表、单线程、极慢⭐
xtrabackup同版本同引擎物理迁移速度快、不锁表不支持异构、不能过滤数据⭐⭐⭐
DataX异构数据源全量迁移多线程、支持过滤、可切分任务无增量能力⭐⭐⭐⭐⭐
CanalMySQL 增量同步阿里开源、成熟稳定、低延迟仅支持 MySQL⭐⭐⭐⭐⭐
Debezium多数据源 CDC生态好、支持 Kafka部署复杂⭐⭐⭐⭐

推荐方案：用 DataX 按主键范围切分 100-200 个并行任务做全量迁移，速度可达 10 万行 / 秒以上，10 亿数据约 2-3 小时完成。

增量同步与双写

  • 基于Binlog ROW 格式做 CDC（必须提前确认 binlog_format=ROW）
  • 用 Canal 监听源库 Binlog，实时同步到目标库，延迟控制在 1 秒内
  • 全量完成后开启业务双写：先写旧库，再异步写新库，失败重试 3 次，保证最终一致

3.3 迁移后：验证与灰度切换

1. 数据一致性校验（三层校验）

  • 粗校验：SELECT TABLE_ROWS FROM information_schema.TABLES对比行数
  • 细校验：用pt-table-checksum批量校验，或按主键范围抽样 1% 数据逐行对比
  • 业务校验：核心接口跑回归测试，验证读写逻辑正确性

2. 灰度流量切换

  • 第 1 步：切 1% 流量到新库，观察 1 小时
  • 第 2 步：切 10% 流量，观察 4 小时
  • 第 3 步：切 50% 流量，观察 1 天
  • 第 4 步：切 100% 流量，稳定运行 3 天
  1. 最终下线：确认无问题后，关闭双写，下线旧库

高频踩坑与优化技巧 💡

踩坑点后果解决方案
提前建二级索引全量迁移速度慢 10 倍以上全量完成后用ALTER TABLE ALGORITHM=INPLACE批量建索引
Binlog 格式为 STATEMENT增量同步数据不一致提前修改binlog_format=ROW，重启 MySQL 生效
大字段导致单条记录过大迁移速度骤降、OOM大字段单独拆表，或迁移时过滤非必要历史数据
主键不连续导致数据倾斜部分 DataX 任务跑不完手动指定主键切分范围，均匀分配任务
目标库磁盘空间不足迁移到一半失败提前计算空间：源库数据量 ×1.5（含索引）

加分项延伸思考

  • 如果迁移同时需要分库分表：可以在 DataX 全量迁移时直接按分片规则写入目标分表，增量同步时也按分片规则路由
  • 如果允许凌晨停机：可以用 xtrabackup 物理备份恢复，停机时间 = 备份时间 + 恢复时间（约 1-2 小时）
  • 云环境优化：优先使用云厂商 DTS 工具（阿里云 DTS、腾讯云 CDB 迁移），支持自动全量 + 增量 + 切换，省心省力

核心代码实现（带技术亮点标注）

6.1 DataX 全量迁移配置（按主键切分 + 多线程并行）

技术亮点：主键范围手动切分解决数据倾斜、多通道并行提速、只迁移必要字段过滤大字段

{
  "job": {
    "setting": {
      "speed": {
        "channel": 16, // 按CPU核心数设置，16核=16通道，10万行/秒
        "byte": 104857600 // 单通道每秒100MB
      },
      "errorLimit": {
        "record": 0, // 零错误容忍，有一条失败就终止
        "percentage": 0.02
      }
    },
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "username": "root",
            "password": "xxx",
            "column": ["id", "user_id", "order_no", "amount", "create_time"], // 过滤TEXT/BLOB大字段
            "connection": [
              {
                "jdbcUrl": ["jdbc:mysql://old-db:3306/order_db?useSSL=false"],
                // 手动切分主键范围，解决数据倾斜（核心优化）
                "querySql": [
                  "SELECT id,user_id,order_no,amount,create_time FROM t_order WHERE id >= 1 AND id < 50000000",
                  "SELECT id,user_id,order_no,amount,create_time FROM t_order WHERE id >= 50000000 AND id < 100000000",
                  // ... 共切分200个范围，每个范围500万行
                ]
              }
            ]
          }
        },
        "writer": {
          "name": "mysqlwriter",
          "parameter": {
            "username": "root",
            "password": "xxx",
            "column": ["id", "user_id", "order_no", "amount", "create_time"],
            "connection": [
              {
                "jdbcUrl": "jdbc:mysql://new-db:3306/order_db?useSSL=false&rewriteBatchedStatements=true",
                "table": ["t_order"]
              }
            ],
            "batchSize": 10000, // 批量提交1万行，速度提升3倍
            "preSql": ["SET FOREIGN_KEY_CHECKS=0"] // 临时关闭外键约束
          }
        }
      }
    ]
  }
}

6.2 Canal 增量同步客户端（批量处理 + 幂等性保证）

技术亮点：批量消费 Binlog、INSERT ... ON DUPLICATE KEY UPDATE 幂等、异常重试机制

@Component
public class CanalIncrementSyncService {
    @Autowired
    private JdbcTemplate jdbcTemplate;
    
    // 批量处理大小，平衡延迟与吞吐量
    private static final int BATCH_SIZE = 5000;
    // 幂等性SQL模板（核心：避免重复消费导致数据错误）
    private static final String UPSERT_SQL = "INSERT INTO t_order (id,user_id,order_no,amount,create_time,update_time) " +
            "VALUES (?,?,?,?,?,?) ON DUPLICATE KEY UPDATE " +
            "user_id=VALUES(user_id), order_no=VALUES(order_no), amount=VALUES(amount), update_time=VALUES(update_time)";

    @PostConstruct
    public void startCanalConsumer() {
        // 连接Canal Server
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("canal-server", 11111), "example", "", "");
        connector.connect();
        connector.subscribe("order_db.t_order"); // 只订阅目标表

        while (true) {
            Message message = connector.getWithoutAck(10000); // 批量获取1万条
            long batchId = message.getId();
            List<CanalEntry.Entry> entries = message.getEntries();
            
            if (batchId == -1 || entries.isEmpty()) {
                try { Thread.sleep(100); } catch (InterruptedException e) {}
                continue;
            }

            try {
                processEntries(entries);
                connector.ack(batchId); // 处理成功才ACK
            } catch (Exception e) {
                log.error("增量同步失败，batchId:{}", batchId, e);
                connector.rollback(batchId); // 失败回滚，下次重新消费
            }
        }
    }

    private void processEntries(List<CanalEntry.Entry> entries) {
        List<Object[]> batchArgs = new ArrayList<>(BATCH_SIZE);
        
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    // 处理INSERT/UPDATE/DELETE，统一用UPSERT保证幂等
                    List<CanalEntry.Column> columns = rowData.getAfterColumnsList();
                    Object[] args = new Object[6];
                    args[0] = Long.parseLong(getColumnValue(columns, "id"));
                    args[1] = Long.parseLong(getColumnValue(columns, "user_id"));
                    args[2] = getColumnValue(columns, "order_no");
                    args[3] = new BigDecimal(getColumnValue(columns, "amount"));
                    args[4] = Timestamp.valueOf(getColumnValue(columns, "create_time"));
                    args[5] = Timestamp.valueOf(getColumnValue(columns, "update_time"));
                    batchArgs.add(args);

                    // 批量提交
                    if (batchArgs.size() >= BATCH_SIZE) {
                        jdbcTemplate.batchUpdate(UPSERT_SQL, batchArgs);
                        batchArgs.clear();
                    }
                }
            }
        }
        
        // 提交剩余数据
        if (!batchArgs.isEmpty()) {
            jdbcTemplate.batchUpdate(UPSERT_SQL, batchArgs);
        }
    }

    private String getColumnValue(List<CanalEntry.Column> columns, String columnName) {
        for (CanalEntry.Column column : columns) {
            if (column.getName().equals(columnName)) {
                return column.getValue();
            }
        }
        return null;
    }
}

6.3 业务双写实现（异步 + 重试 + 兜底）

技术亮点：异步线程池隔离、失败重试 3 次、Canal 兜底最终一致

@Service
public class OrderService {
    @Autowired
    private OrderMapper oldOrderMapper;
    @Autowired
    private OrderMapper newOrderMapper;
    // 独立线程池，隔离双写逻辑，不影响主业务
    @Qualifier("doubleWriteExecutor")
    private ThreadPoolTaskExecutor doubleWriteExecutor;

    @Transactional(rollbackFor = Exception.class)
    public void createOrder(Order order) {
        // 1. 先写旧库（主库），事务保证
        oldOrderMapper.insert(order);
        
        // 2. 异步写新库，失败重试3次
        doubleWriteExecutor.execute(() -> {
            try {
                newOrderMapper.insert(order);
            } catch (Exception e) {
                log.error("双写新库失败，orderId:{}", order.getId(), e);
                // 重试3次，仍失败则记录日志，由Canal兜底同步
                RetryTemplate.builder()
                        .maxAttempts(3)
                        .backoffOptions(new FixedBackOffOptions(1000, 2))
                        .build()
                        .execute(context -> newOrderMapper.insert(order));
            }
        });
    }
}

6.4 数据一致性校验代码（MD5 抽样对比）

技术亮点：按主键范围分页抽样、MD5 行级对比、高效发现不一致

@Service
public class DataConsistencyChecker {
    @Autowired
    private JdbcTemplate oldJdbcTemplate;
    @Autowired
    private JdbcTemplate newJdbcTemplate;

    // 抽样校验1%数据，按主键范围分1000批
    public boolean checkConsistency(long startId, long endId) {
        long batchSize = (endId - startId) / 1000;
        AtomicBoolean isConsistent = new AtomicBoolean(true);

        IntStream.range(0, 1000).parallel().forEach(i -> {
            long batchStart = startId + i * batchSize;
            long batchEnd = (i == 999) ? endId : batchStart + batchSize;
            
            // 计算源库和目标库该范围的MD5总和
            String oldMd5 = calculateMd5Sum(oldJdbcTemplate, batchStart, batchEnd);
            String newMd5 = calculateMd5Sum(newJdbcTemplate, batchStart, batchEnd);
            
            if (!oldMd5.equals(newMd5)) {
                log.error("数据不一致，范围:[{},{}]", batchStart, batchEnd);
                isConsistent.set(false);
                // 进一步逐行对比找出具体不一致的行
                findInconsistentRows(batchStart, batchEnd);
            }
        });

        return isConsistent.get();
    }

    private String calculateMd5Sum(JdbcTemplate jdbcTemplate, long startId, long endId) {
        String sql = "SELECT MD5(CONCAT(id,user_id,order_no,amount,create_time)) AS row_md5 " +
                "FROM t_order WHERE id >= ? AND id < ?";
        List<String> md5List = jdbcTemplate.queryForList(sql, String.class, startId, endId);
        return DigestUtils.md5Hex(String.join("", md5List));
    }
}

核心技术难点与解决方案（面试官必问）

技术难点问题本质解决方案优化效果
⚠️ 10 亿数据全量迁移速度慢 + 数据倾斜主键分布不均匀（如 UUID、自增断档）导致部分任务数据量过大；单条插入效率低1. 手动按主键范围切分任务，保证每个任务 500 万行
2. 开启 rewriteBatchedStatements 批量提交
3. 全量迁移时只建主键索引，二级索引后建		迁移速度从 1 万行 / 秒提升至 10-15 万行 / 秒，10 亿数据 2-3 小时完成
⚠️ 增量同步延迟过高（秒级以上）单条消费 Binlog；目标库写入压力大；Canal 参数不合理1. 批量消费 + 批量提交（5000 行 / 批）
2. 目标库临时调优：innodb_flush_log_at_trx_commit=2
3. 调整 Canal 参数：canal.instance.batchSize=10000		延迟稳定控制在100ms 以内，业务无感知
⚠️ 双写一致性与性能平衡同步双写会增加接口响应时间；异步双写可能丢失数据1. 异步双写 + 独立线程池隔离
2. 失败重试 3 次 + 日志记录
3. Canal 增量同步作为兜底，保证最终一致		接口响应时间增加 < 1ms，数据一致性 100%
⚠️ 大字段 (TEXT/BLOB) 迁移瓶颈单条记录过大导致网络 IO 和磁盘 IO 瓶颈；内存溢出1. 迁移时过滤非必要大字段
2. 大字段单独拆表，单独迁移
3. 开启 MySQL 压缩协议		迁移速度提升 5-10 倍，避免 OOM
⚠️ 迁移过程中源库性能影响全量迁移会占用源库大量 CPU 和 IO，影响线上业务1. 迁移时间选在业务低峰期（凌晨 2-6 点）
2. 限制 DataX 单通道速度（100MB / 秒）
3. 源库开启只读账号，避免写操作		源库 CPU 使用率 < 30%，业务 QPS 无明显下降
⚠️ 灰度切换时的流量控制与回滚流量突增导致新库雪崩；新库有问题无法快速回滚1. 按比例灰度切换：1%→10%→50%→100%
2. 网关层配置流量路由，支持一键切回
3. 保留旧库 7 天数据，双写机制持续到切换完成		回滚时间 < 1 秒，业务无感知
⚠️ 分库分表场景下的路由一致性迁移同时需要分库分表，数据路由错误导致数据丢失1. DataX 全量迁移时直接按分片规则写入目标分表
2. Canal 增量同步时也按相同分片规则路由
3. 校验时按分片维度分别校验		分库分表迁移零数据丢失，路由准确率 100%

面试加分金句（一句话体现实战经验）

"我之前在做电商订单表 12 亿数据迁移时，就是用这套方案，最终实现了零停机、零数据丢失，整个迁移过程业务完全无感知，唯一的影响是凌晨低峰期源库 CPU 短暂升到 28%。"

真实现场模拟面试

真实面试对话场景

面试官 👨‍💼：

“咱们有个 MySQL 单表，线上数据马上就要到 10 亿了，现在打算把它迁到一个新实例，或者换成新表结构。要求不能长时间停服，也不能把线上业务拖垮。如果是你，你会怎么设计这个迁移方案？”

候选人 😎：

“10 亿级别的大表迁移，最忌讳的就是想‘一把梭’，绝对不能直接 mysqldump 然后锁表导。我的总体思路就是九个字：化整为零、在线无锁、全量+增量。

核心原则就三条：

  • 利用自增主键或时间字段，把大表切成无数小段；
  • 先搬全量，再通过 binlog 增量追赶实时变化；
  • 全程不加锁、不产生大事务，绝不让线上业务感知到。”

面试官 🤔：

“嗯，思路很清晰。那你按步骤详细说说，第一步会做什么？”

候选人 🧑‍💻：

“第一步当然是评估 + 准备。

先摸清表的‘脾气’：有没有自增 ID？有没有时间索引？确认好迁移窗口和回滚方案。

然后在新库里把表结构建好，字符集、引擎这些必须和源库一致。而且这是做表分区的绝佳机会，比如直接在新表上按时间建分区，一步到位。

最关键一点：源表必须有能索引的切分条件，ID 最好，create_time 也可以，否则得想办法加一列。”

面试官 🧐：

“好，那具体怎么搬全量数据？10 亿行啊，总不能一条条 select 吧？”

候选人 💡：

“全量迁移我绝对不会手写脚本，直接用 Percona Toolkit 里的 pt-archiver，这个工具专干在线归档。

命令大概长这样：

pt-archiver \
  --source h=old_host,D=db,t=huge_table \
  --dest h=new_host,D=db,t=huge_table \
  --where "id >= 1 AND id < 50000000" \
  --limit 1000 --txn-size 1000 --no-delete \
  --progress 50000
  • 按 ID 范围 多进程并行 跑，每个进程认领一个区间，比如 5000 万一段；
  • --txn-size 控制事务大小，避免长事务把 undo log 撑爆或者主从延迟飞涨。
  • 原理是 WHERE id BETWEEN ... 走主键索引，RC/RR 隔离级别下都是一致性快照读，几乎不锁表。

我还会同时画一个流程图方便理解：

面试官 🚦：

“这个图很直观。那全量跑完，中间产生的增量数据怎么追？”

候选人 📍：

“全量完成后，我会立刻在源库执行 SHOW MASTER STATUS;，记下 File 和 Position，这就是增量同步的起点。

然后有三种主流方式追增量：

  • 云厂商 DTS（最简单）：设置源和目标，指定起始位点，自动完成全量+增量；
  • 自建 Canal：伪装成 MySQL 从库拉取 binlog，自己写 client 写入新库；
  • 临时从库法：把新库配成源库的 MySQL 从库，CHANGE MASTER TO 从记录的位点开始追，START SLAVE。

等增量延迟稳定到毫秒级，才敢谈下一步。”

面试官 ⚠️：

“那如果这张表没有自增 ID，也没有时间字段呢？你怎么办？”

候选人 🕳️：

“这确实是个大坑。如果没有自增 ID 也没时间索引，我会分情况：

  1. 业务允许加字段：那就立刻加一个自增 ID 或者 modified_time 索引，利用低峰期 Online DDL 工具（gh-ost 或 pt-osc）不锁表完成；
  2. 实在加不了：只能靠 LIMIT offset, size，但深分页会越来越慢，10 亿数据根本走不通。这时候就得跟业务方争取一个稍长的停机窗口，用 SELECT INTO OUTFILE + LOAD DATA 的方式暴力迁，但那不是‘在线’方案了。

所以我在前期评估时，就会逼着大家想办法造出合适的切分键，不然在线平滑迁移没戏。”

面试官 🔍：

“数据搬过去了，你怎么确认新旧库数据完全一致？”

候选人 ✅：

“不能靠感觉，必须上工具：

  • 首选 pt-table-checksum，在主库上生成校验值，再在新库（从库）上对比，连行级差异都能抓到；
  • 或者按分片对比 CRC32 / COUNT；
  • 业务侧再抽验核心用户或订单，三重保险。

只有校验全部通过，我心里才敢想切换。”

面试官 🚀：

“最后切换环节，怎么尽可能平滑？”

候选人 🤝：

“所有准备就绪后，选一个凌晨低峰期：

  1. 先短暂暂停源库写入（挂一个维护状态），等增量完全追平，延迟归零；
  2. 修改应用配置或数据库中间件，把流量切到新库；
  3. 验证核心接口，没问题后旧表保留 N 天作为回滚备份，同步链路也可以保留一段时间，万一出问题能快速切回。

这整个过程写操作暂停可能只有几秒钟，基本无感。”

面试官 😊：

“很好，整体方案很完整。最后用一句话总结下？”

候选人 🎯：

“一句话：全量分批 + binlog 增量 + 数据校验 + 秒级停写切换。工具有 pt-archiver、pt-table-checksum，配合 DTS/Canal，只要抓住有索引的切分键，10 亿数据在线迁移也能稳如老狗。😄”

面试官 👨‍💻：

“前面方案聊得挺细了。你能不能展示一下实际会用到的核心代码片段？比如你刚才说的并行分批导出、增量同步、数据校验，有什么技术亮点？另外，再帮我总结一下这个场景的技术难点以及你的解决方案。”

候选人 💻：

“好的，我直接把代码思路和亮点亮出来，然后给您梳理一下难点矩阵。”

🔥 1. 核心代码 & 技术亮点

🧩 亮点1：Java 多线程并行分批迁移（无锁，断点续传）

下面是一个简化但生产可用的迁移 Worker，亮点在于：

  • ✅ 分片并发：按 ID 范围切段，线程池控制并发度
  • ✅ 断点续传：记录每个分片的最大 ID，重启不丢进度
  • ✅ 流控保护：LIMIT 控制单次查询大小，BETWEEN 走索引
  • ✅ 批量写入：rewriteBatchedStatements=true，减少网络往返
// 分段任务
class SegmentTask {
    long startId;
    long endId;
    // getter/setter...
}

// 迁移主流程（伪代码）
ExecutorService pool = Executors.newFixedThreadPool(parallelism);
CompletionService<Void> completion = new ExecutorCompletionService<>(pool);

// 分段：每 50 万一个区间
long batchSize = 500_000;
for (long start = 1; start <= maxId; start += batchSize) {
    long end = Math.min(start + batchSize - 1, maxId);
    SegmentTask task = new SegmentTask(start, end);
    completion.submit(() -> {
        migrateSegment(task);
        return null;
    });
}

// 收集结果，处理异常重试
for (int i = 0; i < segments; i++) {
    Future<Void> future = completion.take();
    future.get(); // 检查异常，可重试
}
pool.shutdown();
// 核心迁移方法：一次处理一个 ID 范围
void migrateSegment(SegmentTask task) {
    long lastId = task.startId;
    while (lastId < task.endId) {
        // 从源库流式读取，游标查询，避免内存溢出
        List<Data> batch = sourceRepo.findByRange(lastId, 
                                  Math.min(lastId + LIMIT, task.endId));
        if (batch.isEmpty()) break;
        
        // 批量写入目标库（开启 rewriteBatchedStatements）
        targetRepo.batchInsert(batch);
        
        lastId = batch.get(batch.size() - 1).getId();
        // 记录进度到 Redis/文件，实现断点续传
        progressRecorder.update(task.startId, lastId);
    }
}

🎯 技术亮点：

  • findByRange 使用 SELECT ... WHERE id >= ? AND id < ? ORDER BY id ASC，利用主键索引，一次网络往返只拉 1000 行，避免客户端 OOM。
  • 每个分片记录进度，重启时从 progressRecorder 取出最后的 lastId 继续，10 亿数据迁移不怕中断。
  • 通过控制 parallelism 和 LIMIT，精确调整对源库的负载压力。

🔄 亮点2：基于 Binlog 的增量同步（Canal 客户端示例）

Canal 可以伪装成 MySQL 从库实时捞取 binlog，我们只关心需要的表和事件类型。

// Canal 客户端核心处理逻辑（简化版）
@EventHandler
public void onEvent(CanalEntry.Entry entry) {
    if (entry.getEntryType() != EntryType.ROWDATA) return;
    
    RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
    if (rowChange.getIsDdl()) return; // 忽略DDL
    
    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
        switch (rowChange.getEventType()) {
            case INSERT:
                List<Column> afterColumns = rowData.getAfterColumnsList();
                targetRepo.insert(buildEntity(afterColumns));
                break;
            case UPDATE:
                // 按主键更新
                targetRepo.updateByPk(buildEntity(rowData.getAfterColumnsList()));
                break;
            case DELETE:
                // 按主键删除
                targetRepo.deleteByPk(getPkValue(rowData.getBeforeColumnsList()));
                break;
        }
    }
}

🎯 技术亮点：

  • 顺序消费：单线程处理，保证与源库执行顺序一致，避免乱序导致数据不一致。
  • 幂等设计：INSERT 使用 INSERT ... ON DUPLICATE KEY UPDATE，即使重复消费也不出错。
  • 位点持久化：处理完一个 batch 后，将 binlog 位点写入 Redis/DB，重启从上次位点继续，增量不丢不重。

🔍 亮点3：数据一致性快速校验（CRC32 分片比对）

避免全表 count(*)，采用分片 CRC32 对比，高效且准确。

-- 在源库和新库同时执行，分片比较 CRC32 值
SELECT 
    (id DIV 100000) AS chunk,        -- 每 10 万行一个块
    COUNT(*) AS cnt,
    CRC32(GROUP_CONCAT(CRC32(
        CONCAT_WS('#', id, col1, col2, update_time)
    ))) AS checksum
FROM huge_table
WHERE id BETWEEN 1 AND 10000000
GROUP BY chunk
ORDER BY chunk;

🎯 技术亮点：

  • 利用 CONCAT_WS 和嵌套 CRC32 对整个块的行生成数字指纹，分片对比，秒级发现差异。
  • 对比双方结果集，checksum 不同则缩小分片范围进一步定位，比 pt-table-checksum 更轻量且不依赖 Percona 工具集。

📊 2. 技术难点 vs 解决方案矩阵

我用一个图表把核心痛点和应对策略串起来，更直观。👇

难点逐条拆解 📝

技术难点具体问题解决方案使用工具 / 技术点
海量数据迁移性能单线程、深分页会导致源库压力大且越跑越慢按索引分段多线程并行，BETWEEN 走主键，游标式读取pt-archiver、自定义Java线程池、LIMIT 控制
无合适切分键表没有自增ID或时间索引评估阶段必须加字段，利用 gh-ost 等在线DDL工具；实在不行则与业务协商停机窗口gh-ost、pt-osc
增量同步延迟全量+增量期间业务写入量大，新库一直追不平多线程并行消费binlog，合并微小事务，控制目标库写入速度；必要时暂停非核心业务写入Canal、DTS、临时从库
数据一致性校验10亿行数据逐行比对不现实分块计算CRC32指纹对比，或使用 pt-table-checksum自定义CRC32脚本、pt-table-checksum
切换瞬间停机时间完全停服切流可能导致分钟级不可用短时间暂停写入（秒级），等增量完全追上后切换，配合数据库中间件动态路由预发布配置、ProxySQL 等
大事务风险一次性提交大量数据导致长事务，undo爆增，主从延迟每批只提交1000行，--txn-size 严格控制pt-archiver 参数、手动提交循环

🧠 总结升华

“10亿级数据迁移，本质上是在性能、一致性、业务连续性三者间找平衡。代码上抓住分段并行、断点续传、幂等处理；策略上做好全量+增量+校验；难点通过工具链+索引设计+流控一一击破。这样即使数据再翻一倍，架构依然稳得住。”

面试官 😄：

“这个总结很好，代码亮点和难点都讲透了。看来你确实亲自趟过这些坑，很不错。”

候选人 🙏：

“谢谢！都是实战逼出来的经验。”

最近更新:
贡献者: 402304710
默认页脚
Copyright © 2026 Java大神