前言

数据迁移作为分库分表架构演进中的重要环节，直接影响着系统的可用性、数据一致性和业务连续性。通过合理的平滑迁移策略，能够在不影响业务运行的情况下完成数据迁移，确保系统的稳定运行。本文从数据迁移策略到平滑迁移方案，从基础实现到企业级应用，系统梳理数据迁移的完整解决方案。

一、数据迁移架构设计

1.1 数据迁移整体架构

1.2 平滑迁移策略架构

二、数据迁移策略实现

2.1 双写策略

2.1.1 双写策略服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
/**
 * 双写策略服务
 */
@Service
public class DualWriteStrategyService {

    @Autowired
    private SourceDatabaseService sourceDatabaseService;

    @Autowired
    private TargetDatabaseService targetDatabaseService;

    @Autowired
    private DataValidationService dataValidationService;

    @Autowired
    private MigrationConfigService migrationConfigService;

    /**
     * 双写数据
     */
    @Transactional
    public void dualWrite(String tableName, String operation, Map<String, Object> data) {
        try {
            // 1. 写入源数据库
            writeToSourceDatabase(tableName, operation, data);

            // 2. 写入目标数据库
            writeToTargetDatabase(tableName, operation, data);

            // 3. 数据校验
            validateDataConsistency(tableName, data);

        } catch (Exception e) {
            log.error("双写失败", e);
            throw new MigrationException("双写失败", e);
        }
    }

    /**
     * 写入源数据库
     */
    private void writeToSourceDatabase(String tableName, String operation, Map<String, Object> data) {
        try {
            switch (operation.toUpperCase()) {
                case "INSERT":
                    sourceDatabaseService.insert(tableName, data);
                    break;
                case "UPDATE":
                    sourceDatabaseService.update(tableName, data);
                    break;
                case "DELETE":
                    sourceDatabaseService.delete(tableName, data);
                    break;
                default:
                    throw new IllegalArgumentException("不支持的操作类型: " + operation);
            }
        } catch (Exception e) {
            log.error("写入源数据库失败", e);
            throw new MigrationException("写入源数据库失败", e);
        }
    }

    /**
     * 写入目标数据库
     */
    private void writeToTargetDatabase(String tableName, String operation, Map<String, Object> data) {
        try {
            switch (operation.toUpperCase()) {
                case "INSERT":
                    targetDatabaseService.insert(tableName, data);
                    break;
                case "UPDATE":
                    targetDatabaseService.update(tableName, data);
                    break;
                case "DELETE":
                    targetDatabaseService.delete(tableName, data);
                    break;
                default:
                    throw new IllegalArgumentException("不支持的操作类型: " + operation);
            }
        } catch (Exception e) {
            log.error("写入目标数据库失败", e);
            throw new MigrationException("写入目标数据库失败", e);
        }
    }

    /**
     * 数据校验
     */
    private void validateDataConsistency(String tableName, Map<String, Object> data) {
        try {
            // 获取主键
            String primaryKey = getPrimaryKey(tableName);
            Object primaryKeyValue = data.get(primaryKey);

            if (primaryKeyValue == null) {
                return;
            }

            // 从源数据库查询
            Map<String, Object> sourceData = sourceDatabaseService.queryByPrimaryKey(tableName, primaryKeyValue);

            // 从目标数据库查询
            Map<String, Object> targetData = targetDatabaseService.queryByPrimaryKey(tableName, primaryKeyValue);

            // 数据一致性校验
            boolean isConsistent = dataValidationService.validateDataConsistency(sourceData, targetData);

            if (!isConsistent) {
                log.warn("数据不一致: table={}, primaryKey={}", tableName, primaryKeyValue);
                // 记录不一致数据
                recordInconsistentData(tableName, primaryKeyValue, sourceData, targetData);
            }

        } catch (Exception e) {
            log.error("数据校验失败", e);
            // 不抛出异常，避免影响主流程
        }
    }

    /**
     * 获取主键
     */
    private String getPrimaryKey(String tableName) {
        // 从配置中获取主键信息
        return migrationConfigService.getPrimaryKey(tableName);
    }

    /**
     * 记录不一致数据
     */
    private void recordInconsistentData(String tableName, Object primaryKeyValue, 
            Map<String, Object> sourceData, Map<String, Object> targetData) {
        try {
            InconsistentDataRecord record = new InconsistentDataRecord();
            record.setTableName(tableName);
            record.setPrimaryKeyValue(primaryKeyValue);
            record.setSourceData(sourceData);
            record.setTargetData(targetData);
            record.setCreateTime(new Date());

            // 保存到不一致数据表
            saveInconsistentDataRecord(record);

        } catch (Exception e) {
            log.error("记录不一致数据失败", e);
        }
    }

    /**
     * 保存不一致数据记录
     */
    private void saveInconsistentDataRecord(InconsistentDataRecord record) {
        // 实现保存逻辑
    }

    /**
     * 批量双写
     */
    @Transactional
    public void batchDualWrite(String tableName, List<DualWriteOperation> operations) {
        try {
            // 批量写入源数据库
            batchWriteToSourceDatabase(tableName, operations);

            // 批量写入目标数据库
            batchWriteToTargetDatabase(tableName, operations);

            // 批量数据校验
            batchValidateDataConsistency(tableName, operations);

        } catch (Exception e) {
            log.error("批量双写失败", e);
            throw new MigrationException("批量双写失败", e);
        }
    }

    /**
     * 批量写入源数据库
     */
    private void batchWriteToSourceDatabase(String tableName, List<DualWriteOperation> operations) {
        try {
            for (DualWriteOperation operation : operations) {
                writeToSourceDatabase(tableName, operation.getOperation(), operation.getData());
            }
        } catch (Exception e) {
            log.error("批量写入源数据库失败", e);
            throw new MigrationException("批量写入源数据库失败", e);
        }
    }

    /**
     * 批量写入目标数据库
     */
    private void batchWriteToTargetDatabase(String tableName, List<DualWriteOperation> operations) {
        try {
            for (DualWriteOperation operation : operations) {
                writeToTargetDatabase(tableName, operation.getOperation(), operation.getData());
            }
        } catch (Exception e) {
            log.error("批量写入目标数据库失败", e);
            throw new MigrationException("批量写入目标数据库失败", e);
        }
    }

    /**
     * 批量数据校验
     */
    private void batchValidateDataConsistency(String tableName, List<DualWriteOperation> operations) {
        try {
            for (DualWriteOperation operation : operations) {
                validateDataConsistency(tableName, operation.getData());
            }
        } catch (Exception e) {
            log.error("批量数据校验失败", e);
            // 不抛出异常，避免影响主流程
        }
    }
}

2.2 读写分离策略

2.2.1 读写分离策略服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
/**
 * 读写分离策略服务
 */
@Service
public class ReadWriteSeparationStrategyService {

    @Autowired
    private SourceDatabaseService sourceDatabaseService;

    @Autowired
    private TargetDatabaseService targetDatabaseService;

    @Autowired
    private MigrationConfigService migrationConfigService;

    @Autowired
    private DataSyncService dataSyncService;

    /**
     * 读操作路由
     */
    public Object routeRead(String tableName, String sql, Object... params) {
        try {
            // 检查迁移状态
            MigrationStatus migrationStatus = migrationConfigService.getMigrationStatus(tableName);

            switch (migrationStatus) {
                case NOT_STARTED:
                    // 未开始迁移，从源数据库读取
                    return sourceDatabaseService.query(sql, params);

                case IN_PROGRESS:
                    // 迁移进行中，根据数据范围选择读取源
                    return routeReadDuringMigration(tableName, sql, params);

                case COMPLETED:
                    // 迁移完成，从目标数据库读取
                    return targetDatabaseService.query(sql, params);

                default:
                    throw new IllegalArgumentException("未知的迁移状态: " + migrationStatus);
            }

        } catch (Exception e) {
            log.error("读操作路由失败", e);
            throw new MigrationException("读操作路由失败", e);
        }
    }

    /**
     * 迁移过程中的读操作路由
     */
    private Object routeReadDuringMigration(String tableName, String sql, Object... params) {
        try {
            // 解析SQL获取查询条件
            QueryCondition condition = parseQueryCondition(sql, params);

            // 检查数据是否已迁移
            boolean isDataMigrated = dataSyncService.isDataMigrated(tableName, condition);

            if (isDataMigrated) {
                // 数据已迁移，从目标数据库读取
                return targetDatabaseService.query(sql, params);
            } else {
                // 数据未迁移，从源数据库读取
                return sourceDatabaseService.query(sql, params);
            }

        } catch (Exception e) {
            log.error("迁移过程中读操作路由失败", e);
            // 降级到源数据库
            return sourceDatabaseService.query(sql, params);
        }
    }

    /**
     * 写操作路由
     */
    @Transactional
    public void routeWrite(String tableName, String operation, Map<String, Object> data) {
        try {
            // 检查迁移状态
            MigrationStatus migrationStatus = migrationConfigService.getMigrationStatus(tableName);

            switch (migrationStatus) {
                case NOT_STARTED:
                    // 未开始迁移，写入源数据库
                    writeToSourceDatabase(tableName, operation, data);
                    break;

                case IN_PROGRESS:
                    // 迁移进行中，双写
                    dualWriteDuringMigration(tableName, operation, data);
                    break;

                case COMPLETED:
                    // 迁移完成，写入目标数据库
                    writeToTargetDatabase(tableName, operation, data);
                    break;

                default:
                    throw new IllegalArgumentException("未知的迁移状态: " + migrationStatus);
            }

        } catch (Exception e) {
            log.error("写操作路由失败", e);
            throw new MigrationException("写操作路由失败", e);
        }
    }

    /**
     * 迁移过程中的双写
     */
    private void dualWriteDuringMigration(String tableName, String operation, Map<String, Object> data) {
        try {
            // 检查数据是否已迁移
            String primaryKey = migrationConfigService.getPrimaryKey(tableName);
            Object primaryKeyValue = data.get(primaryKey);

            boolean isDataMigrated = dataSyncService.isDataMigrated(tableName, primaryKeyValue);

            if (isDataMigrated) {
                // 数据已迁移，写入目标数据库
                writeToTargetDatabase(tableName, operation, data);
            } else {
                // 数据未迁移，双写
                writeToSourceDatabase(tableName, operation, data);
                writeToTargetDatabase(tableName, operation, data);
            }

        } catch (Exception e) {
            log.error("迁移过程中双写失败", e);
            throw new MigrationException("迁移过程中双写失败", e);
        }
    }

    /**
     * 写入源数据库
     */
    private void writeToSourceDatabase(String tableName, String operation, Map<String, Object> data) {
        try {
            switch (operation.toUpperCase()) {
                case "INSERT":
                    sourceDatabaseService.insert(tableName, data);
                    break;
                case "UPDATE":
                    sourceDatabaseService.update(tableName, data);
                    break;
                case "DELETE":
                    sourceDatabaseService.delete(tableName, data);
                    break;
                default:
                    throw new IllegalArgumentException("不支持的操作类型: " + operation);
            }
        } catch (Exception e) {
            log.error("写入源数据库失败", e);
            throw new MigrationException("写入源数据库失败", e);
        }
    }

    /**
     * 写入目标数据库
     */
    private void writeToTargetDatabase(String tableName, String operation, Map<String, Object> data) {
        try {
            switch (operation.toUpperCase()) {
                case "INSERT":
                    targetDatabaseService.insert(tableName, data);
                    break;
                case "UPDATE":
                    targetDatabaseService.update(tableName, data);
                    break;
                case "DELETE":
                    targetDatabaseService.delete(tableName, data);
                    break;
                default:
                    throw new IllegalArgumentException("不支持的操作类型: " + operation);
            }
        } catch (Exception e) {
            log.error("写入目标数据库失败", e);
            throw new MigrationException("写入目标数据库失败", e);
        }
    }

    /**
     * 解析查询条件
     */
    private QueryCondition parseQueryCondition(String sql, Object... params) {
        try {
            QueryCondition condition = new QueryCondition();

            // 简单的SQL解析，实际应用中需要更复杂的解析
            if (sql.contains("WHERE")) {
                String whereClause = sql.substring(sql.indexOf("WHERE") + 5);
                condition.setWhereClause(whereClause);
                condition.setParams(Arrays.asList(params));
            }

            return condition;

        } catch (Exception e) {
            log.error("解析查询条件失败", e);
            return new QueryCondition();
        }
    }

    /**
     * 切换读操作到目标数据库
     */
    public void switchReadToTarget(String tableName) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.COMPLETED);

            log.info("读操作已切换到目标数据库: {}", tableName);

        } catch (Exception e) {
            log.error("切换读操作到目标数据库失败", e);
            throw new MigrationException("切换读操作到目标数据库失败", e);
        }
    }

    /**
     * 切换写操作到目标数据库
     */
    public void switchWriteToTarget(String tableName) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.COMPLETED);

            log.info("写操作已切换到目标数据库: {}", tableName);

        } catch (Exception e) {
            log.error("切换写操作到目标数据库失败", e);
            throw new MigrationException("切换写操作到目标数据库失败", e);
        }
    }
}

2.3 灰度迁移策略

2.3.1 灰度迁移策略服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
/**
 * 灰度迁移策略服务
 */
@Service
public class GrayMigrationStrategyService {

    @Autowired
    private SourceDatabaseService sourceDatabaseService;

    @Autowired
    private TargetDatabaseService targetDatabaseService;

    @Autowired
    private MigrationConfigService migrationConfigService;

    @Autowired
    private DataSyncService dataSyncService;

    /**
     * 灰度迁移
     */
    public void grayMigration(String tableName, GrayMigrationConfig config) {
        try {
            // 1. 初始化灰度迁移
            initializeGrayMigration(tableName, config);

            // 2. 执行灰度迁移
            executeGrayMigration(tableName, config);

            // 3. 监控迁移进度
            monitorMigrationProgress(tableName, config);

            // 4. 验证迁移结果
            validateMigrationResult(tableName, config);

            // 5. 完成灰度迁移
            completeGrayMigration(tableName, config);

        } catch (Exception e) {
            log.error("灰度迁移失败", e);
            throw new MigrationException("灰度迁移失败", e);
        }
    }

    /**
     * 初始化灰度迁移
     */
    private void initializeGrayMigration(String tableName, GrayMigrationConfig config) {
        try {
            // 设置迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.IN_PROGRESS);

            // 创建迁移任务
            MigrationTask task = createMigrationTask(tableName, config);
            migrationConfigService.saveMigrationTask(task);

            // 初始化数据同步
            dataSyncService.initializeDataSync(tableName, config);

            log.info("灰度迁移初始化完成: {}", tableName);

        } catch (Exception e) {
            log.error("初始化灰度迁移失败", e);
            throw new MigrationException("初始化灰度迁移失败", e);
        }
    }

    /**
     * 创建迁移任务
     */
    private MigrationTask createMigrationTask(String tableName, GrayMigrationConfig config) {
        MigrationTask task = new MigrationTask();
        task.setTableName(tableName);
        task.setMigrationType(MigrationType.GRAY);
        task.setGrayRatio(config.getGrayRatio());
        task.setBatchSize(config.getBatchSize());
        task.setStatus(TaskStatus.PENDING);
        task.setCreateTime(new Date());

        return task;
    }

    /**
     * 执行灰度迁移
     */
    private void executeGrayMigration(String tableName, GrayMigrationConfig config) {
        try {
            // 获取迁移任务
            MigrationTask task = migrationConfigService.getMigrationTask(tableName);

            // 分批迁移数据
            int batchSize = config.getBatchSize();
            int offset = 0;

            while (true) {
                // 获取一批数据
                List<Map<String, Object>> batchData = sourceDatabaseService.queryBatch(tableName, offset, batchSize);

                if (batchData.isEmpty()) {
                    break;
                }

                // 灰度选择
                List<Map<String, Object>> grayData = selectGrayData(batchData, config.getGrayRatio());

                // 迁移数据
                migrateBatchData(tableName, grayData);

                // 更新进度
                updateMigrationProgress(task, batchData.size());

                offset += batchSize;

                // 休眠一段时间，避免对数据库造成压力
                Thread.sleep(config.getSleepInterval());
            }

            // 更新任务状态
            task.setStatus(TaskStatus.COMPLETED);
            migrationConfigService.updateMigrationTask(task);

        } catch (Exception e) {
            log.error("执行灰度迁移失败", e);
            throw new MigrationException("执行灰度迁移失败", e);
        }
    }

    /**
     * 灰度选择
     */
    private List<Map<String, Object>> selectGrayData(List<Map<String, Object>> batchData, double grayRatio) {
        List<Map<String, Object>> grayData = new ArrayList<>();

        for (Map<String, Object> data : batchData) {
            // 使用哈希算法进行灰度选择
            String primaryKey = data.get("id").toString();
            int hash = Math.abs(primaryKey.hashCode());
            double ratio = (double) (hash % 100) / 100.0;

            if (ratio < grayRatio) {
                grayData.add(data);
            }
        }

        return grayData;
    }

    /**
     * 迁移批量数据
     */
    private void migrateBatchData(String tableName, List<Map<String, Object>> batchData) {
        try {
            for (Map<String, Object> data : batchData) {
                // 插入到目标数据库
                targetDatabaseService.insert(tableName, data);

                // 记录迁移日志
                recordMigrationLog(tableName, data);
            }
        } catch (Exception e) {
            log.error("迁移批量数据失败", e);
            throw new MigrationException("迁移批量数据失败", e);
        }
    }

    /**
     * 记录迁移日志
     */
    private void recordMigrationLog(String tableName, Map<String, Object> data) {
        try {
            MigrationLog log = new MigrationLog();
            log.setTableName(tableName);
            log.setPrimaryKeyValue(data.get("id"));
            log.setOperation("INSERT");
            log.setStatus("SUCCESS");
            log.setCreateTime(new Date());

            migrationConfigService.saveMigrationLog(log);

        } catch (Exception e) {
            log.error("记录迁移日志失败", e);
        }
    }

    /**
     * 更新迁移进度
     */
    private void updateMigrationProgress(MigrationTask task, int processedCount) {
        try {
            task.setProcessedCount(task.getProcessedCount() + processedCount);
            task.setProgress((double) task.getProcessedCount() / task.getTotalCount() * 100);

            migrationConfigService.updateMigrationTask(task);

        } catch (Exception e) {
            log.error("更新迁移进度失败", e);
        }
    }

    /**
     * 监控迁移进度
     */
    private void monitorMigrationProgress(String tableName, GrayMigrationConfig config) {
        try {
            // 启动监控线程
            Thread monitorThread = new Thread(() -> {
                while (true) {
                    try {
                        // 获取迁移任务
                        MigrationTask task = migrationConfigService.getMigrationTask(tableName);

                        if (task.getStatus() == TaskStatus.COMPLETED) {
                            break;
                        }

                        // 检查迁移进度
                        checkMigrationProgress(task);

                        // 休眠一段时间
                        Thread.sleep(config.getMonitorInterval());

                    } catch (Exception e) {
                        log.error("监控迁移进度失败", e);
                        break;
                    }
                }
            });

            monitorThread.start();

        } catch (Exception e) {
            log.error("启动迁移进度监控失败", e);
        }
    }

    /**
     * 检查迁移进度
     */
    private void checkMigrationProgress(MigrationTask task) {
        try {
            // 检查是否有异常
            List<MigrationLog> errorLogs = migrationConfigService.getErrorMigrationLogs(task.getTableName());

            if (!errorLogs.isEmpty()) {
                log.warn("发现迁移异常: {}", errorLogs.size());
                // 处理异常
                handleMigrationErrors(task, errorLogs);
            }

            // 检查迁移速度
            double migrationSpeed = calculateMigrationSpeed(task);
            if (migrationSpeed < task.getMinSpeed()) {
                log.warn("迁移速度过慢: {}", migrationSpeed);
                // 调整迁移策略
                adjustMigrationStrategy(task);
            }

        } catch (Exception e) {
            log.error("检查迁移进度失败", e);
        }
    }

    /**
     * 处理迁移异常
     */
    private void handleMigrationErrors(MigrationTask task, List<MigrationLog> errorLogs) {
        try {
            for (MigrationLog errorLog : errorLogs) {
                // 重试迁移
                retryMigration(task.getTableName(), errorLog);
            }
        } catch (Exception e) {
            log.error("处理迁移异常失败", e);
        }
    }

    /**
     * 重试迁移
     */
    private void retryMigration(String tableName, MigrationLog errorLog) {
        try {
            // 从源数据库重新获取数据
            Map<String, Object> data = sourceDatabaseService.queryByPrimaryKey(tableName, errorLog.getPrimaryKeyValue());

            if (data != null) {
                // 重新迁移
                targetDatabaseService.insert(tableName, data);

                // 更新日志状态
                errorLog.setStatus("SUCCESS");
                errorLog.setRetryCount(errorLog.getRetryCount() + 1);
                migrationConfigService.updateMigrationLog(errorLog);
            }

        } catch (Exception e) {
            log.error("重试迁移失败", e);
        }
    }

    /**
     * 计算迁移速度
     */
    private double calculateMigrationSpeed(MigrationTask task) {
        try {
            long currentTime = System.currentTimeMillis();
            long startTime = task.getCreateTime().getTime();
            long duration = currentTime - startTime;

            if (duration == 0) {
                return 0;
            }

            return (double) task.getProcessedCount() / duration * 1000; // 每秒处理数

        } catch (Exception e) {
            log.error("计算迁移速度失败", e);
            return 0;
        }
    }

    /**
     * 调整迁移策略
     */
    private void adjustMigrationStrategy(MigrationTask task) {
        try {
            // 增加批次大小
            int newBatchSize = task.getBatchSize() * 2;
            task.setBatchSize(newBatchSize);

            // 更新任务
            migrationConfigService.updateMigrationTask(task);

            log.info("调整迁移策略: 批次大小调整为 {}", newBatchSize);

        } catch (Exception e) {
            log.error("调整迁移策略失败", e);
        }
    }

    /**
     * 验证迁移结果
     */
    private void validateMigrationResult(String tableName, GrayMigrationConfig config) {
        try {
            // 数据一致性校验
            DataConsistencyResult result = dataSyncService.validateDataConsistency(tableName);

            if (!result.isConsistent()) {
                log.error("数据一致性校验失败: {}", result.getErrorMessage());
                throw new MigrationException("数据一致性校验失败: " + result.getErrorMessage());
            }

            // 性能测试
            PerformanceTestResult performanceResult = dataSyncService.performanceTest(tableName);

            if (performanceResult.getResponseTime() > config.getMaxResponseTime()) {
                log.warn("性能测试不达标: 响应时间 {}", performanceResult.getResponseTime());
            }

            log.info("迁移结果验证通过: {}", tableName);

        } catch (Exception e) {
            log.error("验证迁移结果失败", e);
            throw new MigrationException("验证迁移结果失败", e);
        }
    }

    /**
     * 完成灰度迁移
     */
    private void completeGrayMigration(String tableName, GrayMigrationConfig config) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.COMPLETED);

            // 清理临时数据
            cleanupTemporaryData(tableName);

            // 发送完成通知
            sendCompletionNotification(tableName, config);

            log.info("灰度迁移完成: {}", tableName);

        } catch (Exception e) {
            log.error("完成灰度迁移失败", e);
            throw new MigrationException("完成灰度迁移失败", e);
        }
    }

    /**
     * 清理临时数据
     */
    private void cleanupTemporaryData(String tableName) {
        try {
            // 清理迁移日志
            migrationConfigService.cleanupMigrationLogs(tableName);

            // 清理临时表
            migrationConfigService.cleanupTemporaryTables(tableName);

        } catch (Exception e) {
            log.error("清理临时数据失败", e);
        }
    }

    /**
     * 发送完成通知
     */
    private void sendCompletionNotification(String tableName, GrayMigrationConfig config) {
        try {
            // 发送邮件通知
            EmailNotification email = new EmailNotification();
            email.setSubject("数据迁移完成通知");
            email.setContent("表 " + tableName + " 的灰度迁移已完成");
            email.setRecipients(config.getNotificationRecipients());

            // 发送通知
            notificationService.sendEmail(email);

        } catch (Exception e) {
            log.error("发送完成通知失败", e);
        }
    }
}

三、数据同步方案

3.1 实时数据同步

3.1.1 实时数据同步服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
/**
 * 实时数据同步服务
 */
@Service
public class RealTimeDataSyncService {

    @Autowired
    private SourceDatabaseService sourceDatabaseService;

    @Autowired
    private TargetDatabaseService targetDatabaseService;

    @Autowired
    private DataValidationService dataValidationService;

    @Autowired
    private MigrationConfigService migrationConfigService;

    /**
     * 启动实时数据同步
     */
    public void startRealTimeSync(String tableName) {
        try {
            // 创建数据同步任务
            DataSyncTask task = createDataSyncTask(tableName);
            migrationConfigService.saveDataSyncTask(task);

            // 启动同步线程
            Thread syncThread = new Thread(() -> {
                try {
                    executeRealTimeSync(task);
                } catch (Exception e) {
                    log.error("实时数据同步执行失败", e);
                }
            });

            syncThread.start();

            log.info("实时数据同步已启动: {}", tableName);

        } catch (Exception e) {
            log.error("启动实时数据同步失败", e);
            throw new MigrationException("启动实时数据同步失败", e);
        }
    }

    /**
     * 创建数据同步任务
     */
    private DataSyncTask createDataSyncTask(String tableName) {
        DataSyncTask task = new DataSyncTask();
        task.setTableName(tableName);
        task.setSyncType(SyncType.REAL_TIME);
        task.setStatus(TaskStatus.RUNNING);
        task.setCreateTime(new Date());

        return task;
    }

    /**
     * 执行实时数据同步
     */
    private void executeRealTimeSync(DataSyncTask task) {
        try {
            while (task.getStatus() == TaskStatus.RUNNING) {
                // 获取变更数据
                List<DataChange> changes = getDataChanges(task.getTableName());

                if (!changes.isEmpty()) {
                    // 同步变更数据
                    syncDataChanges(task.getTableName(), changes);
                }

                // 休眠一段时间
                Thread.sleep(1000); // 1秒
            }

        } catch (Exception e) {
            log.error("执行实时数据同步失败", e);
            task.setStatus(TaskStatus.FAILED);
            migrationConfigService.updateDataSyncTask(task);
        }
    }

    /**
     * 获取变更数据
     */
    private List<DataChange> getDataChanges(String tableName) {
        try {
            // 从源数据库获取变更数据
            List<DataChange> changes = sourceDatabaseService.getDataChanges(tableName);

            return changes;

        } catch (Exception e) {
            log.error("获取变更数据失败", e);
            return new ArrayList<>();
        }
    }

    /**
     * 同步变更数据
     */
    private void syncDataChanges(String tableName, List<DataChange> changes) {
        try {
            for (DataChange change : changes) {
                // 同步单个变更
                syncSingleChange(tableName, change);
            }
        } catch (Exception e) {
            log.error("同步变更数据失败", e);
        }
    }

    /**
     * 同步单个变更
     */
    private void syncSingleChange(String tableName, DataChange change) {
        try {
            switch (change.getOperation()) {
                case INSERT:
                    targetDatabaseService.insert(tableName, change.getData());
                    break;
                case UPDATE:
                    targetDatabaseService.update(tableName, change.getData());
                    break;
                case DELETE:
                    targetDatabaseService.delete(tableName, change.getData());
                    break;
                default:
                    log.warn("不支持的操作类型: {}", change.getOperation());
                    break;
            }

            // 记录同步日志
            recordSyncLog(tableName, change);

        } catch (Exception e) {
            log.error("同步单个变更失败", e);
            // 记录错误日志
            recordErrorLog(tableName, change, e);
        }
    }

    /**
     * 记录同步日志
     */
    private void recordSyncLog(String tableName, DataChange change) {
        try {
            SyncLog log = new SyncLog();
            log.setTableName(tableName);
            log.setPrimaryKeyValue(change.getPrimaryKeyValue());
            log.setOperation(change.getOperation());
            log.setStatus("SUCCESS");
            log.setCreateTime(new Date());

            migrationConfigService.saveSyncLog(log);

        } catch (Exception e) {
            log.error("记录同步日志失败", e);
        }
    }

    /**
     * 记录错误日志
     */
    private void recordErrorLog(String tableName, DataChange change, Exception e) {
        try {
            SyncLog log = new SyncLog();
            log.setTableName(tableName);
            log.setPrimaryKeyValue(change.getPrimaryKeyValue());
            log.setOperation(change.getOperation());
            log.setStatus("FAILED");
            log.setErrorMessage(e.getMessage());
            log.setCreateTime(new Date());

            migrationConfigService.saveSyncLog(log);

        } catch (Exception ex) {
            log.error("记录错误日志失败", ex);
        }
    }

    /**
     * 停止实时数据同步
     */
    public void stopRealTimeSync(String tableName) {
        try {
            // 获取同步任务
            DataSyncTask task = migrationConfigService.getDataSyncTask(tableName);

            if (task != null) {
                // 更新任务状态
                task.setStatus(TaskStatus.STOPPED);
                migrationConfigService.updateDataSyncTask(task);

                log.info("实时数据同步已停止: {}", tableName);
            }

        } catch (Exception e) {
            log.error("停止实时数据同步失败", e);
            throw new MigrationException("停止实时数据同步失败", e);
        }
    }

    /**
     * 数据一致性校验
     */
    public DataConsistencyResult validateDataConsistency(String tableName) {
        try {
            DataConsistencyResult result = new DataConsistencyResult();

            // 获取源数据库数据量
            long sourceCount = sourceDatabaseService.getDataCount(tableName);

            // 获取目标数据库数据量
            long targetCount = targetDatabaseService.getDataCount(tableName);

            // 比较数据量
            if (sourceCount != targetCount) {
                result.setConsistent(false);
                result.setErrorMessage("数据量不一致: 源=" + sourceCount + ", 目标=" + targetCount);
                return result;
            }

            // 抽样校验数据内容
            List<Map<String, Object>> sampleData = sourceDatabaseService.getSampleData(tableName, 1000);

            for (Map<String, Object> data : sampleData) {
                String primaryKey = data.get("id").toString();

                // 从目标数据库查询
                Map<String, Object> targetData = targetDatabaseService.queryByPrimaryKey(tableName, primaryKey);

                if (targetData == null) {
                    result.setConsistent(false);
                    result.setErrorMessage("目标数据库缺少数据: " + primaryKey);
                    return result;
                }

                // 比较数据内容
                if (!dataValidationService.validateDataConsistency(data, targetData)) {
                    result.setConsistent(false);
                    result.setErrorMessage("数据内容不一致: " + primaryKey);
                    return result;
                }
            }

            result.setConsistent(true);
            return result;

        } catch (Exception e) {
            log.error("数据一致性校验失败", e);
            DataConsistencyResult result = new DataConsistencyResult();
            result.setConsistent(false);
            result.setErrorMessage("数据一致性校验失败: " + e.getMessage());
            return result;
        }
    }
}

3.2 增量数据同步

3.2.1 增量数据同步服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
/**
 * 增量数据同步服务
 */
@Service
public class IncrementalDataSyncService {

    @Autowired
    private SourceDatabaseService sourceDatabaseService;

    @Autowired
    private TargetDatabaseService targetDatabaseService;

    @Autowired
    private DataValidationService dataValidationService;

    @Autowired
    private MigrationConfigService migrationConfigService;

    /**
     * 启动增量数据同步
     */
    public void startIncrementalSync(String tableName) {
        try {
            // 创建增量同步任务
            IncrementalSyncTask task = createIncrementalSyncTask(tableName);
            migrationConfigService.saveIncrementalSyncTask(task);

            // 启动同步线程
            Thread syncThread = new Thread(() -> {
                try {
                    executeIncrementalSync(task);
                } catch (Exception e) {
                    log.error("增量数据同步执行失败", e);
                }
            });

            syncThread.start();

            log.info("增量数据同步已启动: {}", tableName);

        } catch (Exception e) {
            log.error("启动增量数据同步失败", e);
            throw new MigrationException("启动增量数据同步失败", e);
        }
    }

    /**
     * 创建增量同步任务
     */
    private IncrementalSyncTask createIncrementalSyncTask(String tableName) {
        IncrementalSyncTask task = new IncrementalSyncTask();
        task.setTableName(tableName);
        task.setSyncType(SyncType.INCREMENTAL);
        task.setStatus(TaskStatus.RUNNING);
        task.setCreateTime(new Date());

        // 获取最后同步时间
        Date lastSyncTime = migrationConfigService.getLastSyncTime(tableName);
        task.setLastSyncTime(lastSyncTime);

        return task;
    }

    /**
     * 执行增量数据同步
     */
    private void executeIncrementalSync(IncrementalSyncTask task) {
        try {
            while (task.getStatus() == TaskStatus.RUNNING) {
                // 获取增量数据
                List<Map<String, Object>> incrementalData = getIncrementalData(task);

                if (!incrementalData.isEmpty()) {
                    // 同步增量数据
                    syncIncrementalData(task.getTableName(), incrementalData);

                    // 更新最后同步时间
                    updateLastSyncTime(task);
                }

                // 休眠一段时间
                Thread.sleep(task.getSyncInterval());
            }

        } catch (Exception e) {
            log.error("执行增量数据同步失败", e);
            task.setStatus(TaskStatus.FAILED);
            migrationConfigService.updateIncrementalSyncTask(task);
        }
    }

    /**
     * 获取增量数据
     */
    private List<Map<String, Object>> getIncrementalData(IncrementalSyncTask task) {
        try {
            // 从源数据库获取增量数据
            List<Map<String, Object>> incrementalData = sourceDatabaseService.getIncrementalData(
                    task.getTableName(), task.getLastSyncTime());

            return incrementalData;

        } catch (Exception e) {
            log.error("获取增量数据失败", e);
            return new ArrayList<>();
        }
    }

    /**
     * 同步增量数据
     */
    private void syncIncrementalData(String tableName, List<Map<String, Object>> incrementalData) {
        try {
            for (Map<String, Object> data : incrementalData) {
                // 同步单条数据
                syncSingleData(tableName, data);
            }
        } catch (Exception e) {
            log.error("同步增量数据失败", e);
        }
    }

    /**
     * 同步单条数据
     */
    private void syncSingleData(String tableName, Map<String, Object> data) {
        try {
            // 检查数据是否已存在
            String primaryKey = data.get("id").toString();
            Map<String, Object> existingData = targetDatabaseService.queryByPrimaryKey(tableName, primaryKey);

            if (existingData != null) {
                // 更新数据
                targetDatabaseService.update(tableName, data);
            } else {
                // 插入数据
                targetDatabaseService.insert(tableName, data);
            }

            // 记录同步日志
            recordSyncLog(tableName, data);

        } catch (Exception e) {
            log.error("同步单条数据失败", e);
            // 记录错误日志
            recordErrorLog(tableName, data, e);
        }
    }

    /**
     * 记录同步日志
     */
    private void recordSyncLog(String tableName, Map<String, Object> data) {
        try {
            SyncLog log = new SyncLog();
            log.setTableName(tableName);
            log.setPrimaryKeyValue(data.get("id"));
            log.setOperation("SYNC");
            log.setStatus("SUCCESS");
            log.setCreateTime(new Date());

            migrationConfigService.saveSyncLog(log);

        } catch (Exception e) {
            log.error("记录同步日志失败", e);
        }
    }

    /**
     * 记录错误日志
     */
    private void recordErrorLog(String tableName, Map<String, Object> data, Exception e) {
        try {
            SyncLog log = new SyncLog();
            log.setTableName(tableName);
            log.setPrimaryKeyValue(data.get("id"));
            log.setOperation("SYNC");
            log.setStatus("FAILED");
            log.setErrorMessage(e.getMessage());
            log.setCreateTime(new Date());

            migrationConfigService.saveSyncLog(log);

        } catch (Exception ex) {
            log.error("记录错误日志失败", ex);
        }
    }

    /**
     * 更新最后同步时间
     */
    private void updateLastSyncTime(IncrementalSyncTask task) {
        try {
            Date currentTime = new Date();
            task.setLastSyncTime(currentTime);

            // 更新数据库
            migrationConfigService.updateLastSyncTime(task.getTableName(), currentTime);

        } catch (Exception e) {
            log.error("更新最后同步时间失败", e);
        }
    }

    /**
     * 停止增量数据同步
     */
    public void stopIncrementalSync(String tableName) {
        try {
            // 获取同步任务
            IncrementalSyncTask task = migrationConfigService.getIncrementalSyncTask(tableName);

            if (task != null) {
                // 更新任务状态
                task.setStatus(TaskStatus.STOPPED);
                migrationConfigService.updateIncrementalSyncTask(task);

                log.info("增量数据同步已停止: {}", tableName);
            }

        } catch (Exception e) {
            log.error("停止增量数据同步失败", e);
            throw new MigrationException("停止增量数据同步失败", e);
        }
    }

    /**
     * 全量数据同步
     */
    public void fullDataSync(String tableName) {
        try {
            // 创建全量同步任务
            FullSyncTask task = createFullSyncTask(tableName);
            migrationConfigService.saveFullSyncTask(task);

            // 执行全量同步
            executeFullSync(task);

            log.info("全量数据同步完成: {}", tableName);

        } catch (Exception e) {
            log.error("全量数据同步失败", e);
            throw new MigrationException("全量数据同步失败", e);
        }
    }

    /**
     * 创建全量同步任务
     */
    private FullSyncTask createFullSyncTask(String tableName) {
        FullSyncTask task = new FullSyncTask();
        task.setTableName(tableName);
        task.setSyncType(SyncType.FULL);
        task.setStatus(TaskStatus.PENDING);
        task.setCreateTime(new Date());

        return task;
    }

    /**
     * 执行全量同步
     */
    private void executeFullSync(FullSyncTask task) {
        try {
            // 更新任务状态
            task.setStatus(TaskStatus.RUNNING);
            migrationConfigService.updateFullSyncTask(task);

            // 获取全量数据
            List<Map<String, Object>> fullData = sourceDatabaseService.getAllData(task.getTableName());

            // 分批同步
            int batchSize = 1000;
            int totalCount = fullData.size();
            int processedCount = 0;

            for (int i = 0; i < totalCount; i += batchSize) {
                int endIndex = Math.min(i + batchSize, totalCount);
                List<Map<String, Object>> batchData = fullData.subList(i, endIndex);

                // 同步批量数据
                syncBatchData(task.getTableName(), batchData);

                processedCount += batchData.size();

                // 更新进度
                task.setProcessedCount(processedCount);
                task.setProgress((double) processedCount / totalCount * 100);
                migrationConfigService.updateFullSyncTask(task);

                // 休眠一段时间，避免对数据库造成压力
                Thread.sleep(100);
            }

            // 更新任务状态
            task.setStatus(TaskStatus.COMPLETED);
            migrationConfigService.updateFullSyncTask(task);

        } catch (Exception e) {
            log.error("执行全量同步失败", e);
            task.setStatus(TaskStatus.FAILED);
            migrationConfigService.updateFullSyncTask(task);
        }
    }

    /**
     * 同步批量数据
     */
    private void syncBatchData(String tableName, List<Map<String, Object>> batchData) {
        try {
            for (Map<String, Object> data : batchData) {
                // 检查数据是否已存在
                String primaryKey = data.get("id").toString();
                Map<String, Object> existingData = targetDatabaseService.queryByPrimaryKey(tableName, primaryKey);

                if (existingData != null) {
                    // 更新数据
                    targetDatabaseService.update(tableName, data);
                } else {
                    // 插入数据
                    targetDatabaseService.insert(tableName, data);
                }
            }
        } catch (Exception e) {
            log.error("同步批量数据失败", e);
        }
    }
}

四、企业级数据迁移应用

4.1 电商数据迁移方案

4.1.1 电商数据迁移服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
/**
 * 电商数据迁移服务
 */
@Service
public class EcommerceDataMigrationService {

    @Autowired
    private DualWriteStrategyService dualWriteStrategyService;

    @Autowired
    private ReadWriteSeparationStrategyService readWriteSeparationService;

    @Autowired
    private GrayMigrationStrategyService grayMigrationStrategyService;

    @Autowired
    private RealTimeDataSyncService realTimeDataSyncService;

    @Autowired
    private IncrementalDataSyncService incrementalDataSyncService;

    /**
     * 用户表数据迁移
     */
    public void migrateUserTable() {
        try {
            String tableName = "user";

            // 1. 启动双写策略
            startDualWriteStrategy(tableName);

            // 2. 启动实时数据同步
            startRealTimeDataSync(tableName);

            // 3. 执行灰度迁移
            executeGrayMigration(tableName);

            // 4. 切换读操作到目标数据库
            switchReadToTarget(tableName);

            // 5. 切换写操作到目标数据库
            switchWriteToTarget(tableName);

            // 6. 停止双写策略
            stopDualWriteStrategy(tableName);

            log.info("用户表数据迁移完成");

        } catch (Exception e) {
            log.error("用户表数据迁移失败", e);
            throw new MigrationException("用户表数据迁移失败", e);
        }
    }

    /**
     * 订单表数据迁移
     */
    public void migrateOrderTable() {
        try {
            String tableName = "order";

            // 1. 启动双写策略
            startDualWriteStrategy(tableName);

            // 2. 启动实时数据同步
            startRealTimeDataSync(tableName);

            // 3. 执行灰度迁移
            executeGrayMigration(tableName);

            // 4. 切换读操作到目标数据库
            switchReadToTarget(tableName);

            // 5. 切换写操作到目标数据库
            switchWriteToTarget(tableName);

            // 6. 停止双写策略
            stopDualWriteStrategy(tableName);

            log.info("订单表数据迁移完成");

        } catch (Exception e) {
            log.error("订单表数据迁移失败", e);
            throw new MigrationException("订单表数据迁移失败", e);
        }
    }

    /**
     * 商品表数据迁移
     */
    public void migrateProductTable() {
        try {
            String tableName = "product";

            // 1. 启动双写策略
            startDualWriteStrategy(tableName);

            // 2. 启动实时数据同步
            startRealTimeDataSync(tableName);

            // 3. 执行灰度迁移
            executeGrayMigration(tableName);

            // 4. 切换读操作到目标数据库
            switchReadToTarget(tableName);

            // 5. 切换写操作到目标数据库
            switchWriteToTarget(tableName);

            // 6. 停止双写策略
            stopDualWriteStrategy(tableName);

            log.info("商品表数据迁移完成");

        } catch (Exception e) {
            log.error("商品表数据迁移失败", e);
            throw new MigrationException("商品表数据迁移失败", e);
        }
    }

    /**
     * 启动双写策略
     */
    private void startDualWriteStrategy(String tableName) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.IN_PROGRESS);

            log.info("双写策略已启动: {}", tableName);

        } catch (Exception e) {
            log.error("启动双写策略失败", e);
            throw new MigrationException("启动双写策略失败", e);
        }
    }

    /**
     * 启动实时数据同步
     */
    private void startRealTimeDataSync(String tableName) {
        try {
            realTimeDataSyncService.startRealTimeSync(tableName);

            log.info("实时数据同步已启动: {}", tableName);

        } catch (Exception e) {
            log.error("启动实时数据同步失败", e);
            throw new MigrationException("启动实时数据同步失败", e);
        }
    }

    /**
     * 执行灰度迁移
     */
    private void executeGrayMigration(String tableName) {
        try {
            // 创建灰度迁移配置
            GrayMigrationConfig config = createGrayMigrationConfig(tableName);

            // 执行灰度迁移
            grayMigrationStrategyService.grayMigration(tableName, config);

            log.info("灰度迁移已执行: {}", tableName);

        } catch (Exception e) {
            log.error("执行灰度迁移失败", e);
            throw new MigrationException("执行灰度迁移失败", e);
        }
    }

    /**
     * 创建灰度迁移配置
     */
    private GrayMigrationConfig createGrayMigrationConfig(String tableName) {
        GrayMigrationConfig config = new GrayMigrationConfig();

        switch (tableName) {
            case "user":
                config.setGrayRatio(0.1); // 10%灰度
                config.setBatchSize(1000);
                config.setSleepInterval(100);
                config.setMonitorInterval(5000);
                config.setMaxResponseTime(1000);
                break;

            case "order":
                config.setGrayRatio(0.05); // 5%灰度
                config.setBatchSize(500);
                config.setSleepInterval(200);
                config.setMonitorInterval(10000);
                config.setMaxResponseTime(2000);
                break;

            case "product":
                config.setGrayRatio(0.2); // 20%灰度
                config.setBatchSize(2000);
                config.setSleepInterval(50);
                config.setMonitorInterval(3000);
                config.setMaxResponseTime(500);
                break;

            default:
                config.setGrayRatio(0.1);
                config.setBatchSize(1000);
                config.setSleepInterval(100);
                config.setMonitorInterval(5000);
                config.setMaxResponseTime(1000);
                break;
        }

        return config;
    }

    /**
     * 切换读操作到目标数据库
     */
    private void switchReadToTarget(String tableName) {
        try {
            readWriteSeparationService.switchReadToTarget(tableName);

            log.info("读操作已切换到目标数据库: {}", tableName);

        } catch (Exception e) {
            log.error("切换读操作到目标数据库失败", e);
            throw new MigrationException("切换读操作到目标数据库失败", e);
        }
    }

    /**
     * 切换写操作到目标数据库
     */
    private void switchWriteToTarget(String tableName) {
        try {
            readWriteSeparationService.switchWriteToTarget(tableName);

            log.info("写操作已切换到目标数据库: {}", tableName);

        } catch (Exception e) {
            log.error("切换写操作到目标数据库失败", e);
            throw new MigrationException("切换写操作到目标数据库失败", e);
        }
    }

    /**
     * 停止双写策略
     */
    private void stopDualWriteStrategy(String tableName) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.COMPLETED);

            log.info("双写策略已停止: {}", tableName);

        } catch (Exception e) {
            log.error("停止双写策略失败", e);
            throw new MigrationException("停止双写策略失败", e);
        }
    }

    /**
     * 数据迁移回滚
     */
    public void rollbackMigration(String tableName) {
        try {
            // 1. 停止双写策略
            stopDualWriteStrategy(tableName);

            // 2. 停止实时数据同步
            realTimeDataSyncService.stopRealTimeSync(tableName);

            // 3. 切换读操作回源数据库
            switchReadToSource(tableName);

            // 4. 切换写操作回源数据库
            switchWriteToSource(tableName);

            // 5. 清理目标数据库数据
            cleanupTargetDatabase(tableName);

            log.info("数据迁移回滚完成: {}", tableName);

        } catch (Exception e) {
            log.error("数据迁移回滚失败", e);
            throw new MigrationException("数据迁移回滚失败", e);
        }
    }

    /**
     * 切换读操作回源数据库
     */
    private void switchReadToSource(String tableName) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.NOT_STARTED);

            log.info("读操作已切换回源数据库: {}", tableName);

        } catch (Exception e) {
            log.error("切换读操作回源数据库失败", e);
            throw new MigrationException("切换读操作回源数据库失败", e);
        }
    }

    /**
     * 切换写操作回源数据库
     */
    private void switchWriteToSource(String tableName) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.NOT_STARTED);

            log.info("写操作已切换回源数据库: {}", tableName);

        } catch (Exception e) {
            log.error("切换写操作回源数据库失败", e);
            throw new MigrationException("切换写操作回源数据库失败", e);
        }
    }

    /**
     * 清理目标数据库数据
     */
    private void cleanupTargetDatabase(String tableName) {
        try {
            // 清理目标数据库数据
            targetDatabaseService.truncateTable(tableName);

            log.info("目标数据库数据已清理: {}", tableName);

        } catch (Exception e) {
            log.error("清理目标数据库数据失败", e);
            throw new MigrationException("清理目标数据库数据失败", e);
        }
    }
}

4.2 社交数据迁移方案

4.2.1 社交数据迁移服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
/**
 * 社交数据迁移服务
 */
@Service
public class SocialDataMigrationService {

    @Autowired
    private DualWriteStrategyService dualWriteStrategyService;

    @Autowired
    private ReadWriteSeparationStrategyService readWriteSeparationService;

    @Autowired
    private GrayMigrationStrategyService grayMigrationStrategyService;

    @Autowired
    private RealTimeDataSyncService realTimeDataSyncService;

    @Autowired
    private IncrementalDataSyncService incrementalDataSyncService;

    /**
     * 用户表数据迁移
     */
    public void migrateUserTable() {
        try {
            String tableName = "user";

            // 1. 启动双写策略
            startDualWriteStrategy(tableName);

            // 2. 启动实时数据同步
            startRealTimeDataSync(tableName);

            // 3. 执行灰度迁移
            executeGrayMigration(tableName);

            // 4. 切换读操作到目标数据库
            switchReadToTarget(tableName);

            // 5. 切换写操作到目标数据库
            switchWriteToTarget(tableName);

            // 6. 停止双写策略
            stopDualWriteStrategy(tableName);

            log.info("用户表数据迁移完成");

        } catch (Exception e) {
            log.error("用户表数据迁移失败", e);
            throw new MigrationException("用户表数据迁移失败", e);
        }
    }

    /**
     * 帖子表数据迁移
     */
    public void migratePostTable() {
        try {
            String tableName = "post";

            // 1. 启动双写策略
            startDualWriteStrategy(tableName);

            // 2. 启动实时数据同步
            startRealTimeDataSync(tableName);

            // 3. 执行灰度迁移
            executeGrayMigration(tableName);

            // 4. 切换读操作到目标数据库
            switchReadToTarget(tableName);

            // 5. 切换写操作到目标数据库
            switchWriteToTarget(tableName);

            // 6. 停止双写策略
            stopDualWriteStrategy(tableName);

            log.info("帖子表数据迁移完成");

        } catch (Exception e) {
            log.error("帖子表数据迁移失败", e);
            throw new MigrationException("帖子表数据迁移失败", e);
        }
    }

    /**
     * 评论表数据迁移
     */
    public void migrateCommentTable() {
        try {
            String tableName = "comment";

            // 1. 启动双写策略
            startDualWriteStrategy(tableName);

            // 2. 启动实时数据同步
            startRealTimeDataSync(tableName);

            // 3. 执行灰度迁移
            executeGrayMigration(tableName);

            // 4. 切换读操作到目标数据库
            switchReadToTarget(tableName);

            // 5. 切换写操作到目标数据库
            switchWriteToTarget(tableName);

            // 6. 停止双写策略
            stopDualWriteStrategy(tableName);

            log.info("评论表数据迁移完成");

        } catch (Exception e) {
            log.error("评论表数据迁移失败", e);
            throw new MigrationException("评论表数据迁移失败", e);
        }
    }

    /**
     * 启动双写策略
     */
    private void startDualWriteStrategy(String tableName) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.IN_PROGRESS);

            log.info("双写策略已启动: {}", tableName);

        } catch (Exception e) {
            log.error("启动双写策略失败", e);
            throw new MigrationException("启动双写策略失败", e);
        }
    }

    /**
     * 启动实时数据同步
     */
    private void startRealTimeDataSync(String tableName) {
        try {
            realTimeDataSyncService.startRealTimeSync(tableName);

            log.info("实时数据同步已启动: {}", tableName);

        } catch (Exception e) {
            log.error("启动实时数据同步失败", e);
            throw new MigrationException("启动实时数据同步失败", e);
        }
    }

    /**
     * 执行灰度迁移
     */
    private void executeGrayMigration(String tableName) {
        try {
            // 创建灰度迁移配置
            GrayMigrationConfig config = createGrayMigrationConfig(tableName);

            // 执行灰度迁移
            grayMigrationStrategyService.grayMigration(tableName, config);

            log.info("灰度迁移已执行: {}", tableName);

        } catch (Exception e) {
            log.error("执行灰度迁移失败", e);
            throw new MigrationException("执行灰度迁移失败", e);
        }
    }

    /**
     * 创建灰度迁移配置
     */
    private GrayMigrationConfig createGrayMigrationConfig(String tableName) {
        GrayMigrationConfig config = new GrayMigrationConfig();

        switch (tableName) {
            case "user":
                config.setGrayRatio(0.1); // 10%灰度
                config.setBatchSize(1000);
                config.setSleepInterval(100);
                config.setMonitorInterval(5000);
                config.setMaxResponseTime(1000);
                break;

            case "post":
                config.setGrayRatio(0.05); // 5%灰度
                config.setBatchSize(500);
                config.setSleepInterval(200);
                config.setMonitorInterval(10000);
                config.setMaxResponseTime(2000);
                break;

            case "comment":
                config.setGrayRatio(0.02); // 2%灰度
                config.setBatchSize(200);
                config.setSleepInterval(500);
                config.setMonitorInterval(15000);
                config.setMaxResponseTime(3000);
                break;

            default:
                config.setGrayRatio(0.1);
                config.setBatchSize(1000);
                config.setSleepInterval(100);
                config.setMonitorInterval(5000);
                config.setMaxResponseTime(1000);
                break;
        }

        return config;
    }

    /**
     * 切换读操作到目标数据库
     */
    private void switchReadToTarget(String tableName) {
        try {
            readWriteSeparationService.switchReadToTarget(tableName);

            log.info("读操作已切换到目标数据库: {}", tableName);

        } catch (Exception e) {
            log.error("切换读操作到目标数据库失败", e);
            throw new MigrationException("切换读操作到目标数据库失败", e);
        }
    }

    /**
     * 切换写操作到目标数据库
     */
    private void switchWriteToTarget(String tableName) {
        try {
            readWriteSeparationService.switchWriteToTarget(tableName);

            log.info("写操作已切换到目标数据库: {}", tableName);

        } catch (Exception e) {
            log.error("切换写操作到目标数据库失败", e);
            throw new MigrationException("切换写操作到目标数据库失败", e);
        }
    }

    /**
     * 停止双写策略
     */
    private void stopDualWriteStrategy(String tableName) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.COMPLETED);

            log.info("双写策略已停止: {}", tableName);

        } catch (Exception e) {
            log.error("停止双写策略失败", e);
            throw new MigrationException("停止双写策略失败", e);
        }
    }

    /**
     * 数据迁移回滚
     */
    public void rollbackMigration(String tableName) {
        try {
            // 1. 停止双写策略
            stopDualWriteStrategy(tableName);

            // 2. 停止实时数据同步
            realTimeDataSyncService.stopRealTimeSync(tableName);

            // 3. 切换读操作回源数据库
            switchReadToSource(tableName);

            // 4. 切换写操作回源数据库
            switchWriteToSource(tableName);

            // 5. 清理目标数据库数据
            cleanupTargetDatabase(tableName);

            log.info("数据迁移回滚完成: {}", tableName);

        } catch (Exception e) {
            log.error("数据迁移回滚失败", e);
            throw new MigrationException("数据迁移回滚失败", e);
        }
    }

    /**
     * 切换读操作回源数据库
     */
    private void switchReadToSource(String tableName) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.NOT_STARTED);

            log.info("读操作已切换回源数据库: {}", tableName);

        } catch (Exception e) {
            log.error("切换读操作回源数据库失败", e);
            throw new MigrationException("切换读操作回源数据库失败", e);
        }
    }

    /**
     * 切换写操作回源数据库
     */
    private void switchWriteToSource(String tableName) {
        try {
            // 更新迁移状态
            migrationConfigService.updateMigrationStatus(tableName, MigrationStatus.NOT_STARTED);

            log.info("写操作已切换回源数据库: {}", tableName);

        } catch (Exception e) {
            log.error("切换写操作回源数据库失败", e);
            throw new MigrationException("切换写操作回源数据库失败", e);
        }
    }

    /**
     * 清理目标数据库数据
     */
    private void cleanupTargetDatabase(String tableName) {
        try {
            // 清理目标数据库数据
            targetDatabaseService.truncateTable(tableName);

            log.info("目标数据库数据已清理: {}", tableName);

        } catch (Exception e) {
            log.error("清理目标数据库数据失败", e);
            throw new MigrationException("清理目标数据库数据失败", e);
        }
    }
}

五、性能优化与监控

5.1 性能优化

5.1.1 数据迁移性能优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
/**
 * 数据迁移性能优化服务
 */
@Service
public class DataMigrationPerformanceOptimizationService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private CaffeineCache localCache;

    private final String MIGRATION_CACHE_PREFIX = "migration_cache:";

    /**
     * 缓存迁移结果
     */
    public void cacheMigrationResult(String tableName, String operation, Object result) {
        String cacheKey = MIGRATION_CACHE_PREFIX + tableName + ":" + operation;

        try {
            // 写入本地缓存
            localCache.put(cacheKey, result);

            // 写入Redis缓存
            String redisCacheKey = "redis_cache:" + cacheKey;
            redisTemplate.opsForValue().set(redisCacheKey, result, Duration.ofHours(1));

        } catch (Exception e) {
            log.error("缓存迁移结果失败", e);
        }
    }

    /**
     * 获取缓存的迁移结果
     */
    public Object getCachedMigrationResult(String tableName, String operation) {
        String cacheKey = MIGRATION_CACHE_PREFIX + tableName + ":" + operation;

        try {
            // 从本地缓存获取
            Object cachedResult = localCache.getIfPresent(cacheKey);
            if (cachedResult != null) {
                return cachedResult;
            }

            // 从Redis获取
            String redisCacheKey = "redis_cache:" + cacheKey;
            Object redisResult = redisTemplate.opsForValue().get(redisCacheKey);
            if (redisResult != null) {
                localCache.put(cacheKey, redisResult);
                return redisResult;
            }

        } catch (Exception e) {
            log.error("获取缓存的迁移结果失败", e);
        }

        return null;
    }

    /**
     * 批量迁移优化
     */
    public void batchMigrationOptimized(String tableName, List<Map<String, Object>> dataList) {
        try {
            // 分批处理
            int batchSize = 1000;
            int totalCount = dataList.size();

            for (int i = 0; i < totalCount; i += batchSize) {
                int endIndex = Math.min(i + batchSize, totalCount);
                List<Map<String, Object>> batchData = dataList.subList(i, endIndex);

                // 并行处理批次
                processBatchData(tableName, batchData);

                // 休眠一段时间，避免对数据库造成压力
                Thread.sleep(100);
            }

        } catch (Exception e) {
            log.error("批量迁移优化失败", e);
            throw new MigrationException("批量迁移优化失败", e);
        }
    }

    /**
     * 处理批次数据
     */
    private void processBatchData(String tableName, List<Map<String, Object>> batchData) {
        try {
            // 使用并行流处理
            batchData.parallelStream().forEach(data -> {
                try {
                    // 处理单条数据
                    processSingleData(tableName, data);
                } catch (Exception e) {
                    log.error("处理单条数据失败", e);
                }
            });

        } catch (Exception e) {
            log.error("处理批次数据失败", e);
        }
    }

    /**
     * 处理单条数据
     */
    private void processSingleData(String tableName, Map<String, Object> data) {
        try {
            // 检查数据是否已存在
            String primaryKey = data.get("id").toString();
            Object cachedResult = getCachedMigrationResult(tableName, primaryKey);

            if (cachedResult != null) {
                return; // 已处理过
            }

            // 处理数据
            Object result = processData(tableName, data);

            // 缓存结果
            cacheMigrationResult(tableName, primaryKey, result);

        } catch (Exception e) {
            log.error("处理单条数据失败", e);
        }
    }

    /**
     * 处理数据
     */
    private Object processData(String tableName, Map<String, Object> data) {
        // 实现数据处理逻辑
        return data;
    }

    /**
     * 预热迁移缓存
     */
    @PostConstruct
    public void warmupMigrationCache() {
        try {
            // 预热常用表
            List<String> commonTables = Arrays.asList("user", "order", "product", "post", "comment");

            for (String tableName : commonTables) {
                try {
                    // 预热表结构信息
                    cacheMigrationResult(tableName, "structure", getTableStructure(tableName));
                } catch (Exception e) {
                    log.error("预热迁移缓存失败: {}", tableName, e);
                }
            }

        } catch (Exception e) {
            log.error("预热迁移缓存失败", e);
        }
    }

    /**
     * 获取表结构
     */
    private Object getTableStructure(String tableName) {
        // 实现获取表结构的逻辑
        return new Object();
    }

    /**
     * 清理过期缓存
     */
    @Scheduled(fixedRate = 300000) // 5分钟
    public void cleanupExpiredCache() {
        try {
            // 清理本地缓存
            localCache.cleanUp();

            // 清理Redis过期缓存
            cleanupRedisExpiredCache();

        } catch (Exception e) {
            log.error("清理过期缓存失败", e);
        }
    }

    /**
     * 清理Redis过期缓存
     */
    private void cleanupRedisExpiredCache() {
        try {
            Set<String> cacheKeys = redisTemplate.keys("redis_cache:" + MIGRATION_CACHE_PREFIX + "*");

            for (String key : cacheKeys) {
                Long ttl = redisTemplate.getExpire(key);
                if (ttl != null && ttl <= 0) {
                    redisTemplate.delete(key);
                }
            }

        } catch (Exception e) {
            log.error("清理Redis过期缓存失败", e);
        }
    }
}

5.2 监控告警

5.2.1 监控指标

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
/**
 * 数据迁移监控指标
 */
@Component
public class DataMigrationMetrics {

    private final MeterRegistry meterRegistry;

    public DataMigrationMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }

    /**
     * 记录数据迁移次数
     */
    public void recordDataMigrationCount(String tableName, String operation) {
        Counter.builder("data.migration.count")
                .description("数据迁移次数")
                .tag("table_name", tableName)
                .tag("operation", operation)
                .register(meterRegistry)
                .increment();
    }

    /**
     * 记录数据迁移时间
     */
    public void recordDataMigrationTime(String tableName, String operation, long duration) {
        Timer.builder("data.migration.time")
                .description("数据迁移时间")
                .tag("table_name", tableName)
                .tag("operation", operation)
                .register(meterRegistry)
                .record(duration, TimeUnit.MILLISECONDS);
    }

    /**
     * 记录数据迁移成功率
     */
    public void recordDataMigrationSuccess(String tableName, String operation) {
        Counter.builder("data.migration.success")
                .description("数据迁移成功次数")
                .tag("table_name", tableName)
                .tag("operation", operation)
                .register(meterRegistry)
                .increment();
    }

    /**
     * 记录数据迁移失败次数
     */
    public void recordDataMigrationFailure(String tableName, String operation) {
        Counter.builder("data.migration.failure")
                .description("数据迁移失败次数")
                .tag("table_name", tableName)
                .tag("operation", operation)
                .register(meterRegistry)
                .increment();
    }

    /**
     * 记录数据迁移进度
     */
    public void recordDataMigrationProgress(String tableName, double progress) {
        Gauge.builder("data.migration.progress")
                .description("数据迁移进度")
                .tag("table_name", tableName)
                .register(meterRegistry, progress);
    }

    /**
     * 记录数据迁移速度
     */
    public void recordDataMigrationSpeed(String tableName, double speed) {
        Gauge.builder("data.migration.speed")
                .description("数据迁移速度")
                .tag("table_name", tableName)
                .register(meterRegistry, speed);
    }

    /**
     * 记录数据一致性校验
     */
    public void recordDataConsistencyCheck(String tableName, boolean isConsistent) {
        Counter.builder("data.consistency.check")
                .description("数据一致性校验")
                .tag("table_name", tableName)
                .tag("is_consistent", String.valueOf(isConsistent))
                .register(meterRegistry)
                .increment();
    }
}

5.2.2 告警规则

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
# prometheus-rules.yml
groups:
- name: data_migration_alerts
  rules:
  - alert: HighDataMigrationFailureRate
    expr: rate(data_migration_failure[5m]) / (rate(data_migration_success[5m]) + rate(data_migration_failure[5m])) > 0.1
    for: 2m
    labels:
      severity: warning
    annotations:
      summary: "数据迁移失败率过高"
      description: "数据迁移失败率超过10%，当前值: {{ $value }}"
  
  - alert: LowDataMigrationSpeed
    expr: data_migration_speed < 100
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "数据迁移速度过慢"
      description: "数据迁移速度低于100条/秒，当前值: {{ $value }}"
  
  - alert: DataMigrationProgressStalled
    expr: increase(data_migration_progress[10m]) == 0
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "数据迁移进度停滞"
      description: "数据迁移进度在10分钟内无变化"
  
  - alert: DataConsistencyCheckFailed
    expr: rate(data_consistency_check{is_consistent="false"}[5m]) > 0
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "数据一致性校验失败"
      description: "数据一致性校验失败，请立即检查"
  
  - alert: HighDataMigrationTime
    expr: data_migration_time{quantile="0.95"} > 5000
    for: 2m
    labels:
      severity: warning
    annotations:
      summary: "数据迁移时间过长"
      description: "数据迁移时间P95超过5秒，当前值: {{ $value }}ms"

六、总结

数据迁移作为分库分表架构演进中的重要环节，通过合理的平滑迁移策略，能够在不影响业务运行的情况下完成数据迁移。本文从数据迁移策略到平滑迁移方案，从基础实现到企业级应用，系统梳理了数据迁移的完整解决方案。

6.1 关键要点

  1. 迁移策略：双写策略、读写分离、灰度迁移等策略各有特点，需要根据业务场景选择合适的策略
  2. 数据同步：实时同步、增量同步、全量同步等同步方式需要根据数据特征选择
  3. 一致性保证：数据校验、冲突解决、事务保证等机制确保数据一致性
  4. 性能优化：通过缓存、批量处理等手段提高迁移性能
  5. 监控告警：建立完善的监控体系，及时发现和处理问题

6.2 最佳实践

  1. 策略选择：根据业务特征、数据特征、性能需求选择合适的数据迁移策略
  2. 平滑迁移：使用双写、读写分离等策略实现平滑迁移
  3. 数据同步：使用实时同步、增量同步等方式保证数据一致性
  4. 性能优化：使用缓存和批量处理提高迁移性能
  5. 监控告警：建立完善的监控体系，确保迁移过程稳定运行

通过以上措施，可以构建一个高效、稳定、可扩展的数据迁移方案，为企业的各种业务场景提供数据迁移支持。