使用任务表,实现两个数据库表数据迁移

步骤概述

1. 创建任务表结构
2. 编写实体类和Mapper接口
3. 编写迁移服务类
4. 处理数据和计算/转换
5. 错误处理与恢复

1. 创建任务表结构

CREATE TABLE migration_tasks (
    id INT AUTO_INCREMENT PRIMARY KEY,
    oracle_table_name VARCHAR(255) NOT NULL,
    mysql_table_name VARCHAR(255) NOT NULL,
    status VARCHAR(50) NOT NULL,
    start_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    end_time TIMESTAMP
);

2. 编写实体类和Mapper接口

迁移任务实体类 MigrationTask.java

public class MigrationTask {
    private Long id;
    private String oracleTableName;
    private String mysqlTableName;
    private String status;
    private LocalDateTime startTime;
    private LocalDateTime endTime;
    
    // 省略 getter 和 setter 方法
}

Mapper接口 MigrationTaskMapper.java

@Mapper
public interface MigrationTaskMapper {
    void insert(MigrationTask task);
    void update(MigrationTask task);
    MigrationTask findById(Long id);
}

3. 编写迁移服务类

迁移服务类 MigrationService.java

@Service
@Transactional
public class MigrationService {

    @Autowired
    private MigrationTaskMapper taskMapper;

    @Autowired
    private OracleDataMapper oracleDataMapper;

    @Autowired
    private MySQLDataMapper mysqlDataMapper;

    public void migrateTable(String oracleTableName, String mysqlTableName) {
        // 记录任务开始
        MigrationTask task = new MigrationTask();
        task.setOracleTableName(oracleTableName);
        task.setMysqlTableName(mysqlTableName);
        task.setStatus("RUNNING");
        task.setStartTime(LocalDateTime.now());
        taskMapper.insert(task);

        int pageSize = 1000; // 每页大小
        int page = 0;

        try {
            do {
                List<Object[]> data = getDataFromOracle(oracleTableName, page, pageSize);

                // 处理数据
                processData(data);

                // 批量插入到MySQL
                batchInsertIntoMySQL(mysqlTableName, data);

                page++;
            } while (hasMoreDataFromOracle(oracleTableName, page, pageSize));

            // 更新任务状态为完成
            task.setStatus("COMPLETED");
            task.setEndTime(LocalDateTime.now());
            taskMapper.update(task);
        } catch (Exception e) {
            // 发生错误时记录错误状态
            task.setStatus("ERROR");
            task.setEndTime(LocalDateTime.now());
            taskMapper.update(task);
            throw new RuntimeException("Migration failed: " + e.getMessage(), e);
        }
    }

    private List<Object[]> getDataFromOracle(String tableName, int page, int pageSize) {
        return oracleDataMapper.getData(tableName, page, pageSize);
    }

    private void processData(List<Object[]> data) {
        // 这里可以对数据进行计算或转换
        for (Object[] row : data) {
            // 例如：对某列进行计算
            // row[0] = (Integer)row[0] * 2;
        }
    }

    private void batchInsertIntoMySQL(String tableName, List<Object[]> data) {
        mysqlDataMapper.batchInsert(tableName, data);
    }

    private boolean hasMoreDataFromOracle(String tableName, int page, int pageSize) {
        // 实现根据页码和页大小判断是否还有更多数据
        return true; // 需要根据实际情况实现
    }
}

4. 数据库操作类

OracleDataMapper.java

@Mapper
public interface OracleDataMapper {
    List<Object[]> getData(@Param("tableName") String tableName, 
                           @Param("page") int page, 
                           @Param("pageSize") int pageSize);
}

MySQLDataMapper.java

@Mapper
public interface MySQLDataMapper {
    void batchInsert(@Param("tableName") String tableName, 
                     @Param("data") List<Object[]> data);
}

5. 错误处理与恢复

在发生错误时，可以通过以下方式进行恢复：

• 记录错误状态：在任务表中记录错误状态，并记录错误消息和时间。

• 手动恢复：根据错误信息分析失败原因，修复问题后可以手动重启迁移任务。通过任务表中的记录，可以找到上次失败的位置，从而避免数据的重复插入或丢失。

总结

通过以上步骤，您可以使用Spring Boot和MyBatis实现从Oracle到MySQL的数据迁移，并在迁移过程中进行数据计算或转换。任务表的使用可以有效地管理迁移任务的状态，确保数据迁移的可靠性和一致性，同时提供了错误处理和人工恢复的机制。