Java实现系统迁移案例

wen java案例 2026-08-04 186

我来为您提供一个完整的Java系统迁移案例,包括代码实现和详细说明。

Java实现系统迁移案例

系统迁移案例：从单体架构到微服务

项目背景

将传统单体应用迁移到Spring Cloud微服务架构,实现系统的可扩展性和维护性。

核心代码实现

1 数据迁移工具类

package com.example.migration.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.ResultSetMetaData;
import java.util.*;
/**
 * 历史数据迁移工具类
 */
@Component
public class DataMigrationUtil {
    private static final Logger logger = LoggerFactory.getLogger(DataMigrationUtil.class);
    private final DataSource sourceDataSource;
    private final DataSource targetDataSource;
    private final JdbcTemplate sourceJdbcTemplate;
    private final JdbcTemplate targetJdbcTemplate;
    private final int BATCH_SIZE = 1000;
    public DataMigrationUtil(DataSource sourceDataSource, DataSource targetDataSource) {
        this.sourceDataSource = sourceDataSource;
        this.targetDataSource = targetDataSource;
        this.sourceJdbcTemplate = new JdbcTemplate(sourceDataSource);
        this.targetJdbcTemplate = new JdbcTemplate(targetDataSource);
    }
    /**
     * 迁移全表数据
     * @param tableName 表名
     * @param query 查询语句
     * @throws Exception 迁移异常
     */
    public void migrateTable(String tableName, String query) throws Exception {
        logger.info("开始迁移表: {} 数据", tableName);
        // 获取源表所有数据
        List<Map<String, Object>> rows = sourceJdbcTemplate.queryForList(query);
        logger.info("源表 {} 共有 {} 条数据", tableName, rows.size());
        // 分批次迁移
        for (int i = 0; i < rows.size(); i += BATCH_SIZE) {
            int end = Math.min(i + BATCH_SIZE, rows.size());
            List<Map<String, Object>> batch = rows.subList(i, end);
            migrateBatch(tableName, batch);
            logger.info("已迁移 {}/{} 条数据", end, rows.size());
        }
        logger.info("完成迁移表: {} 数据", tableName);
    }
    /**
     * 迁移一批数据
     */
    @SuppressWarnings("unchecked")
    private void migrateBatch(String tableName, List<Map<String, Object>> rows) {
        if (rows.isEmpty()) return;
        Map<String, Object> firstRow = rows.get(0);
        List<String> columns = new ArrayList<>(firstRow.keySet());
        // 构建INSERT语句
        StringBuilder insertSQL = new StringBuilder()
            .append("INSERT INTO ").append(tableName).append(" (")
            .append(String.join(", ", columns))
            .append(") VALUES (");
        for (int i = 0; i < columns.size(); i++) {
            insertSQL.append(i == 0 ? "?" : ", ?");
        }
        insertSQL.append(")");
        // 批量插入
        targetJdbcTemplate.batchUpdate(insertSQL.toString(), new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement ps, int rowNum) throws SQLException {
                Map<String, Object> row = rows.get(rowNum);
                for (int i = 0; i < columns.size(); i++) {
                    Object value = row.get(columns.get(i));
                    ps.setObject(i + 1, value);
                }
            }
            @Override
            public int getBatchSize() {
                return rows.size();
            }
        });
    }
    /**
     * 迁移增量数据
     * @param tableName 表名
     * @param lastMigrateTime 上次迁移时间
     */
    public void migrateIncremental(String tableName, Date lastMigrateTime) throws Exception {
        String query = "SELECT * FROM " + tableName + " WHERE create_time > ?";
        List<Map<String, Object>> rows = sourceJdbcTemplate.queryForList(query, lastMigrateTime);
        if (!rows.isEmpty()) {
            migrateTable(tableName, query);
        }
    }
}

2 服务迁移器

package com.example.migration.service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.beans.factory.annotation.Autowired;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * 服务迁移管理器
 */
@Component
public class ServiceMigrationManager {
    private static final Logger logger = LoggerFactory.getLogger(ServiceMigrationManager.class);
    @Autowired
    private LegacyService legacyService;
    @Autowired
    private MicroService microService;
    private ExecutorService executor = Executors.newFixedThreadPool(4);
    /**
     * 执行服务迁移
     */
    public void executeMigration() {
        logger.info("开始执行服务迁移...");
        // 并行执行迁移
        CompletableFuture<Void> orderFuture = CompletableFuture.runAsync(() -> {
            migrateOrderData();
        }, executor);
        CompletableFuture<Void> userFuture = CompletableFuture.runAsync(() -> {
            migrateUserData();
        }, executor);
        CompletableFuture<Void> productFuture = CompletableFuture.runAsync(() -> {
            migrateProductData();
        }, executor);
        CompletableFuture<Void> inventoryFuture = CompletableFuture.runAsync(() -> {
            migrateInventoryData();
        }, executor);
        // 等待所有迁移完成
        CompletableFuture.allOf(orderFuture, userFuture, productFuture, inventoryFuture).join();
        logger.info("服务迁移完成");
    }
    /**
     * 迁移订单数据
     */
    private void migrateOrderData() {
        try {
            // 从旧系统获取订单数据
            List<Order> legacyOrders = legacyService.getLegacyOrders();
            // 转换并迁移数据
            List<NewOrder> newOrders = legacyOrders.stream()
                .map(OrderConverter::convertToNewOrder)
                .collect(Collectors.toList());
            // 插入到新系统
            microService.saveOrders(newOrders);
            logger.info("订单数据迁移完成，共迁移 {} 条", newOrders.size());
        } catch (Exception e) {
            logger.error("订单数据迁移失败", e);
        }
    }
    /**
     * 迁移用户数据
     */
    private void migrateUserData() {
        try {
            List<LegacyUser> legacyUsers = legacyService.getLegacyUsers();
            List<NewUser> newUsers = legacyUsers.stream()
                .map(UserConverter::convertToNewUser)
                .collect(Collectors.toList());
            microService.saveUsers(newUsers);
            logger.info("用户数据迁移完成，共迁移 {} 条", newUsers.size());
        } catch (Exception e) {
            logger.error("用户数据迁移失败", e);
        }
    }
    /**
     * 迁移产品数据
     */
    private void migrateProductData() {
        try {
            List<Product> products = legacyService.getCompatibleProducts();
            microService.saveProducts(products);
            logger.info("产品数据迁移完成，共迁移 {} 条", products.size());
        } catch (Exception e) {
            logger.error("产品数据迁移失败", e);
        }
    }
    /**
     * 迁移库存数据
     */
    private void migrateInventoryData() {
        try {
            List<Inventory> inventories = legacyService.getInventories();
            microService.saveInventories(inventories);
            logger.info("库存数据迁移完成，共迁移 {} 条", inventories.size());
        } catch (Exception e) {
            logger.error("库存数据迁移失败", e);
        }
    }
}

3 数据转换器

package com.example.migration.converter;
import com.example.legacy.model.LegacyOrder;
import com.example.legacy.model.LegacyUser;
import com.example.microservice.model.NewOrder;
import com.example.microservice.model.NewUser;
/**
 * 数据转换器
 */
public class DataConverters {
    /**
     * 旧订单转换为新订单
     */
    public static NewOrder convertToNewOrder(LegacyOrder legacyOrder) {
        return NewOrder.builder()
            .id(legacyOrder.getOrderId())
            .userId(legacyOrder.getCustomerId())
            .amount(legacyOrder.getTotalAmount())
            .status(mapOrderStatus(legacyOrder.getStatus()))
            .createdTime(legacyOrder.getOrderTime())
            .updatedTime(legacyOrder.getUpdateTime())
            .build();
    }
    /**
     * 旧用户转换为新用户
     */
    public static NewUser convertToNewUser(LegacyUser legacyUser) {
        return NewUser.builder()
            .id(legacyUser.getUserId())
            .username(legacyUser.getUserName())
            .email(legacyUser.getEmailAddress())
            .phone(legacyUser.getPhoneNumber())
            .status(mapUserStatus(legacyUser.getStatus()))
            .createdTime(legacyUser.getRegisterTime())
            .build();
    }
    /**
     * 映射订单状态
     */
    private static String mapOrderStatus(String legacyStatus) {
        switch (legacyStatus) {
            case "0": return "PENDING";
            case "1": return "PAID";
            case "2": return "SHIPPED";
            case "3": return "COMPLETED";
            case "4": return "CANCELLED";
            default: return "UNKNOWN";
        }
    }
    /**
     * 映射用户状态
     */
    private static String mapUserStatus(String legacyStatus) {
        switch (legacyStatus) {
            case "0": return "ACTIVE";
            case "1": return "INACTIVE";
            case "2": return "FROZEN";
            default: return "UNKNOWN";
        }
    }
}

4 迁移监控器

package com.example.migration.monitor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.concurrent.atomic.AtomicLong;
/**
 * 迁移监控器
 */
@Component
public class MigrationMonitor {
    private static final Logger logger = LoggerFactory.getLogger(MigrationMonitor.class);
    private final AtomicLong totalCount = new AtomicLong(0);
    private final AtomicLong successCount = new AtomicLong(0);
    private final AtomicLong failCount = new AtomicLong(0);
    private final Map<String, Long> tableProgress = new ConcurrentHashMap<>();
    /**
     * 记录成功迁移
     */
    public void recordSuccess(String tableName, int count) {
        successCount.addAndGet(count);
        totalCount.addAndGet(count);
        tableProgress.put(tableName, (long) count);
    }
    /**
     * 记录失败迁移
     */
    public void recordFail(String tableName, int count) {
        failCount.addAndGet(count);
        logger.error("数据表 {} 迁移失败 {} 条", tableName, count);
    }
    /**
     * 定时打印迁移进度
     */
    @Scheduled(fixedDelay = 60000)
    public void printProgress() {
        logger.info("========== 迁移进度报告 ==========");
        logger.info("总迁移数量: {}", totalCount.get());
        logger.info("成功数量: {}", successCount.get());
        logger.info("失败数量: {}", failCount.get());
        logger.info("成功率: {:.2f}%", 
            successCount.get() * 100.0 / totalCount.get());
        if (!tableProgress.isEmpty()) {
            logger.info("各表迁移进度: {}", tableProgress);
        }
        logger.info("===================================");
    }
    /**
     * 导出迁移报告
     */
    public MigrationReport generateReport() {
        return MigrationReport.builder()
            .total(totalCount.get())
            .success(successCount.get())
            .fail(failCount.get())
            .tableProgress(new HashMap<>(tableProgress))
            .successRate(successCount.get() * 100.0 / totalCount.get())
            .build();
    }
}

5 配置类

package com.example.migration.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
import javax.sql.DataSource;
/**
 * 迁移配置
 */
@Configuration
public class MigrationConfig {
    /**
     * 数据源配置
     */
    @Bean
    @ConfigurationProperties(prefix = "source.datasource")
    public DataSource sourceDataSource() {
        return DataSourceBuilder.create().build();
    }
    @Bean
    @ConfigurationProperties(prefix = "target.datasource")
    public DataSource targetDataSource() {
        return DataSourceBuilder.create().build();
    }
    /**
     * 事务管理器
     */
    @Bean
    public PlatformTransactionManager transactionManager(DataSource targetDataSource) {
        return new DataSourceTransactionManager(targetDataSource);
    }
    /**
     * 迁移线程池
     */
    @Bean
    public ThreadPoolTaskExecutor migrationTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("migration-");
        return executor;
    }
}

6 主程序

package com.example.migration;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.boot.CommandLineRunner;
import com.example.migration.service.ServiceMigrationManager;
import com.example.migration.util.DataMigrationUtil;
@SpringBootApplication
@EnableScheduling
public class SystemMigrationApplication implements CommandLineRunner {
    private final ServiceMigrationManager migrationManager;
    private final DataMigrationUtil dataMigrationUtil;
    public SystemMigrationApplication(ServiceMigrationManager migrationManager,
                                     DataMigrationUtil dataMigrationUtil) {
        this.migrationManager = migrationManager;
        this.dataMigrationUtil = dataMigrationUtil;
    }
    public static void main(String[] args) {
        SpringApplication.run(SystemMigrationApplication.class, args);
    }
    @Override
    public void run(String... args) throws Exception {
        System.out.println("开始系统迁移...");
        // 1. 预检查
        preCheck();
        // 2. 数据迁移
        migrateData();
        // 3. 服务迁移
        migrationManager.executeMigration();
        // 4. 后验证
        postVerification();
        System.out.println("系统迁移完成！");
    }
    /**
     * 预检查
     */
    private void preCheck() {
        System.out.println("执行迁移前检查...");
        // 检查目标系统可用性
        // 检查数据库连接
        // 校验数据完整性
    }
    /**
     * 数据迁移
     */
    private void migrateData() throws Exception {
        System.out.println("开始数据迁移...");
        // 迁移字典数据
        dataMigrationUtil.migrateTable("sys_dict", 
            "SELECT * FROM sys_dict WHERE del_flag = 0");
        // 迁移用户数据
        dataMigrationUtil.migrateTable("sys_user", 
            "SELECT * FROM sys_user WHERE del_flag = 0");
        // 迁移角色数据
        dataMigrationUtil.migrateTable("sys_role", 
            "SELECT * FROM sys_role WHERE del_flag = 0");
        System.out.println("数据迁移完成");
    }
    /**
     * 后验证
     */
    private void postVerification() {
        System.out.println("执行迁移后验证...");
        // 验证数据一致性
        // 验证API可用性
        // 验证系统功能
    }
}

配置文件 application.yml

spring:
  datasource:
    # 源数据库配置
    source:
      url: jdbc:mysql://old-server:3306/legacy_db
      username: old_user
      password: old_password
      driver-class-name: com.mysql.cj.jdbc.Driver
    # 目标数据库配置  
    target:
      url: jdbc:mysql://new-server:3306/microservice_db
      username: new_user
      password: new_password
      driver-class-name: com.mysql.cj.jdbc.Driver
# 迁移配置
migration:
  batch-size: 1000
  max-threads: 4
  enable-incremental: true
  incremental-interval: 60000
# 监控配置
monitoring:
  enable: true
  report-format: json
  save-report: true

使用说明

// 手动触发迁移
@RestController
@RequestMapping("/migration")
public class MigrationController {
    @PostMapping("/start")
    public ApiResponse startMigration() {
        // 启动异步迁移任务
        CompletableFuture.runAsync(() -> {
            migrationManager.executeMigration();
        });
        return ApiResponse.success("迁移任务已启动");
    }
    @GetMapping("/progress")
    public ApiResponse getProgress() {
        return ApiResponse.success(migrationMonitor.generateReport());
    }
    @PostMapping("/rollback")
    public ApiResponse rollback() {
        // 回滚操作
        return ApiResponse.success("已回滚");
    }
}