我来为您提供一个完整的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("已回滚");
}
}
关键要点
- 安全性：在迁移前做好数据备份
- 一致性：保证数据迁移过程中的一致性
- 性能：使用批量处理提升迁移效率
- 监控：实时监控迁移进度和异常
- 回滚：提供完整的回滚机制
这个案例展示了完整的系统迁移方案，包括数据迁移、服务迁移、监控和配置管理等核心功能。
