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

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

批量导入的核心不是把 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 账号只有目标库的必要权限。不要拿生产文件直接试第一遍，先用几十行样本覆盖空值、中文、重复编号和日期格式错误。

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

本地运行后，按三组数字验收

准备一个包含 1,200 行数据的测试文件，其中故意放 3 行日期错误、2 行重复 `external_no`。验收不要只看页面上的“导入完成”，至少记录以下三组数字：

• 读取行数：去掉表头后实际扫描了多少行。
• 成功行数：事务提交后确实写入的数量。
• 失败行数：格式失败、唯一键冲突和批次异常分别多少行。

成功行数加失败行数应等于读取行数；如果某批异常回滚，成功数不能把那一批已经尝试过的行算进去。数据库侧再执行 `SELECT COUNT(*)`，并检查 `uk_customer_external_no` 没有重复。导入脚本应输出批次号和耗时，方便比较 100、500、1000 行批次的差异。

常见问题：内存、重复导入和批次大小

为什么不建议先用 file 读取整个 CSV？

`file()` 会把每行都放进 PHP 数组，数据量变大时内存峰值很快超过 `memory_limit`。流式读取让内存主要由当前行、预编译语句和当前批次计数占用。

导入重跑会不会插入重复客户？

`external_no` 需要唯一索引。是否跳过重复、更新已有记录，应该由业务明确选择；不要依赖应用层先查询再插入来“猜测”唯一性。

500 行是不是固定最佳值？

不是。先用 100、500、1000 行做测试，观察单批耗时、锁等待、内存峰值和失败重试成本。事务越大，吞吐可能更高，但回滚代价也更大。

验收清单

真正接入定时任务前，先用测试库跑一遍：CSV 表头已校验；空字段和日期格式有明确错误；每批有提交或回滚；失败记录能定位到行号；数据库唯一键已建立；重新导入同一文件的结果符合预期；最终成功数、失败数和数据库行数能够对上。