PHP数据迁移怎么保证一致性

wen PHP项目 2026-07-10 207

本文目录导读：

PHP数据迁移怎么保证一致性

核心原则：ACID 事务
分页+断点续传（处理大数据量）
数据校验（一致性检查）
并发控制（避免写入冲突）
幂等性设计
日志与监控
完整迁移脚本模板
一致性保证的核心措施

在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();

一致性保证的核心措施

措施	说明
事务	所有操作在一个事务内，确保原子性
分批处理	每批单独事务，支持断点续传
幂等性	重复执行不会破坏数据
数据校验	迁移前后对比行数、哈希值
并发控制	乐观锁或悲观锁防止冲突
日志记录	记录每行的迁移状态
回滚机制	出错时自动回滚当前批次