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

{
  "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"] // 临时关闭外键约束
          }
        }
      }
    ]
  }
}

@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;
    }
}

@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));
            }
        });
    }
}

@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));
    }
}

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

// 分段任务
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);
    }
}

// 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;
        }
    }
}

-- 在源库和新库同时执行，分片比较 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;