PHP大型文件高效处理：分行读取与即时处理策略

<?php

class FileReader
{
    /**
     * 读取文件并返回所有解析后的行数据
     * @param string $file 文件路径
     * @return array
     * @throws Exception
     */
    public function read(string $file): array
    {
        $fileHandle = fopen($file, "r");

        if ($fileHandle === false) {
            throw new Exception('无法获取文件句柄: ' . $file);
        }

        $lines = [];
        while (!feof($fileHandle)) {
            $line = fgets($fileHandle);
            if ($line !== false) { // 确保读取到有效行
                $decodedLine = json_decode($line);
                if ($decodedLine !== null) { // 确保JSON解码成功
                    $lines[] = $decodedLine;
                }
            }
        }

        fclose($fileHandle);
        return $lines;
    }
}

// 示例用法
$reader = new FileReader();
try {
    $users = $reader->read('users.jsonl'); // users.jsonl 包含多行 {"user_id":1,"user_name":"Alex"}
    // ... 后续处理 $users 数组 ...
    echo "文件读取完成，共 " . count($users) . " 条记录。\n";
} catch (Exception $e) {
    echo "错误: " . $e->getMessage() . "\n";
}

?>

<?php

class StreamFileReader
{
    /**
     * 逐行读取文件，并对每行数据执行回调函数
     * @param string $file 文件路径
     * @param callable $rowProcessor 处理单行数据的回调函数
     * @throws Exception
     */
    public function readAndProcess(string $file, callable $rowProcessor): void
    {
        $fileHandle = fopen($file, "r");

        if ($fileHandle === false) {
            throw new Exception('无法获取文件句柄: ' . $file);
        }

        while (!feof($fileHandle)) {
            $line = fgets($fileHandle);
            if ($line !== false) {
                $decodedLine = json_decode($line);
                if ($decodedLine !== null) {
                    // 将解码后的单行数据传递给回调函数进行处理
                    $rowProcessor($decodedLine);
                } else {
                    // 记录JSON解码错误，但不中断整个流程
                    error_log("Warning: Could not decode JSON line: " . $line);
                }
            }
        }

        fclose($fileHandle);
    }
}

?>

<?php

// 实例化流式文件读取器
$streamReader = new StreamFileReader();

/**
 * 处理并写入JSON数据到CSV文件
 * @param string $inputFilename 输入JSON行文件路径
 * @param string $outputFilename 输出CSV文件路径
 * @throws Exception
 */
function processAndWriteJsonToCsv(string $inputFilename, string $outputFilename): void
{
    global $streamReader; // 假设 StreamFileReader 实例在全局或通过其他方式可访问

    $writer = fopen($outputFilename, 'w');
    if ($writer === false) {
        throw new Exception('无法创建或打开输出CSV文件: ' . $outputFilename);
    }

    // 写入CSV头部
    fputcsv($writer, ['User ID', 'User Name']);

    try {
        $streamReader->readAndProcess($inputFilename, function ($row) use ($writer) {
            // 对单行数据进行处理
            $processedRow = [
                $row->user_id,
                strtoupper($row->user_name)
            ];
            // 将处理后的数据立即写入CSV文件
            fputcsv($writer, $processedRow);
        });
    } finally {
        // 无论成功与否，确保关闭文件句柄
        fclose($writer);
    }
    echo "数据已成功处理并导出到 " . $outputFilename . "\n";
}

// 示例调用
try {
    processAndWriteJsonToCsv('users.jsonl', 'output.csv');
} catch (Exception $e) {
    echo "处理失败: " . $e->getMessage() . "\n";
}

?>

<?php

class GeneratorFileReader
{
    /**
     * 使用生成器逐行读取并解码JSON文件
     * @param string $file 文件路径
     * @return Generator
     * @throws Exception
     */
    public function readJsonLines(string $file): Generator
    {
        $fileHandle = fopen($file, "r");

        if ($fileHandle === false) {
            throw new Exception('无法获取文件句柄: ' . $file);
        }

        try {
            while (!feof($fileHandle)) {
                $line = fgets($fileHandle);
                if ($line !== false) {
                    $decodedLine = json_decode($line);
                    if ($decodedLine !== null) {
                        yield $decodedLine; // 每次 yield 一行数据
                    } else {
                        error_log("Warning: Could not decode JSON line: " . $line);
                    }
                }
            }
        } finally {
            // 确保文件句柄在生成器完成或被销毁时关闭
            fclose($fileHandle);
        }
    }
}

// 使用生成器处理并写入CSV
$generatorReader = new GeneratorFileReader();
$outputFilename = 'output_generator.csv';
$writer = fopen($outputFilename, 'w');

if ($writer === false) {
    throw new Exception('无法创建或打开输出CSV文件: ' . $outputFilename);
}

fputcsv($writer, ['User ID', 'User Name']);

try {
    foreach ($generatorReader->readJsonLines('users.jsonl') as $row) {
        $processedRow = [
            $row->user_id,
            strtoupper($row->user_name)
        ];
        fputcsv($writer, $processedRow);
    }
    echo "数据已通过生成器成功处理并导出到 " . $outputFilename . "\n";
} catch (Exception $e) {
    echo "处理失败: " . $e->getMessage() . "\n";
} finally {
    fclose($writer);
}

?>