Java案例如何跨表迁移数据？

Java案例：高效实现跨表数据迁移的完整指南

目录导读

1. 跨表数据迁移的核心场景与挑战
2. 技术方案选型：JDBC vs ORM框架
3. 实战案例：基于Spring Boot的跨表迁移
4. 性能优化策略与异常处理
5. 数据一致性保障与事务管理
6. 常见问题与解决方案（FAQ）

跨表数据迁移的核心场景与挑战

1 业务场景举例

• 数据归档：将一年前的订单数据从orders表迁移到orders_archive表
• 表结构拆分：将用户信息从单表拆分为user_base和user_extend
• 数据库迁移：从MySQL迁移到PostgreSQL过程中需要跨库迁移
• ETL处理：将多个源表数据整合到目标数据仓库表

2 技术挑战

根据搜索引擎综合经验,跨表迁移最常见的坑包括：

• 数据一致性：迁移过程中源表被修改导致数据不一致
• 性能瓶颈：百万级数据逐条插入导致超时
• 编码问题：字符集不一致导致乱码
• 主键冲突：自增主键在目标表重置
• 外键依赖：父子表迁移顺序错误

3 数据迁移的ACID原则

在Java实现跨表迁移时,必须考虑事务的四个特性：

• 原子性：要么全部迁移成功，要么全部回滚
• 一致性：迁移前后数据约束保持不变
• 隔离性：迁移过程不影响其他并发操作
• 持久性：迁移完成的数据不会丢失

技术方案选型：JDBC vs ORM框架

1 方案对比表

2 为什么推荐JDBC+批处理？

从搜索引擎优化角度看，企业级项目90%的数据迁移场景使用JDBC原生方案,原因如下：

• 避免ORM框架带来的额外内存开销
• 直接控制SQL执行参数，更快定位性能问题
• 支持数据库原生优化（如Oracle的FORALL）

3 一个典型的JDBC迁移流程

// 伪代码展示核心逻辑
Connection sourceConn = getSourceConnection();
Connection targetConn = getTargetConnection();
try {
    sourceConn.setAutoCommit(false);
    targetConn.setAutoCommit(false);
    String selectSql = "SELECT * FROM source_table WHERE create_time < ?";
    PreparedStatement ps = sourceConn.prepareStatement(selectSql);
    ps.setTimestamp(1, boundaryTime);
    ResultSet rs = ps.executeQuery();
    String insertSql = "INSERT INTO target_table (col1, col2) VALUES (?, ?)";
    PreparedStatement insertPs = targetConn.prepareStatement(insertSql);
    int batchSize = 500;
    int count = 0;
    while (rs.next()) {
        insertPs.setString(1, rs.getString("col1"));
        insertPs.setInt(2, rs.getInt("col2"));
        insertPs.addBatch();
        count++;
        if (count % batchSize == 0) {
            insertPs.executeBatch();
            targetConn.commit();
        }
    }
    insertPs.executeBatch();
    targetConn.commit();
} catch (Exception e) {
    sourceConn.rollback();
    targetConn.rollback();
    throw e;
}

实战案例：基于Spring Boot的跨表迁移

1 案例背景

一个电商系统需要将order_2023表的数据迁移到order_archive表,要求：

• 源表约200万条记录
• 迁移时间窗口为凌晨2:00-4:00
• 需保留历史数据的完整性

2 代码实现（核心类）

步骤1：定义迁移配置类

@Component
@ConfigurationProperties(prefix = "migration")
public class MigrationConfig {
    private int batchSize = 500;
    private int threadPoolSize = 4;
    private String sourceTable = "order_2023";
    private String targetTable = "order_archive";
    private String dateColumn = "create_time";
    // getter/setter省略
}

步骤2：实现数据读取与分批处理

@Service
public class TableMigrationService {
    @Autowired
    private JdbcTemplate jdbcTemplate;
    public void migrateByPagination(int pageSize) {
        long minId = getMinId();
        long maxId = getMaxId();
        long currentId = minId;
        while (currentId < maxId) {
            long nextId = currentId + pageSize;
            String sql = "SELECT * FROM " + sourceTable + 
                         " WHERE id BETWEEN ? AND ?";
            List<Map<String, Object>> rows = jdbcTemplate.queryForList(sql, currentId, nextId);
            batchInsert(rows);
            currentId = nextId;
        }
    }
    private void batchInsert(List<Map<String, Object>> rows) {
        String sql = "INSERT INTO " + targetTable + 
                     " (id, order_no, user_id, amount, create_time) VALUES (?,?,?,?,?)";
        jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement ps, int i) throws SQLException {
                Map<String, Object> row = rows.get(i);
                ps.setLong(1, (Long) row.get("id"));
                ps.setString(2, (String) row.get("order_no"));
                ps.setLong(3, (Long) row.get("user_id"));
                ps.setBigDecimal(4, (BigDecimal) row.get("amount"));
                ps.setTimestamp(5, (Timestamp) row.get("create_time"));
            }
            @Override
            public int getBatchSize() {
                return rows.size();
            }
        });
    }
}

步骤3：添加事务回滚机制

@Transactional(rollbackFor = Exception.class)
public void executeMigrationWithTransaction() {
    // 使用编程式事务控制
    TransactionDefinition def = new DefaultTransactionDefinition();
    TransactionStatus status = transactionManager.getTransaction(def);
    try {
        // 执行迁移逻辑
        transactionManager.commit(status);
    } catch (Exception e) {
        transactionManager.rollback(status);
        log.error("Migration failed, rolled back", e);
        throw new BusinessException("数据迁移失败，已回滚");
    }
}

3 问答环节

问：如果目标表已有数据，如何处理主键冲突？ 答：推荐两种策略：

1. 时间戳标记：在目标表增加migrated_at字段，通过查询条件过滤
2. UPSERT操作：使用数据库自带的ON DUPLICATE KEY UPDATE（MySQL）或MERGE（Oracle）

问：迁移过程中如何监控进度？ 答：在Redis中记录当前迁移的ID范围,配合定时任务生成日志：

redisTemplate.opsForValue().set("migration:progress", currentId);
// 启动一个线程每10秒打印进度

性能优化策略与异常处理

1 大数据量迁移的5个关键优化点

2 异常处理最佳实践

public class MigrationExceptionHandler {
    // 1. 记录失败的主键范围
    private List<Long> failedIds = new ArrayList<>();
    // 2. 重试机制（退避算法）
    public void retryOnFailure(Runnable task, int maxRetries) {
        int attempts = 0;
        while (attempts < maxRetries) {
            try {
                task.run();
                return;
            } catch (DataAccessException e) {
                attempts++;
                if (attempts >= maxRetries) {
                    throw new MigrationException("重试失败", e);
                }
                Thread.sleep(1000 * attempts); // 指数退避
            }
        }
    }
    // 3. 失败数据导出到文件
    public void exportFailedIds() {
        // 将failedIds写入CSV文件，便于人工处理
    }
}

3 问答环节

问：为什么使用executeBatch()后速度反而变慢？ 答：常见原因：

1. 批处理大小不当：MySQL推荐500-1000条，PostgreSQL推荐100-200条
2. SQL语句参数过多：超过数据库参数限制（如MySQL的max_allowed_packet）
3. 未关闭预编译语句缓存：使用PreparedStatement但未调用clearBatch()

问：如何防止迁移期间系统OOM？ 答：采用流式查询,避免一次性加载所有数据到内存：

// 使用游标方式（MySQL JDBC需要设置useCursorFetch=true）
statement.setFetchSize(Integer.MIN_VALUE); // 流式读取
ResultSet rs = statement.executeQuery();
while (rs.next()) {
    // 逐条处理，不保留完整结果集
}

数据一致性保障与事务管理

1 采用最终一致性方案

对于海量数据，强事务可能无法满足性能要求，推荐最终一致性+补偿机制：

• 记录迁移日志表（migration_log）
• 启动定时任务检查目标表与源表的数据差异
• 对差异数据执行增量同步

2 使用数据库的MVCC特性

-- 在源库创建快照，确保迁移期间数据一致性
SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;
START TRANSACTION;
-- 执行数据读取
COMMIT;

3 分布式事务方案（跨库迁移）

当迁移涉及不同数据库实例时,推荐使用：

1. 本地消息表：在源库记录迁移事件，通过消息队列异步处理
2. Seata AT模式：通过全局事务ID协调两阶段提交
3. 可靠消息服务：使用RocketMQ的事务消息

4 校验机制设计

public class DataConsistencyVerifier {
    public boolean verify(Long minId, Long maxId) {
        long sourceCount = countSource(minId, maxId);
        long targetCount = countTarget(minId, maxId);
        if (sourceCount != targetCount) {
            // 定位差异数据
            List<Long> diffIds = findMissingIds(minId, maxId);
            log.warn("发现{}条差异数据: {}", diffIds.size(), diffIds);
            return false;
        }
        return true;
    }
}

常见问题与解决方案（FAQ）

Q1：迁移时出现Deadlock found when trying to get lock怎么办？

解决方案：

1. 按固定顺序访问表（如先源表后目标表）
2. 减少事务持有锁的时间
3. 使用SELECT ... FOR UPDATE NOWAIT避免等待

Q2：目标表字段比源表少如何映射？

处理策略：

• 使用INSERT INTO target (col1, col3) SELECT col1, col3 FROM source
• 在Java代码中显式指定需要插入的列

Q3：迁移过程中如何不影响线上业务？

推荐做法：

1. 使用READ UNCOMMITTED隔离级别（允许脏读,避免行锁）
2. 在从库执行SELECT操作
3. 利用数据库的pt-archiver工具（MySQL）

Q4：千亿级数据迁移的通用方案是什么？

企业级方案：

1. 使用分布式ETL工具（如Apache NiFi）
2. 基于Spark的批量处理框架
3. 采用增量+全量混合模式

Q5：如何选择迁移时间窗口？

经验总结：

• 分析数据库的慢查询日志，选择业务低谷期
• 设置max_execution_time避免影响其他请求
• 通过SET GLOBAL max_connections = 200临时提升连接数

跨表数据迁移是Java开发中的高频需求，本文从技术选型、代码实现、性能优化到一致性保障，提供了完整的解决方案，实际项目中，建议根据数据量大小（万级/百万级/亿级）选择不同的策略组合，如果大家遇到其他迁移场景,欢迎留言交流。