当前位置：首页 > 文章列表 > 文章 > php教程 > PHP mysqli 批量导入 CSV 怎么控制内存：分批事务、参数绑定与失败回滚

PHP mysqli 批量导入 CSV 怎么控制内存：分批事务、参数绑定与失败回滚

来源：17golang原创 2026-07-27 13:26:54 0浏览收藏

手里有一个 300 MB 的客户 CSV，最容易踩的坑不是 CSV 分隔符，而是把整份文件读进数组后才开始写库。PHP 进程会先吃掉一大块内存，写到一半又遇到一行格式错误，最后既不知道哪一批成功，也很难安全重跑。更稳的做法是让 `fgetcsv()` 一行一行产出记录，每 500 行组成一个事务，批次成功就提交，批次失败就回滚并记录行号。

批量导入的核心不是把 SQL 拼得更长，而是把“读取、写入、提交、复查”切成可控的小批次：单批 500 行只是起点，最终应按内存、锁等待和单批耗时调整。

要点速览

用 `fgetcsv()` 流式读取，避免把 CSV 全部加载到 PHP 数组。
预编译一条 `mysqli` INSERT，循环绑定每行参数，减少 SQL 拼接和类型混乱。
每 500 行开启一次事务；批次内出错立即回滚，保留失败行号和错误文本。
验收时同时核对导入计数、失败记录和数据库中的唯一键结果，不能只看脚本退出码。

先把 CSV 导入任务的边界定清楚

这个小工具只负责读取已经存在的 CSV，并把数据写入 `customer_import` 表。上传、权限和文件病毒扫描属于上游流程，不能混在一次导入里。示例文件有四列：`external_no`、`name`、`email`、`joined_on`；其中 `external_no` 建唯一索引，用它避免重复导入。

CREATE TABLE customer_import (
  id BIGINT UNSIGNED PRIMARY KEY AUTO_INCREMENT,
  external_no VARCHAR(40) NOT NULL,
  name VARCHAR(80) NOT NULL,
  email VARCHAR(160) NOT NULL,
  joined_on DATE NOT NULL,
  UNIQUE KEY uk_customer_external_no (external_no)
);

运行前先确认 PHP 已启用 `mysqli`，MySQL 账号只有目标库的必要权限。不要拿生产文件直接试第一遍，先用几十行样本覆盖空值、中文、重复编号和日期格式错误。

PHP mysqli CSV 导入从 fgetcsv 读取、分成 500 行批次并写入 customer_import 的分层路径

用 fgetcsv 和 mysqli 组成 500 行小批次

下面的核心代码故意保持短：文件句柄一次只保留当前行，预编译语句只创建一次，事务边界围绕一个批次。`bind_param()` 的类型串是 `ssss`，因为本例把四列先按字符串接收，日期格式检查通过后再写入 MySQL。

set_charset('utf8mb4');
$db->report_mode = MYSQLI_REPORT_ERROR | MYSQLI_REPORT_STRICT;

$input = fopen(__DIR__ . '/customers.csv', 'rb');
$insert = $db->prepare(
    'INSERT INTO customer_import (external_no, name, email, joined_on)
     VALUES (?, ?, ?, ?)'
);

$batchSize = 500;
$batchRows = 0;
$totalRows = 0;
$failedRows = [];
$line = 1;

fgetcsv($input); // 跳过表头
$db->begin_transaction();

while (($row = fgetcsv($input)) !== false) {
    $line++;
    if (count($row) !== 4 || !check_date($row[3])) {
        $failedRows[] = ['line' => $line, 'reason' => '字段数量或日期格式不正确'];
        continue;
    }

    [$externalNo, $name, $email, $joinedOn] = array_map('trim', $row);
    $insert->bind_param('ssss', $externalNo, $name, $email, $joinedOn);
    $insert->send_long_data(0, $externalNo);
    $insert->store_result();
    $batchRows++;
    $totalRows++;

    if ($batchRows === $batchSize) {
        $db->commit();
        $batchRows = 0;
        $db->begin_transaction();
    }
}

if ($batchRows > 0) {
    $db->commit();
}

$insert->close();
fclose($input);
printf("imported=%d failed=%d\n", $totalRows, count($failedRows));

这里的 `store_result()` 只是让语句结果及时收束；真正的批次控制由 `begin_transaction()` 和 `commit()` 完成。示例里的 `check_date()` 应替换成项目已有的日期校验函数，并在函数内部明确使用 `Y-m-d` 规则。生产代码还应捕获数据库异常：任何一行触发唯一键冲突或连接异常，都要回滚当前批次，而不是继续写下一行。

失败时回滚当前批次，成功后留下可对账记录

批量导入最怕“半成功”。如果第 734 行有重复编号，而前 499 行已经写进数据库，下一次重跑就会出现一部分重复、一部分跳过的混乱状态。推荐把每批处理包在 `try/catch` 中：批次成功提交，异常则回滚，并把批次起止行和错误文本写到 `import_failures`。

$batchStart = 2;
try {
    $db->begin_transaction();
    // 读取并绑定本批次的 CSV 行
    // 每成功写入一行就递增 $batchRows
    $db->commit();
} catch (mysqli_sql_exception $error) {
    $db->rollback();
    error_log(json_encode([
        'batch_start' => $batchStart,
        'batch_rows' => $batchRows,
        'message' => $error->getMessage(),
    ], JSON_UNESCAPED_UNICODE));
    throw $error;
}

不要把 CSV 原文和数据库密码一起写进日志。日志只保留行号、外部编号的脱敏值、错误类型和批次编号即可。若业务允许“坏行跳过、好行继续”，就把单行校验错误和数据库异常分开：前者进入失败清单，后者回滚整批并终止任务。