本文目录导读:

在PHP项目中保证数据迁移的一致性是关键问题,通常涉及数据完整性、事务一致性、并发控制和可回滚性,以下是系统性的解决方案和最佳实践:
核心原则:ACID 事务
使用数据库事务(如果数据库支持)
try {
$pdo->beginTransaction();
// 迁移步骤1:读取旧数据
$oldData = $pdo->query("SELECT * FROM old_table WHERE status = 'pending'")->fetchAll();
// 迁移步骤2:写入新表
foreach ($oldData as $row) {
$stmt = $pdo->prepare("INSERT INTO new_table (id, name, status) VALUES (?, ?, ?)");
$stmt->execute([$row['id'], $row['name'], 'migrated']);
}
// 迁移步骤3:更新状态
$pdo->exec("UPDATE old_table SET status = 'migrated' WHERE status = 'pending'");
$pdo->commit();
} catch (Exception $e) {
$pdo->rollBack();
// 记录错误日志
}
关键点:
- 所有操作在一个事务内:确保全部成功或全部失败
- 避免死锁:按固定顺序访问表(如:先锁旧表,再锁新表)
- 设置事务隔离级别:通常是
READ COMMITTED或REPEATABLE READ
分页+断点续传(处理大数据量)
当数据量很大时,单次事务可能超时或内存溢出。
const BATCH_SIZE = 1000;
$lastId = 0; // 从文件中读取上次中断点
$totalMigrated = 0;
while (true) {
try {
$pdo->beginTransaction();
// 使用游标或LIMIT分批处理
$rows = $pdo->query("SELECT * FROM old_table WHERE id > $lastId ORDER BY id LIMIT " . BATCH_SIZE)->fetchAll();
if (empty($rows)) break;
// 批量插入新表
$insertData = [];
foreach ($rows as $row) {
$insertData[] = "({$row['id']}, '{$row['name']}', 'migrated')";
$lastId = $row['id'];
}
$pdo->exec("INSERT INTO new_table (id, name, status) VALUES " . implode(',', $insertData));
// 标记旧表状态
$pdo->exec("UPDATE old_table SET status = 'migrated' WHERE id <= $lastId AND status = 'pending'");
$pdo->commit();
$totalMigrated += count($rows);
// 保存断点:$lastId 到文件或数据库
file_put_contents('migration_progress.txt', $lastId);
} catch (Exception $e) {
$pdo->rollBack();
// 指数退避重试
sleep(2);
}
}
优点:
- 支持大规模数据(百万级)
- 崩溃后可从中断点继续
- 每批都是独立事务,避免长事务锁定
数据校验(一致性检查)
迁移前校验
// 原始数据总行数
$oldCount = $pdo->query("SELECT COUNT(*) FROM old_table WHERE status = 'pending'")->fetchColumn();
echo "需迁移: $oldCount 行\n";
// 目标表预期行数(初始为0)
$newCount = $pdo->query("SELECT COUNT(*) FROM new_table WHERE status = 'migrated'")->fetchColumn();
echo "已迁移: $newCount 行\n";
迁移后校验(完整性验证)
// 使用校验和或哈希对比
$checksumOld = $pdo->query("SELECT CRC32(GROUP_CONCAT(id, name ORDER BY id)) FROM old_table WHERE status = 'migrated'")->fetchColumn();
$checksumNew = $pdo->query("SELECT CRC32(GROUP_CONCAT(id, name ORDER BY id)) FROM new_table WHERE status = 'migrated'")->fetchColumn();
if ($checksumOld !== $checksumNew) {
// 严重不一致,触发回滚或修复
throw new Exception("数据校验失败!");
}
批量验证(大型表)
// 抽样验证:每1000行做一次哈希校验
$sampleQuery = "SELECT id, name FROM old_table WHERE MOD(id, 1000) = 0 AND status = 'migrated'";
$result = $pdo->query($sampleQuery);
while ($row = $result->fetch()) {
$target = $pdo->prepare("SELECT name FROM new_table WHERE id = ?");
$target->execute([$row['id']]);
if ($target->fetchColumn() !== $row['name']) {
echo "数据不一致: ID={$row['id']}\n";
}
}
并发控制(避免写入冲突)
乐观锁(推荐)
// 旧表添加 version 字段
$row = $pdo->query("SELECT id, name, version FROM old_table WHERE status = 'pending' LIMIT 1")->fetch();
$affected = $pdo->exec("UPDATE old_table SET status = 'migrating', version = version + 1 WHERE id = {$row['id']} AND version = {$row['version']}");
if ($affected === 0) {
// 被其他进程修改,跳过或重试
continue;
}
悲观锁(不推荐长事务中使用)
$pdo->exec("SELECT id FROM old_table WHERE status = 'pending' FOR UPDATE NOWAIT");
// 锁定后再处理
幂等性设计
确保重复执行迁移不会导致重复数据(重要!)
// 方案1:使用REPLACE INTO
$pdo->exec("REPLACE INTO new_table (id, name, status) VALUES ({$id}, '{$name}', 'migrated')");
// 方案2:使用INSERT IGNORE或ON DUPLICATE KEY UPDATE
$pdo->exec("INSERT INTO new_table (id, name, status) VALUES ({$id}, '{$name}', 'migrated') ON DUPLICATE KEY UPDATE name = VALUES(name)");
// 方案3:先检查再插入
$exists = $pdo->query("SELECT 1 FROM new_table WHERE id = {$id}")->fetchColumn();
if (!$exists) {
$pdo->exec("INSERT INTO new_table ...");
}
最佳实践:为新表设置唯一索引或主键,配合ON DUPLICATE KEY UPDATE。
日志与监控
// 迁移日志表
CREATE TABLE migration_logs (
id INT AUTO_INCREMENT PRIMARY KEY,
batch_id INT,
old_id INT,
new_id INT,
status ENUM('pending', 'success', 'failed'),
error_message TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
// 记录每行迁移结果
$stmt = $pdo->prepare("INSERT INTO migration_logs (batch_id, old_id, status) VALUES (?, ?, 'failed')");
try {
// ...迁移逻辑...
$stmt->execute([$batchId, $row['id']]);
// 标记成功
$pdo->exec("UPDATE migration_logs SET status = 'success', new_id = $newId WHERE id = $logId");
} catch (Exception $e) {
$stmt->execute([$batchId, $row['id']]);
// 记录错误信息
$pdo->exec("UPDATE migration_logs SET error_message = '" . addslashes($e->getMessage()) . "' WHERE id = $logId");
}
完整迁移脚本模板
<?php
class DataMigration {
private $pdo;
private $batchId;
private $batchSize = 1000;
public function run() {
try {
$this->pdo = new PDO('mysql:host=localhost;dbname=test', 'user', 'pass');
$this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$this->batchId = time();
$lastId = $this->getLastProcessedId();
$total = 0;
while (true) {
try {
$this->pdo->beginTransaction();
$stmt = $this->pdo->prepare(
"SELECT * FROM old_table WHERE id > ? AND status = 'pending' ORDER BY id LIMIT ?"
);
$stmt->execute([$lastId, $this->batchSize]);
$rows = $stmt->fetchAll();
if (empty($rows)) break;
// 批量准备
$insertIds = [];
foreach ($rows as $row) {
$insertIds[] = $row['id'];
$this->pdo->exec(
"INSERT INTO new_table (id, name, status)
VALUES ({$row['id']}, '{$row['name']}', 'migrated')
ON DUPLICATE KEY UPDATE name = VALUES(name)"
);
$lastId = $row['id'];
}
// 更新旧表状态
$idsStr = implode(',', $insertIds);
$this->pdo->exec("UPDATE old_table SET status = 'migrated' WHERE id IN ($idsStr)");
$this->pdo->commit();
$total += count($rows);
$this->saveProgress($lastId);
echo "迁移进度: $total 行, 当前ID: $lastId\n";
} catch (Exception $e) {
$this->pdo->rollBack();
echo "错误: " . $e->getMessage() . ", 重试...\n";
usleep(500000); // 0.5秒后重试
}
}
echo "迁移完成! 共迁移 $total 行\n";
$this->verifyChecksum();
} catch (Exception $e) {
echo "致命错误: " . $e->getMessage() . "\n";
exit(1);
}
}
private function getLastProcessedId() {
$progress = @file_get_contents('migration_progress.txt');
return $progress ? (int)$progress : 0;
}
private function saveProgress($lastId) {
file_put_contents('migration_progress.txt', $lastId);
}
private function verifyChecksum() {
// 校验逻辑...
}
}
$migration = new DataMigration();
$migration->run();
一致性保证的核心措施
| 措施 | 说明 |
|---|---|
| 事务 | 所有操作在一个事务内,确保原子性 |
| 分批处理 | 每批单独事务,支持断点续传 |
| 幂等性 | 重复执行不会破坏数据 |
| 数据校验 | 迁移前后对比行数、哈希值 |
| 并发控制 | 乐观锁或悲观锁防止冲突 |
| 日志记录 | 记录每行的迁移状态 |
| 回滚机制 | 出错时自动回滚当前批次 |
对于极端重要的数据,还可以采用双写模式(迁移期间同时写入新旧表),但会增加复杂度,一般情况下,上述方案已能保证99.9%以上的一致性。