PHP中大型文件的高效读取与流式处理实践

public function read(string $file) : array
{
    $fileHandle = fopen($file, "r");

    if ($fileHandle === false) {
        throw new Exception('Could not get file handle for: ' . $file);
    }

    $lines = [];
    while (!feof($fileHandle)) {
        $lineContent = fgets($fileHandle);
        if ($lineContent !== false) { // 确保读取到内容
            $lines[] = json_decode($lineContent);
        }
    }

    fclose($fileHandle);
    return $lines;
}

public function processInput(array $users): array
{
    $data = [];
    foreach ($users as $key => $user) {
        // 假设 $user 是一个对象，例如 {"user_id": 1, "user_name": "Alex"}
        if (is_object($user) && property_exists($user, 'user_id') && property_exists($user, 'user_name')) {
            $data[$key]['user_id'] = $user->user_id;
            $data[$key]['user_name'] = strtoupper($user->user_name);
        }
    }
    return $data;
    // 之后可能会调用函数将 $data 导出到 CSV
}

/**
 * 逐行读取文件并使用回调函数处理每行数据。
 *
 * @param string $file 要读取的文件路径。
 * @param callable $rowProcessor 用于处理每行数据的回调函数。
 *                                 回调函数应接受一个参数，即解码后的行数据。
 * @throws Exception 如果无法打开文件。
 */
public function readLazy(string $file, callable $rowProcessor) : void
{
    $fileHandle = fopen($file, "r");

    if ($fileHandle === false) {
        throw new Exception('Could not get file handle for: ' . $file);
    }

    while (!feof($fileHandle)) {
        $lineContent = fgets($fileHandle);
        if ($lineContent === false) {
            // 文件结束或读取错误，跳过
            continue;
        }

        $decodedLine = json_decode($lineContent);
        // 检查JSON解码是否成功，并确保不是空行引起的null
        if ($decodedLine !== null || trim($lineContent) === '') {
            $rowProcessor($decodedLine);
        }
    }

    fclose($fileHandle);
}

/**
 * 从JSON文件读取数据，处理后直接写入CSV文件。
 *
 * @param string $inputFilename 输入的JSON文件路径。
 * @param string $outputFilename 输出的CSV文件路径。
 */
public function processAndWriteJsonToCsv(string $inputFilename, string $outputFilename): void
{
    $writer = fopen($outputFilename, 'w');
    if ($writer === false) {
        throw new Exception('Could not open output CSV file for writing: ' . $outputFilename);
    }

    // 写入CSV头部
    fputcsv($writer, ['User ID', 'User Name']); // 假设CSV有这些列

    $this->readLazy($inputFilename, function ($row) use ($writer) {
        // 确保 $row 是一个有效的对象且包含所需属性
        if (is_object($row) && property_exists($row, 'user_id') && property_exists($row, 'user_name')) {
            // 进行数据处理
            $processedRow = [
                $row->user_id,
                strtoupper($row->user_name)
            ];
            // 将处理后的单行数据直接写入CSV文件
            fputcsv($writer, $processedRow);
        } else {
            // 可选：处理无效行或记录错误
            error_log("Skipping invalid row: " . json_encode($row));
        }
    });

    fclose($writer);
}

$allProcessedLines = [];
$this->readLazy($inputFilename, function ($row) use (&$allProcessedLines) {
    // 假设这里也进行同样的简单处理
    if (is_object($row) && property_exists($row, 'user_id') && property_exists($row, 'user_name')) {
        $processedRow = [
            'user_id' => $row->user_id,
            'user_name' => strtoupper($row->user_name)
        ];
        $allProcessedLines[] = $processedRow;
    }
});

// 此时 $allProcessedLines 包含了所有处理后的数据，但请注意内存消耗