Stream & backpressure

Đọc cả một file 2 GB vào bộ nhớ rồi mới gửi đi thì process cần 2 GB RAM, và người dùng phải chờ đọc xong mới nhận được byte đầu tiên. Stream xử lý dữ liệu theo từng khúc nhỏ khi nó đến, nên bộ nhớ gần như không đổi dù dữ liệu lớn cỡ nào. Trang này giải thích bốn loại stream, backpressure (cơ chế để nguồn nhanh không làm ngập đích chậm), vì sao nên dùng pipeline() thay cho pipe(), và các ứng dụng hay gặp ở backend.

Bốn loại stream

Loại	Là gì	Ví dụ
Readable	Nguồn dữ liệu, đọc ra từng khúc	`fs.createReadStream`, request HTTP phía server (`req`), response phía client, kết quả query dạng stream
Writable	Đích nhận dữ liệu	`fs.createWriteStream`, response HTTP phía server (`res`), `process.stdout`
Duplex	Vừa đọc vừa ghi, hai chiều độc lập	Socket TCP
Transform	Duplex mà đầu ra được tính từ đầu vào	`zlib.createGzip()`, chuyển dòng database thành CSV, mã hoá

Dữ liệu đi qua stream theo từng chunk. Với stream byte, mỗi chunk là một Buffer. Với stream ở object mode, mỗi chunk là một giá trị JavaScript bất kỳ, ví dụ một dòng kết quả query.

Backpressure

Mỗi Writable có một đệm bên trong, với ngưỡng gọi là highWaterMark: mặc định 64 KiB với stream byte (16 KiB trên Windows) và 16 object với object mode. write(chunk) cho chunk vào đệm rồi trả về:

true nếu đệm còn dưới ngưỡng: ghi tiếp được.
false nếu đệm đã vượt ngưỡng: nên dừng lại, chờ sự kiện 'drain' (đệm đã xả hết) rồi mới ghi tiếp.

write() trả false không có nghĩa là chunk bị bỏ: chunk vẫn được nhận vào đệm. Vì vậy code bỏ qua giá trị trả về vẫn "chạy đúng", chỉ là nếu nguồn nhanh hơn đích, đệm cứ phình ra. Đó là cái bẫy của backpressure. Trên lab, ghi 10.000 dòng vào một đích chậm mà bỏ qua false thì đệm lên tới gần 10 MB, tức toàn bộ dữ liệu. Chờ 'drain' thì đệm không vượt quá khoảng 64 KB, và tổng thời gian gần như bằng nhau, vì tốc độ cuối cùng do đích chậm quyết định. Với file vài GB hay một luồng không dừng, bỏ qua backpressure là cách nhanh nhất làm process hết bộ nhớ.

Nguồn đọc 200 MB/s

Đích ghi 20 MB/s

highWaterMark 64 KB

2 MBdữ liệu nằm chờ trong bộ nhớ, lúc nhiều nhất

15,0 schép xong 300 MB

Thời gian chép bằng nhau ở cả hai cách, vì cuối cùng tốc độ do đích chậm quyết định. Khác nhau là bộ nhớ: tôn trọng backpressure thì đệm không bao giờ vượt xa highWaterMark, bất kể file lớn cỡ nào.

Mô hình liên tục theo từng 10 ms, bỏ qua chi phí của chính stream.

Chỗ backpressure hay bị phá ở backend:

Gửi response lớn cho client mạng chậm. Đọc từ database hay file nhanh hơn nhiều so với tốc độ client tải về.
Chép dữ liệu giữa hai hệ thống (database sang S3, Kafka sang database) bằng vòng lặp for gọi write() liên tục.
Transform tự viết mà push dữ liệu ra không theo nhịp đích đọc.

Không cần tự xử lý 'drain' nếu nối stream bằng pipeline(), hoặc dùng async iterator (for await) ở phía đọc: cả hai tự tôn trọng backpressure.

`pipeline()` thay cho `pipe()`

a.pipe(b) nối stream và xử lý backpressure, nhưng không xử lý lỗi. Tài liệu Node nói rõ: khi Readable báo lỗi giữa chừng, Writable đích không tự đóng. Trên lab, nguồn lỗi giữa chừng với pipe() thì stream đích vẫn chưa bị huỷ và file descriptor vẫn mở, tức rò tài nguyên. Phải tự gắn handler 'error' cho từng stream trong chuỗi và tự huỷ các stream còn lại.

pipeline() làm hết việc đó: nối các stream, xử lý backpressure, và khi một stream lỗi thì huỷ tất cả rồi trả lỗi về một chỗ.

import { pipeline } from 'node:stream/promises'
import { createReadStream, createWriteStream } from 'node:fs'
import { createGzip } from 'node:zlib'

await pipeline(
  createReadStream('orders.csv'),
  createGzip(),
  createWriteStream('orders.csv.gz')
)  // lỗi ở bất kỳ bước nào: mọi stream bị huỷ, await ném lỗi

pipeline nhận cả async generator làm bước biến đổi, nên phần lớn Transform không cần viết bằng class:

await pipeline(
  createReadStream('orders.csv'),
  async function* (source) {
    for await (const chunk of source) yield chunk.toString().toUpperCase()
  },
  createWriteStream('out.csv')
)

Async iterator

Readable là một async iterable, nên đọc được bằng for await…of. Vòng lặp chỉ lấy chunk tiếp theo khi thân vòng lặp xong, nên backpressure tự có: nếu thân vòng lặp await một lệnh ghi chậm, stream nguồn tự dừng đọc. Đọc file theo từng dòng thì dùng readline:

import { createInterface } from 'node:readline'

for await (const line of createInterface({ input: createReadStream('big.csv'), crlfDelay: Infinity })) {
  await handle(line)
}

Trên lab, đếm 1,5 triệu dòng của một file 91 MB bằng cách này chỉ dùng khoảng 35 MB RAM thêm so với process trống, và bộ nhớ không tăng theo kích thước file.

Readable.from(iterable) làm chiều ngược lại: biến một mảng, generator hay async generator thành Readable để đưa vào pipeline.

Ứng dụng ở backend

Tải file về (download). Stream thẳng từ file hay object storage sang res, thay vì đọc hết vào Buffer:

app.get('/files/:id', async (req, res) => {
  res.setHeader('Content-Type', 'application/pdf')
  await pipeline(createReadStream(pathOf(req.params.id)), res)
})

Trên lab, chép một file 91 MB bằng readFile rồi writeFile làm RSS tăng khoảng 94 MB, đúng bằng kích thước file. Bằng pipeline thì chỉ tăng khoảng 22 MB, và con số này không đổi khi file lớn hơn. Với 20 người tải cùng lúc một file 1 GB, cách đọc hết cần 20 GB RAM.

Nhận file tải lên (upload). Stream request vào đích (đĩa, object storage) thay vì gom vào bộ nhớ. Thư viện xử lý multipart như busboy làm theo cách stream. Giới hạn kích thước ngay trong lúc stream, để không phải nhận hết rồi mới biết là quá lớn.

Xuất CSV từ database. Đọc kết quả query theo dạng stream (mysql2 có .stream(), pg có pg-query-stream dùng cursor), biến từng dòng thành một dòng CSV, rồi pipeline sang res. Dòng đầu tiên tới người dùng gần như ngay lập tức, và bộ nhớ không phụ thuộc số dòng (Xử lý dữ liệu lớn theo lô).

Nén. zlib.createGzip() là một Transform, đặt giữa nguồn và đích trong pipeline.

Khi client đóng connection giữa chừng (tắt tab, mất mạng), res bị huỷ, và pipeline huỷ luôn stream đọc file hay query, nên server không tiếp tục đọc dữ liệu cho một client đã đi.

Ghi nhớ

Stream xử lý dữ liệu theo từng chunk: bộ nhớ không tăng theo kích thước dữ liệu, và byte đầu tiên tới sớm.
write() trả false khi đệm vượt highWaterMark (mặc định 64 KiB): dừng lại chờ 'drain'. Bỏ qua nó thì đệm phình tới khi hết bộ nhớ.
Dùng pipeline() thay pipe(): pipe() không đóng đích khi nguồn lỗi, rò file descriptor và connection.
for await trên Readable tự có backpressure; readline để đọc từng dòng; async generator làm bước biến đổi trong pipeline.
Download, upload, xuất CSV từ database đều nên stream thẳng, không gom vào Buffer.

Nguồn tham khảo

Đã đối chiếu (đọc nguồn ngày 2026-10-26):

Nguồn	Dùng cho phần
Node.js docs, Stream	`highWaterMark` mặc định: 16 với object mode, 64 KiB với stream byte (16 KiB trên Windows); `pipe()` không tự đóng đích khi nguồn lỗi
Node.js, Backpressuring in Streams	`write()` trả `false` khi đệm đầy; `'drain'` khi đệm đã xả hết

Kiểm bằng lab: bộ nhớ khi chép file bằng readFile so với pipeline, đếm dòng bằng readline, đệm khi bỏ qua và khi tôn trọng backpressure, pipe() so với pipeline() khi nguồn lỗi. Từ kiến thức chung, chưa đối chiếu lần này: pipeline nhận async generator; busboy, pg-query-stream; pipeline huỷ stream nguồn khi res bị huỷ.

Số đo chi tiết

Kết quả chạy lab/stream/memory.mjs (MacBook M2, Node 24.14, hai lượt)

File CSV 91 MB, 1.500.000 dòng. RSS của một process Node trống: 46 MB.

Cách	RSS cao nhất	Thời gian
`readFile` + `writeFile`	139–140 MB	62–83 ms
`pipeline(createReadStream, createWriteStream)`	68 MB	74–109 ms
`readline` + `for await`, đếm dòng	80–81 MB	224–226 ms

Ghi 10.000 dòng 1 KB vào một Writable mất 1 ms mỗi chunk:

Cách	Đệm lớn nhất	RSS cao nhất	Thời gian
Bỏ qua `write() === false`	9,8–10 MB (gần như toàn bộ dữ liệu)	80–87 MB	11,7 s
Chờ `'drain'`	65 KB	59–66 MB	11,5–11,6 s

Nguồn lỗi giữa chừng: với src.pipe(dst), lỗi bắt được ở src nhưng dst.destroyed = false và file descriptor còn mở; với await pipeline(src, dst), lỗi ném ra ở await và dst.destroyed = true.

Code đã dùng để chạy thử

Chạy bằng npm run lab -- stream, chỉ cần Node.js, khoảng 25 giây. Script tạo file mẫu khoảng 90 MB trong thư mục tạm và luôn xoá khi xong. Xem Lab.

lab/stream/memory.mjs

// Trang nodejs/stream. Mỗi phép đo chạy trong một process con để đo RSS lúc cao nhất cho sạch.
//   copy-buffer / copy-stream: chép file CSV khoảng 90 MB bằng readFile+writeFile so với pipeline(createReadStream, createWriteStream)
//   ignore / respect: ghi 10.000 dòng vào một Writable chậm, bỏ qua write() === false so với chờ 'drain'
//   lines: đếm dòng của file CSV đó bằng readline + for await
//   pipe-error / pipeline-error: nguồn lỗi giữa chừng, đích có bị đóng không
import { spawnSync } from 'node:child_process'
import { createReadStream, createWriteStream } from 'node:fs'
import { readFile, writeFile, mkdtemp, rm, stat } from 'node:fs/promises'
import { pipeline } from 'node:stream/promises'
import { Readable, Writable } from 'node:stream'
import { createInterface } from 'node:readline'
import { tmpdir } from 'node:os'
import { join } from 'node:path'

const mode = process.argv[2]
const mb = (b) => Math.round(b / 1e6) + ' MB'
let peak = 0
const iv = setInterval(() => (peak = Math.max(peak, process.memoryUsage().rss)), 5)
const done = (extra = '') => { clearInterval(iv); peak = Math.max(peak, process.memoryUsage().rss); console.log(JSON.stringify({ peak, extra })) }

if (mode === 'copy-buffer') {
  const t = performance.now()
  await writeFile(process.argv[4], await readFile(process.argv[3]))
  done(`${Math.round(performance.now() - t)} ms`)
} else if (mode === 'copy-stream') {
  const t = performance.now()
  await pipeline(createReadStream(process.argv[3]), createWriteStream(process.argv[4]))
  done(`${Math.round(performance.now() - t)} ms`)
} else if (mode === 'ignore' || mode === 'respect') {
  // Đích chậm: mỗi chunk mất 1 ms (giống ghi qua mạng hay vào database chậm); highWaterMark mặc định (64 KiB, 16 KiB trên Windows)
  const slow = new Writable({ write(chunk, enc, cb) { setTimeout(cb, 1) } })
  let maxBuffered = 0, falses = 0
  const line = 'x'.repeat(1000) + '\n'
  const t = performance.now()
  for (let i = 0; i < 10_000; i++) {
    const ok = slow.write(line)
    maxBuffered = Math.max(maxBuffered, slow.writableLength)
    if (!ok) { falses++; if (mode === 'respect') await new Promise((r) => slow.once('drain', r)) }
    if (i % 1000 === 0) await new Promise((r) => setImmediate(r)) // nhường event loop như một producer thật
  }
  await new Promise((r) => slow.end(r))
  done(`đệm lớn nhất ${Math.round(maxBuffered / 1024).toLocaleString('vi-VN')} KB, write() trả false ${falses} lần, mất ${Math.round(performance.now() - t)} ms`)
} else if (mode === 'lines') {
  const t = performance.now()
  let n = 0
  for await (const l of createInterface({ input: createReadStream(process.argv[3]), crlfDelay: Infinity })) if (l) n++
  done(`${n.toLocaleString('vi-VN')} dòng, ${Math.round(performance.now() - t)} ms`)
} else if (mode === 'pipe-error' || mode === 'pipeline-error') {
  let i = 0
  const src = new Readable({ read() { i++ === 50 ? this.destroy(new Error('nguồn lỗi')) : this.push('x'.repeat(1024)) } })
  const dst = createWriteStream(process.argv[3])
  let caught = 'không có'
  if (mode === 'pipe-error') { src.on('error', (e) => (caught = e.message)); src.pipe(dst) }
  else { try { await pipeline(src, dst) } catch (e) { caught = e.message } }
  await new Promise((r) => setTimeout(r, 100))
  done(`lỗi bắt được: ${caught}; đích destroyed = ${dst.destroyed}, file descriptor còn mở = ${dst.fd !== null && !dst.destroyed}`)
} else if (mode === 'idle') {
  done()
} else if (mode !== undefined) {
  clearInterval(iv)
  throw new Error(`chế độ không hợp lệ: ${mode}`)
} else if (process.env.STREAM_LAB_CHILD) {
  // Chặn đệ quy: process con không bao giờ được chạy nhánh điều phối.
  // (Ngày 2026-10-26 thiếu nhánh 'idle' ở trên làm mỗi process con lại điều phối, tạo 231 file 293 MB.)
  clearInterval(iv)
  throw new Error('process con không được chạy nhánh điều phối')
} else {
  // Điều phối: tạo file mẫu, chạy từng chế độ trong process con
  clearInterval(iv)
  const dir = await mkdtemp(join(tmpdir(), 'stream-'))
  try {
  const big = join(dir, 'big.csv')
  const row = (i) => `${i},user${i % 50000}@example.com,${(i * 37) % 100000},paid,2025-06-01T10:00:00Z\n`
  await pipeline(Readable.from((function* () { for (let i = 0; i < 1_500_000; i++) yield row(i) })()), createWriteStream(big))
  const size = (await stat(big)).size
  const run = (args) => {
    const r = spawnSync(process.execPath, [new URL(import.meta.url).pathname, ...args], { encoding: 'utf8', env: { ...process.env, STREAM_LAB_CHILD: '1' } })
    if (r.status !== 0) throw new Error(`process con ${args[0]} lỗi: ${r.stderr}`)
    return JSON.parse(r.stdout.trim().split('\n').pop())
  }
  const base = run(['idle'])
  console.log(`\n=== File CSV ${mb(size)}, 1.500.000 dòng. RSS của một process Node trống: ${mb(base.peak)} ===`)
  for (const [label, args] of [['readFile + writeFile', ['copy-buffer', big, join(dir, 'a')]], ['pipeline(createReadStream, createWriteStream)', ['copy-stream', big, join(dir, 'b')]],
    ['readline + for await, đếm dòng', ['lines', big]]]) {
    const r = run(args)
    console.log(`${label.padEnd(48)} RSS cao nhất ${mb(r.peak).padStart(7)}   ${r.extra}`)
  }
  console.log('\n=== Ghi 10.000 dòng 1 KB vào một Writable chậm (1 ms mỗi chunk) ===')
  for (const [label, m] of [['Bỏ qua write() === false', 'ignore'], ["Chờ 'drain' khi write() trả false", 'respect']]) {
    const r = run([m])
    console.log(`${label.padEnd(48)} RSS cao nhất ${mb(r.peak).padStart(7)}   ${r.extra}`)
  }
  console.log('\n=== Nguồn lỗi giữa chừng ===')
  for (const [label, m] of [['src.pipe(dst)', 'pipe-error'], ['await pipeline(src, dst)', 'pipeline-error']]) {
    const r = run([m, join(dir, m)])
    console.log(`${label.padEnd(48)} ${r.extra}`)
  }
  } finally {
    await rm(dir, { recursive: true, force: true }) // luôn xoá file mẫu, kể cả khi có lỗi giữa chừng
  }
}

Stream & backpressure ​

Bốn loại stream ​

Backpressure ​

pipeline() thay cho pipe() ​

Async iterator ​

Ứng dụng ở backend ​

Nguồn tham khảo ​

Số đo chi tiết ​

Code đã dùng để chạy thử ​