Node.js Streams — đọc file 10GB mà không nổ RAM

Có lần mình nhận con job export dữ liệu từ S3. File CSV 12GB. Đoạn code đầu tiên mình viết cũng rất tự nhiên:

const data = fs.readFileSync('/tmp/export.csv'); // 💥

Máy chạy 2GB RAM. Container bị kill sau khoảng 40 giây. Từ đó mình mới chịu ngồi đọc lại Streams trong Node.js cho tử tế. Bài này là mấy thứ mình ước mình hiểu sớm hơn.

Đọc cả file vào RAM thì luôn nổ, chỉ là sớm hay muộn

Điểm hay của Stream không nằm ở chỗ "nó nhanh hơn". Nhiều khi nó còn chậm hơn một chút. Cái nó cho bạn là RAM phẳng lì dù file to cỡ nào.

Hình dung dễ nhất: thay vì bưng nguyên cái lu nước, bạn dùng ống nước. Dữ liệu chảy từng mẩu nhỏ (mặc định 64KB với file), bạn xử lý xong mẩu nào thì mẩu đó được giải phóng. Đọc file 12GB, RAM dùng thêm tầm vài chục MB chứ không phải 12GB.

Số liệu đo thật ở job của mình:

• readFileSync: RSS đỉnh ~12.3GB (chết)
• Stream + for await: RSS đỉnh ~95MB
• Thời gian: chậm hơn khoảng 8-12% do phải chờ I/O liên tục

Đổi 10% tốc độ lấy việc không bị OOMKilled là quá hời.

Backpressure — phần bị bỏ qua nhiều nhất

Đây là khái niệm sinh ra để giải câu hỏi: nếu bên đọc chậm mà bên ghi nhanh thì sao?

Ví dụ dễ hiểu: bạn đọc file và ghi vào database. Mỗi dòng insert mất 5ms. Nếu Node cứ đọc hết file rồi nhét đống dòng vào RAM chờ ghi, bạn quay lại đúng vấn đề cũ, chỉ khác là lần này tràn ở tầng application.

Backpressure là cơ chế stream tự tạm dừng đọc khi bên nhận chưa xử lý kịp. Tin vui: khi dùng for await, bạn không cần làm gì cả. Vòng lặp chỉ lấy chunk tiếp theo sau khi thân vòng lặp chạy xong.

async function countRows(path) {
  const stream = fs.createReadStream(path, {
    encoding: 'utf8',
    highWaterMark: 64 * 1024, // 64KB mỗi chunk
  });

  let rows = 0;
  let carry = '';

  for await (const chunk of stream) {
    const lines = (carry + chunk).split('\n');
    carry = lines.pop(); // dòng bị cắt ngang, để dành chunk sau
    rows += lines.length;

    // Nếu chỗ này là `await insertVaoDb(line)` mất 5ms,
    // stream cũng tự pause luôn, không đọc thêm. Đó là backpressure.
  }

  return rows + (carry ? 1 : 0);
}

Ngược lại, nếu bạn dùng stream.on('data', cb) rồi bên trong cb gọi hàm async mà không ai chờ, backpressure mất tác dụng hoàn toàn. Node sẽ đọc hết file vào bộ đệm càng nhanh càng tốt và bạn lại nổ RAM. Đây là lý do mình khuyên bỏ on('data') cho code mới.

pipeline() thay vì pipe()

pipe() trông gọn, nhưng nó có một nhược điểm chết người: khi một stream ở giữa lỗi, các stream còn lại không tự đóng. File descriptor rò rỉ, process treo, và bạn có một lỗi rất khó lần.

const { pipeline } = require('node:stream/promises');
const { Transform } = require('node:stream');
const fs = require('node:fs');

const toUpper = new Transform({
  transform(chunk, _enc, cb) {
    cb(null, chunk.toString().toUpperCase());
  },
});

// ❌ lỗi ở giữa là rò rỉ fd, phải tự tay bắt 'error' từng stream
fs.createReadStream('in.csv').pipe(toUpper).pipe(fs.createWriteStream('out.csv'));

// ✅ lỗi ở đâu cũng lan ra ngoài, mọi stream được destroy sạch
await pipeline(
  fs.createReadStream('in.csv'),
  toUpper,
  fs.createWriteStream('out.csv'),
);

Từ Node 16 trở đi, stream/promises đã ổn định, dùng nó cho mọi thứ.

Mấy cái bẫy mình từng dính

Chunk là Buffer, không phải string. Cắt chuỗi nhiều byte ở giữa là hỏng ký tự tiếng Việt. Truyền encoding: 'utf8' cho createReadStream hoặc dùng StringDecoder cho an toàn.

objectMode: true làm highWaterMark đếm theo số object, không phải byte. Một object to 5MB vẫn tính là 1. Đừng trộn object mode với byte stream rồi thắc mắc tại sao RAM vẫn cao.

CPU-bound thì Stream không cứu được. Nếu transform của bạn là nén gzip hay parse JSON phức tạp, nó chặn event loop và cả app đứng. Lúc đó đẩy sang worker_threads, stream chỉ lo phần I/O.

Đo RAM bằng process.memoryUsage().rss, đừng tin top. Heap của Node có thể giữ lại bộ nhớ đã cấp phát, nhìn top thấy cao nhưng thực tế không tăng theo kích thước file.

Chốt

Quy tắc mình tự đặt ra sau vụ 12GB đó:

• Dữ liệu lớn hơn vài chục MB và mình không cần cả file một lúc → Stream.
• Cần backpressure mà không muốn nghĩ nhiều → for await.
• Nhiều tầng xử lý → pipeline(), không phải pipe().
• Transform nặng CPU → worker_threads.
• Có ghi DB hoặc gọi API trong vòng lặp → kiểm tra con số RSS ngay, đừng chờ prod nổ.

Streams không phải thứ hào nhoáng, nhưng nó là ranh giới giữa một con job chạy được và một cái pod bị kill lúc 3 giờ sáng.

📋 Phụ lục thuật ngữ