PHP generator để xử lý dữ liệu lớn mà không tốn memory: yield, yield from, send() và LazyCollection

Job import ban đêm chạy ngon suốt một năm. Rồi một đối tác bắt đầu gửi nguyên lịch sử đơn hàng thay vì phần thay đổi: một file CSV 2,1 GB. Job chết ngay dòng đầu với Allowed memory size of 536870912 bytes exhausted. Có người nâng memory_limit lên 4G, job chạy được xa hơn chút, rồi container bị OOM-kill kéo theo luôn queue worker.

Code thì quen thuộc: đọc file vô một mảng, map ra mảng khác, rồi lặp. Memory tăng theo kích thước input, nên file nào lớn hơn là thêm một sự cố. Generator lật ngược chuyện đó: xử lý từng dòng một và memory đứng yên dù file lớn cỡ nào. Generator có trong PHP từ 5.5 mà tới giờ vẫn ít người xài, nên đây là trọn bộ đồ nghề: yield, key, yield from, send(), và cách LazyCollection của Laravel xây trên nền đó.

Vì sao bản ngây thơ nổ tung

// Loads the whole file, then builds a second full copy as arrays.
function importOrders(string $path): int
{
    $lines = file($path, FILE_IGNORE_NEW_LINES);   // 2 GB of strings
    $header = str_getcsv(array_shift($lines), ',', '"', '');
    $rows = array_map(
        fn (string $line) => array_combine($header, str_getcsv($line, ',', '"', '')),
        $lines,
    );                                              // another few GB of arrays

    foreach ($rows as $row) {
        // insert...
    }

    return count($rows);
}

String và array trong PHP có overhead. File CSV 2 GB load bằng file() tốn hơn 2 GB, rồi biến mỗi dòng thành mảng kết hợp lại nhân lên mấy lần nữa. Ngay cả array_map trên file nhỏ hơn cũng giữ cả mảng nguồn lẫn mảng kết quả cùng lúc. Cách sửa không phải thêm RAM. Cách sửa là đừng bao giờ giữ quá một dòng.

Generator chạy ra sao

Hàm nào có chứa yield thì thành generator function. Gọi nó không chạy dòng code nào; nó trả về một object Generator. Mỗi lần bên tiêu thụ xin giá trị kế tiếp, thân hàm chạy tiếp từ chỗ đã dừng, chạy tới yield kế, giao ra một giá trị, rồi đứng lại với biến cục bộ còn nguyên.

Trong memory chỉ có dòng hiện tại. File handle, header và bộ đếm dòng nằm trong stack frame đang tạm dừng của generator. Khi bên tiêu thụ dừng lại (một break, một exception, hay generator ra khỏi scope), PHP hủy generator và chạy các khối finally còn treo, nên file vẫn được đóng đàng hoàng.

Cài đặt: stream một file CSV

/**
 * Stream a CSV file one row at a time.
 *
 * @return Generator<int, array<string, string>> line number => associative row
 */
function readCsv(string $path): Generator
{
    $handle = fopen($path, 'rb');
    if ($handle === false) {
        throw new RuntimeException("Cannot open $path");
    }

    try {
        $header = fgetcsv($handle, null, ',', '"', '');
        $line = 1;

        while (($fields = fgetcsv($handle, null, ',', '"', '')) !== false) {
            $line++;
            if ($fields === [null]) {
                continue; // blank line
            }
            yield $line => array_combine($header, $fields);
        }
    } finally {
        fclose($handle); // runs even if the consumer stops early
    }
}

foreach (readCsv('/data/orders.csv') as $line => $row) {
    if ($row['status'] === 'cancelled') {
        continue;
    }
    // process one row; previous rows are already garbage
}

echo round(memory_get_peak_usage() / 1024 / 1024, 1), " MB\n";

Vài chi tiết quan trọng khi chạy production:

• Key. yield $key => $value gán key. Không gán thì key là số nguyên tự tăng giống list. Lấy số dòng làm key là có luôn thông báo lỗi chính xác mà không tốn công.
• try/finally. Dọn dẹp thì bỏ vô finally, đừng để sau vòng lặp. Nếu bên tiêu thụ break sớm, code sau vòng lặp không bao giờ chạy, còn finally thì chạy khi generator bị hủy.
• Escape CSV tường minh. Từ PHP 8.4, dựa vô giá trị mặc định của tham số $escape trong fgetcsv() và str_getcsv() bị deprecated. Truyền '' để có hành vi đúng RFC 4180.
• Dòng trống. fgetcsv() trả về [null] cho dòng rỗng. Bỏ qua nó rõ ràng, không thì array_combine() sẽ throw ValueError vì số phần tử không khớp.

Với bản này, peak memory cho file 2,1 GB khoảng 2 MB, y như với file 2 KB.

Ghép generator bằng yield from

yield from ủy quyền cho một generator khác (hoặc bất kỳ iterable nào) và chuyển thẳng các giá trị của nó ra ngoài. Giá trị của nó khi dùng như biểu thức là giá trị return của generator bên trong, một cách gọn để cộng dồn thống kê trong lúc stream:

/** Concatenate several CSV exports into one stream and report the total. */
function readAll(array $paths): Generator
{
    $total = 0;
    foreach ($paths as $path) {
        // yield from forwards every key/value of the inner generator
        // and evaluates to the inner generator's return value.
        $total += yield from countingReader($path);
    }

    return $total;
}

function countingReader(string $path): Generator
{
    $n = 0;
    foreach (readCsv($path) as $line => $row) {
        $n++;
        yield $line => $row;
    }

    return $n;
}

$stream = readAll(['/data/2026-10.csv', '/data/2026-11.csv']);
foreach ($stream as $line => $row) {
    // keys restart at 2 for each file: they are NOT unique across files
}
echo $stream->getReturn(), " rows\n";

// Pitfall: iterator_to_array() keeps keys by default, so duplicate keys overwrite.
$rows = iterator_to_array(readAll(['/data/a.csv', '/data/b.csv']), preserve_keys: false);

$generator->getReturn() chỉ gọi được sau khi generator chạy xong; gọi sớm là throw. Và coi chừng key: yield from chuyển nguyên key của generator bên trong, nên hai file đều sinh ra key 2. iterator_to_array() mặc định giữ key và âm thầm ghi đè, đó là cách mấy bug kiểu "import chỉ còn dòng của file cuối" ra đời. Truyền preserve_keys: false hoặc đừng gom thành mảng luôn.

Generator hai chiều với send()

Generator còn nhận được giá trị. $gen->send($value) cho generator chạy tiếp và biến biểu thức yield đang dừng thành $value. Vậy là generator thành một coroutine nhỏ, rất tiện để làm một chỗ gom batch:

/**
 * A coroutine: receives rows through send(), flushes them in batches,
 * and returns the number of rows written when it receives null.
 */
function batchWriter(Closure $flush, int $size = 1000): Generator
{
    $buffer = [];
    $written = 0;

    while (true) {
        $row = yield;              // pauses here until send($row)
        if ($row === null) {
            break;
        }
        $buffer[] = $row;
        if (count($buffer) >= $size) {
            $flush($buffer);
            $written += count($buffer);
            $buffer = [];
        }
    }

    if ($buffer !== []) {
        $flush($buffer);
        $written += count($buffer);
    }

    return $written;
}

$writer = batchWriter(fn (array $rows) => printf("flush %d\n", count($rows)), 2);
$writer->current();               // run to the first yield
foreach ([['id' => 1], ['id' => 2], ['id' => 3]] as $row) {
    $writer->send($row);
}
$writer->send(null);              // flush 2, flush 1
echo $writer->getReturn(), "\n";  // 3

Lần send() đầu tiên tự chạy generator tới yield đầu rồi mới giao giá trị, nhưng gọi current() trước cho ý đồ rõ ràng hơn. Đa số code ứng dụng không cần send(); xài nó khi một state machine phía tiêu thụ (gom batch, đếm rate, parser nhận từng khúc dữ liệu) đọc dễ hơn khi viết thành một hàm thay vì một class nhiều field.

Laravel: LazyCollection

LazyCollection bọc một generator và cho mình xài API của Collection (filter, map, chunk, take, each, reduce) mà vẫn lazy: mỗi thao tác kéo từng item đi qua thay vì tạo mảng trung gian. Đây là job import viết lại thành một Artisan command:

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\LazyCollection;

final class ImportOrders extends Command
{
    protected $signature = 'orders:import {path}';

    protected $description = 'Stream a large CSV into the orders table';

    public function handle(): int
    {
        $path = $this->argument('path');

        $imported = self::rows($path)
            ->filter(fn (array $row) => $row['status'] !== 'cancelled')
            ->map(fn (array $row) => [
                'external_id' => $row['id'],
                'email' => strtolower(trim($row['email'])),
                'total_cents' => (int) round(((float) $row['total']) * 100),
                'created_at' => $row['created_at'],
            ])
            ->chunk(1000)
            ->reduce(function (int $count, LazyCollection $chunk) {
                DB::table('orders')->upsert($chunk->values()->all(), ['external_id']);

                return $count + $chunk->count();
            }, 0);

        $this->info("Imported {$imported} rows, peak "
            . round(memory_get_peak_usage(true) / 1048576) . ' MB');

        return self::SUCCESS;
    }

    /** @return LazyCollection<int, array<string, string>> */
    public static function rows(string $path): LazyCollection
    {
        return LazyCollection::make(function () use ($path) {
            $handle = fopen($path, 'rb');
            try {
                $header = fgetcsv($handle, null, ',', '"', '');
                while (($fields = fgetcsv($handle, null, ',', '"', '')) !== false) {
                    if ($fields !== [null]) {
                        yield array_combine($header, $fields);
                    }
                }
            } finally {
                fclose($handle);
            }
        });
    }
}

chunk(1000) sinh ra từng khúc LazyCollection, nên mình gom thao tác ghi database thành batch (một upsert cho 1000 dòng thay vì 1000 query) mà vẫn chỉ giữ tối đa 1000 dòng trong memory. upsert theo external_id giúp chạy lại an toàn nếu job chết giữa chừng.

Eloquent cũng có mấy cách đọc lazy, và chúng không thay nhau được:

use App\Models\Order;

// One query, models hydrated one at a time.
foreach (Order::query()->where('status', 'paid')->cursor() as $order) {
    // ...
}

// Many small queries keyed by id: stable memory, safe while updating rows.
Order::query()->where('status', 'paid')->lazyById(1000)->each(function (Order $order) {
    $order->update(['exported_at' => now()]);
});

cursor() chạy một query và hydrate từng model một, nhưng trên MySQL driver PDO mặc định buffer toàn bộ result set, nên kết quả cực lớn vẫn ngốn memory. lazy() và lazyById() phân trang qua bảng bằng nhiều query nhỏ; lazyById() là lựa chọn an toàn khi mình update chính các dòng đang duyệt, vì nó phân trang theo primary key chứ không theo offset.

Kiểm thử

Có hai thứ đáng test: stream thật sự lazy, và memory bị chặn trên với input lớn. Tạo file 200 nghìn dòng trong test mất khoảng một giây và bắt được kiểu regression khi ai đó thêm ->all() hay ->collect() vô giữa pipeline, ép cả file vô memory.

// tests/Feature/ImportOrdersTest.php
use App\Console\Commands\ImportOrders;
use Illuminate\Support\Facades\DB;

function writeCsv(int $rows): string
{
    $path = tempnam(sys_get_temp_dir(), 'orders');
    $h = fopen($path, 'wb');
    fwrite($h, "id,email,total,status,created_at\n");
    for ($i = 1; $i <= $rows; $i++) {
        $status = $i % 10 === 0 ? 'cancelled' : 'paid';
        fwrite($h, "{$i}, User{$i}@Example.com ,19.99,{$status},2026-11-01 10:00:00\n");
    }
    fclose($h);

    return $path;
}

it('streams rows lazily', function () {
    $path = writeCsv(5);
    $rows = ImportOrders::rows($path);

    expect($rows->first())->toMatchArray(['id' => '1', 'status' => 'paid'])
        ->and($rows->take(3)->count())->toBe(3);
});

it('imports a big file in bounded memory', function () {
    DB::table('orders')->delete();
    $path = writeCsv(200_000);
    $before = memory_get_usage();

    $this->artisan('orders:import', ['path' => $path])->assertSuccessful();

    expect(DB::table('orders')->count())->toBe(180_000)
        ->and(DB::table('orders')->where('external_id', '1')->value('email'))->toBe('[email protected]')
        ->and(memory_get_peak_usage() - $before)->toBeLessThan(64 * 1024 * 1024);
});

Ngưỡng memory cố tình để rộng. Với unit test cho các generator function thuần, assert trên iterator_to_array($gen, false) với input nhỏ, và đưa vô một generator vô hạn để chứng minh nó chỉ kéo đúng số item cần.

Những lỗi hay gặp

• Duyệt hai lần. Generator chỉ xài được một lần. foreach lần hai sẽ throw Exception: Cannot traverse an already closed generator. LazyCollection::make(fn () => ...) nhận closure chính là để tạo generator mới cho mỗi lần duyệt, cũng có nghĩa là mỗi lần duyệt sẽ đọc lại file.
• Gọi count() hay ->all() trên stream. Thứ gì cần cả chuỗi đều sẽ gom hết vô memory. LazyCollection::count() duyệt toàn bộ; ->all() tạo nguyên mảng.
• iterator_to_array() với key trùng. Giá trị bị ghi đè âm thầm. Xài preserve_keys: false.
• Dọn dẹp sau vòng lặp thay vì trong finally. break sớm là bỏ qua luôn, rò rỉ file handle hoặc cursor database.
• Tưởng cursor() trên MySQL là lazy hoàn toàn. Query thì lazy, result buffer thì không. Bảng khổng lồ thì xài lazyById().
• Giữ tham chiếu. Đẩy từng dòng vô $this->seen[] để lọc trùng là dựng lại y nguyên vấn đề memory. Giữ một tập id hoặc hash, hoặc lọc trùng dưới database.

Khi nào nên và không nên xài generator

Nên xài khi:

• input không giới hạn hoặc rất lớn: file, export, phân trang API, bảng database;
• muốn có pipeline (đọc, lọc, biến đổi, gom batch, ghi) mà không cần mảng trung gian;
• cần dừng sớm: take(10) trên stream lazy chỉ đọc đúng phần cần;
• muốn tách "lấy dữ liệu ra sao" khỏi "làm gì với từng item" mà không cần callback.

Không nên khi:

• dữ liệu nhỏ mà cần truy cập ngẫu nhiên, sort, hay duyệt nhiều lần; mảng thường đơn giản và nhanh hơn;
• cần biết tổng số ngay từ đầu, ví dụ cho progress bar, trừ khi lấy được nó rẻ ở chỗ khác;
• mỗi item tốn CPU và mình muốn chạy song song; generator chạy tuần tự, nên chia mỗi chunk thành một queued job;
• debug chuỗi lazy làm team rối hơn là lợi ích memory mang lại.

Memory nên phụ thuộc vô kích thước batch mình chọn, chứ không phụ thuộc vô kích thước file đầu vào.