使用队列异步处理CSV大文件导入是标准解法：上传后仅保存文件并推入队列，由消费者流式读取、分批事务入库、记录错误日志，并通过任务表实现进度反馈。
ThinkPHP 做 CSV 大文件导入时，直接同步读取+写入极易导致超时、内存溢出或请求中断。用 queue 队列异步处理 是标准解法：把“上传→解析→入库”三步拆开，前端上传后立即返回，后台由队列消费者分批处理，安全又可控。
一、上传文件并推入队列
用户上传 CSV 后，不立即解析，只保存原始文件（如到 runtime/upload/），再将关键参数推入队列：
- 文件路径（绝对路径或相对路径+唯一标识）
- 目标数据表名、字段映射规则（如
['name' => 0, 'email' => 1]） - 是否跳过首行、编码格式（UTF-8 / GBK）、分隔符（逗号/制表符）
- 操作人 ID、批次号（用于日志和状态查询）
示例代码（控制器中）：
use think\facade\Queue;
// …… 文件保存后
$data = [
'file_path' => $savePath,
'table' => 'user_import',
'mapping' => ['username' => 0, 'mobile' => 1, 'email' => 2],
'skip_first'=> true,
'encoding' => 'UTF-8',
'batch_no' => date('YmdHis') . uniqid(),
];
Queue::push('app\job\CsvImportJob', $data, 'import_queue');
二、编写队列任务类（CsvImportJob）
在 app/job/CsvImportJob.php 中定义任务逻辑，重点做三件事：流式读取、分批入库、异常防护。
立即学习“PHP免费学习笔记（深入）”；
Miller CSV TSV JSON 数据处理器
下载
Miller (mlr) 是一个命令行工具，用于查询、整形和重新格式化名称索引数据，如 CSV、TSV、JSON 和 JSON Lines。它将 awk、sed、cut、join 和 sort 的功能整合到一个专为结构化数据处理而构建的单一工具中。
- 使用
fopen($filePath, 'r')+fgetcsv()流式读取，避免全量加载 - 每 100–500 行为一批，调用
Db::transaction()批量插入，提升性能 - 对空行、列数不匹配、非法字符做跳过或记录错误日志（不中断整个任务）
- 插入失败时，记录错误行号与内容到
import_error_log表，便于追溯
关键逻辑片段：
public function fire(Job $job, $data)
{
$fp = fopen($data['file_path'], 'r');
if (!$fp) {
Log::error('CSV file not readable', $data);
$job->delete();
return;
}
$line = 0;
$batch = [];
while (($row = fgetcsv($fp, 0, ',')) !== false) {
$line++;
if ($data['skip_first'] && $line === 1) continue;
// 校验列数、过滤空行
if (empty($row) || count($row) < count($data['mapping'])) {
continue;
}
$record = [];
foreach ($data['mapping'] as $field => $index) {
$record[$field] = $index < count($row) ? trim($row[$index]) : '';
}
$batch[] = $record;
if (count($batch) >= 200) {
Db::name($data['table'])->insertAll($batch);
$batch = [];
}
}
if (!empty($batch)) {
Db::name($data['table'])->insertAll($batch);
}
fclose($fp);
$job->delete(); // 成功后删除任务
}
三、配置与启动队列监听器
确保 config/queue.php 已启用 Redis 或数据库驱动（推荐 Redis）：
'connector' => 'redis',
'connections' => [
'redis' => [
'type' => 'redis',
'queue' => 'import_queue',
'host' => env('REDIS_HOST', '127.0.0.1'),
'port' => env('REDIS_PORT', 6379),
'password' => env('REDIS_PASSWORD', ''),
],
],
启动常驻进程监听指定队列：
php think queue:work --queue import_queue --daemon --tries 3
-
--daemon：守护模式，持续运行 -
--tries 3：失败最多重试 3 次，避免死循环 - 建议配合
supervisor管理进程，防止意外退出
四、前端交互与进度反馈
用户上传后，返回一个 task_id 或 batch_no，前端可轮询接口获取状态：
- 新建一张
import_task表，记录batch_no、status（pending/running/success/failed）、processed_rows、error_count、created_at - 任务类中每处理完一批，更新该记录（可用
Cache::inc()或 DB 更新） - 提供 API 如
/api/import/status?batch_no=xxx返回实时进度
用户看到“已提交，正在后台处理…”而非卡死页面，体验更专业。
