PHP数据库批量操作处理_PHP批量插入更新优化技巧

INSERT INTO ... VALUES (), (), ...;

<?php
// 假设 $pdo 是一个已连接的 PDO 实例
$dataToInsert = [
    ['name' => 'Alice', 'email' => 'alice@example.com'],
    ['name' => 'Bob', 'email' => 'bob@example.com'],
    ['name' => 'Charlie', 'email' => 'charlie@example.com'],
    // ... 更多数据
];

$tableName = 'users';
$columns = implode(', ', array_keys($dataToInsert[0])); // 获取列名

$placeholders = [];
$values = [];
foreach ($dataToInsert as $row) {
    $rowPlaceholders = [];
    foreach ($row as $key => $value) {
        $rowPlaceholders[] = '?'; // 为每个值使用占位符
        $values[] = $value;
    }
    $placeholders[] = '(' . implode(', ', $rowPlaceholders) . ')';
}

$sql = "INSERT INTO {$tableName} ({$columns}) VALUES " . implode(', ', $placeholders);

try {
    $pdo->beginTransaction(); // 开启事务
    $stmt = $pdo->prepare($sql);
    $stmt->execute($values);
    $pdo->commit(); // 提交事务
    echo "批量插入成功！";
} catch (PDOException $e) {
    $pdo->rollBack(); // 发生错误回滚事务
    echo "批量插入失败：" . $e->getMessage();
}
?>

CASE WHEN

UPDATE

<?php
// 假设 $pdo 是一个已连接的 PDO 实例
$dataToUpdate = [
    ['id' => 1, 'status' => 'active', 'updated_at' => date('Y-m-d H:i:s')],
    ['id' => 3, 'status' => 'inactive', 'updated_at' => date('Y-m-d H:i:s')],
    ['id' => 5, 'status' => 'pending', 'updated_at' => date('Y-m-d H:i:s')],
    // ... 更多数据
];

$tableName = 'products'; // 假设更新产品表
$idColumn = 'id'; // 用作唯一标识的列

$statusCases = [];
$updatedAtCases = [];
$ids = [];
$values = []; // 用于存储所有绑定值

foreach ($dataToUpdate as $item) {
    $id = $item[$idColumn];
    $ids[] = $id;

    // 为 status 字段构建 CASE WHEN 语句
    $statusCases[] = "WHEN {$idColumn} = ? THEN ?";
    $values[] = $id;
    $values[] = $item['status'];

    // 为 updated_at 字段构建 CASE WHEN 语句
    $updatedAtCases[] = "WHEN {$idColumn} = ? THEN ?";
    $values[] = $id;
    $values[] = $item['updated_at'];
}

// 确保 $ids 不为空，避免生成错误的 WHERE IN ()
if (empty($ids)) {
    echo "没有数据需要更新。";
    exit;
}

$sql = "UPDATE {$tableName} SET ";
$sql .= "status = (CASE " . implode(' ', $statusCases) . " ELSE status END), ";
$sql .= "updated_at = (CASE " . implode(' ', $updatedAtCases) . " ELSE updated_at END) ";
$sql .= "WHERE {$idColumn} IN (" . implode(', ', array_fill(0, count($ids), '?')) . ")";

// 将所有ID添加到绑定值列表的末尾
$values = array_merge($values, $ids);

try {
    $pdo->beginTransaction();
    $stmt = $pdo->prepare($sql);
    $stmt->execute($values);
    $pdo->commit();
    echo "批量更新成功！";
} catch (PDOException $e) {
    $pdo->rollBack();
    echo "批量更新失败：" . $e->getMessage();
}
?>

CASE WHEN

UPDATE

ON DUPLICATE KEY UPDATE

<?php
// 假设 $pdo 是一个已连接的 PDO 实例
$dataToUpsert = [
    ['id' => 1, 'name' => 'Alice', 'email' => 'alice_new@example.com'], // id=1存在，更新
    ['id' => 6, 'name' => 'Frank', 'email' => 'frank@example.com'], // id=6不存在，插入
    // ... 更多数据
];

$tableName = 'users';
$columns = implode(', ', array_keys($dataToUpsert[0]));

$placeholders = [];
$values = [];
foreach ($dataToUpsert as $row) {
    $rowPlaceholders = [];
    foreach ($row as $key => $value) {
        $rowPlaceholders[] = '?';
        $values[] = $value;
    }
    $placeholders[] = '(' . implode(', ', $rowPlaceholders) . ')';
}

// 构建 ON DUPLICATE KEY UPDATE 部分
$updateColumns = [];
foreach (array_keys($dataToUpsert[0]) as $col) {
    if ($col !== 'id') { // 假设 id 是主键或唯一键，不更新它本身
        $updateColumns[] = "{$col} = VALUES({$col})";
    }
}

$sql = "INSERT INTO {$tableName} ({$columns}) VALUES " . implode(', ', $placeholders);
if (!empty($updateColumns)) {
    $sql .= " ON DUPLICATE KEY UPDATE " . implode(', ', $updateColumns);
}

try {
    $pdo->beginTransaction();
    $stmt = $pdo->prepare($sql);
    $stmt->execute($values);
    $pdo->commit();
    echo "批量插入或更新成功！";
} catch (PDOException $e) {
    $pdo->rollBack();
    echo "批量插入或更新失败：" . $e->getMessage();
}
?>

ON DUPLICATE KEY UPDATE

$data = [
    ['col1' => 'val1_1', 'col2' => 'val1_2'],
    ['col1' => 'val2_1', 'col2' => 'val2_2'],
    // ...
];

prepare()

execute()

try {
    $pdo->beginTransaction(); // 开启事务，确保原子性
    $stmt = $pdo->prepare($sql);
    $stmt->execute($allValues); // 将所有绑定值一次性传入
    $pdo->commit(); // 提交事务
    echo "操作成功！";
} catch (PDOException $e) {
    $pdo->rollBack(); // 发生错误时回滚
    error_log("数据库批量操作失败: " . $e->getMessage()); // 记录错误
    echo "操作失败，请重试。";
}

$pdo->beginTransaction()

$pdo->commit()

$pdo->rollBack()

$chunkSize = 1000;
$chunks = array_chunk($largeDataSet, $chunkSize);

try {
    $pdo->beginTransaction();
    foreach ($chunks as $chunk) {
        // 根据 $chunk 构建 SQL 和绑定值，然后执行
        // ... (参考上面的构建SQL和执行部分)
        $stmt->execute($chunkValues);
    }
    $pdo->commit();
    echo "所有分块批量操作成功！";
} catch (PDOException $e) {
    $pdo->rollBack();
    error_log("分块批量操作失败: " . $e->getMessage());
    echo "操作失败，请重试。";
}

PDO

beginTransaction()

commit()

rollBack()

try-catch

PDO

PDOException

try {
    $pdo->beginTransaction();
    // ... 构建并执行批量SQL语句 ...
    $stmt->execute($allValues);
    $pdo->commit();
    // 成功后的逻辑
} catch (PDOException $e) {
    $pdo->rollBack(); // 捕获到异常，立即回滚
    // 错误处理逻辑：
    // 1. 记录日志：将错误信息 ($e->getMessage(), $e->getCode(), $e->getFile(), $e->getLine()) 写入日志文件。
    error_log("批量操作失败: " . $e->getMessage() . " SQL: " . $sql);
    // 2. 向用户反馈：给用户一个友好的错误提示，而不是直接暴露数据库错误。
    echo "系统繁忙，批量操作未能完成，请稍后再试或联系管理员。";
    // 3. 考虑重试机制：对于某些可恢复的错误（如死锁），可以设计有限次数的重试逻辑。
}

try {
    $pdo->beginTransaction(); // 大事务
    foreach ($chunks as $chunk) {
        // 构建并执行当前 $chunk 的批量SQL
        $stmt = $pdo->prepare($sqlForChunk);
        $stmt->execute($valuesForChunk);
    }
    $pdo->commit();
} catch (PDOException $e) {
    $pdo->rollBack(); // 任何一个分块失败，整个大事务回滚
    error_log("分块批量操作中途失败: " . $e->getMessage());
}