Buffer & Stream

Chuỗi JavaScript che mất một sự thật: đĩa cứng và đường mạng không biết "ký tự" là gì - chúng chỉ chuyên chở byte, như bài Hệ nhị phân (khoá Máy tính hoạt động thế nào?) đã chỉ ra. Node sinh ra để đụng vào tệp và mạng, nên cần một kiểu dữ liệu đại diện cho dãy byte trần trụi: Buffer. Nhìn nó mổ xẻ một chữ tiếng Việt:

const b = Buffer.from("Mèo", "utf8");

console.log(b);                       // nhìn thẳng vào từng byte
console.log(b.length, "Mèo".length);  // số byte ≠ số ký tự!
console.log(b.toString("hex"));       // mỗi byte = 2 chữ số hex
console.log(b.toString("base64"));    // encoding khác, cùng dãy byte
console.log(Buffer.from("4dc3a86f", "hex").toString("utf8"));   // khứ hồi
console.log(b instanceof Uint8Array);             // Buffer là Uint8Array mở rộng
console.log(b.subarray(0, 2).toString("utf8"));   // cắt GIỮA chữ "è"!

<Buffer 4d c3 a8 6f>
4 3
4dc3a86f
TcOobw==
Mèo
true
M�

Bốn byte cho ba ký tự: 4d là "M", 6f là "o" - còn chữ "è" chiếm hai byte c3 a8, vì UTF-8 mã hoá chữ ngoài bảng ASCII bằng nhiều byte. Vậy nên cắt theo vị trí byte có thể chém đôi một ký tự: dòng cuối giữ 4d c3, "è" mất nửa sau, thành ký tự lỗi �.

• ▸Buffer = vùng nhớ chứa byte thô; số byte (length) khác số ký tự khi có chữ ngoài ASCII.
• ▸encoding (utf8, hex, base64…) là luật quy đổi chữ ↔ byte - đổi qua lại khứ hồi không mất mát.
• ▸Buffer là Uint8Array mở rộng: chỗ nào nhận TypedArray đều nhận được Buffer.

Ở bài Làm việc với tệp & thư mục, readFile nạp TRỌN tệp vào RAM rồi mới đưa cho mèo con. Tiện - cho đến ngày gặp tệp log 10 GB trên máy 8 GB RAM: chưa kịp xử lý gì đã hết bộ nhớ. Stream giải bài toán bằng cách cho dữ liệu CHẢY qua chương trình từng miếng nhỏ - mỗi miếng là một chunk; RAM chỉ cần đủ chỗ cho miếng đang xử lý. Dựng một tệp thí nghiệm cố định rồi đếm thử:

import { writeFile } from "node:fs/promises";
import { createReadStream } from "node:fs";

// dựng tệp thí nghiệm cố định: 100.000 dòng văn bản
let noiDung = "";
for (let i = 1; i <= 100_000; i++) noiDung += `meo con ghi dong so ${i}\n`;
await writeFile("nhat-ky.log", noiDung);

let soChunk = 0, soByte = 0, soDong = 0;
for await (const chunk of createReadStream("nhat-ky.log")) {
  soChunk += 1;
  soByte += chunk.length;
  for (const byte of chunk) if (byte === 10) soDong += 1;   // 10 = mã byte của "\n"
}
console.log("chunk:", soChunk, "| byte:", soByte, "| dòng:", soDong);

chunk: 40 | byte: 2588895 | dòng: 100000

Để ý vòng lặp: for await...of - không có cú pháp mới nào ở đây cả. Readable stream implement đúng async iterator protocol mèo con học ở bài Generator & async iterator: mỗi vòng là một lượt await lấy chunk kế tiếp, hết tệp thì vòng lặp tự kết thúc - break giữa chừng cũng an toàn, stream tự được huỷ.

• ▸Stream = dữ liệu chảy từng chunk (64 KiB mỗi chunk với stream tệp) - tệp 10 GB hay 10 KB tốn RAM như nhau.
• ▸2.588.895 byte ÷ 64 KiB ≈ 40 chunk - khớp đúng con số chương trình đếm được.
• ▸for await...of trên stream chính là async iterator của bài trước - kiến thức cũ, sân khấu mới.

Đọc từng chunk rồi GHI từng chunk sang nơi khác - thế là thành copy. Stream chia hai phe: readable (nguồn phát: đọc tệp, HTTP request đến…) và writable (đích nhận: ghi tệp, response trả đi…). Cách cổ điển là nghe sự kiện data / end rồi tự write(), hoặc gọi nguon.pipe(dich); cách hiện đại - bài này dùng từ đây về sau - là pipeline, phiên bản Promise để await được:

import { createReadStream, createWriteStream } from "node:fs";
import { pipeline } from "node:stream/promises";
import { stat } from "node:fs/promises";

await pipeline(
  createReadStream("nhat-ky.log"),      // vòi nước: readable
  createWriteStream("ban-sao.log")      // cống thoát: writable
);

console.log("gốc    :", (await stat("nhat-ky.log")).size, "byte");
console.log("bản sao:", (await stat("ban-sao.log")).size, "byte");

gốc    : 2588895 byte
bản sao: 2588895 byte

Mỗi stream mang theo một vùng đệm nội bộ (internal buffer) - chỗ chứa tạm những chunk đã đọc mà chưa ai lấy, hoặc đã write() mà chưa kịp xuống đĩa. Định mức của vùng đệm ấy là highWaterMark (nghĩa đen: vạch nước cao - mức đánh dấu "đầy"), mặc định 64 KiB cho stream tệp. Nó KHÔNG phải giới hạn cứng - vậy để làm gì? Bước kế trả lời.

• ▸readable = nguồn phát chunk, writable = đích nhận; copy tệp = nối hai đầu lại.
• ▸pipeline() trả Promise: await được, lỗi ở mắt xích nào cũng nổi lên đúng một chỗ.
• ▸highWaterMark = định mức vùng đệm nội bộ - ngưỡng báo "đầy", không phải trần cứng.

Câu hỏi trung tâm của mọi hệ thống stream: đọc từ SSD được hàng trăm MB mỗi giây, nhưng đầu ghi - qua mạng chẳng hạn - chỉ nuốt nổi vài MB mỗi giây. Producer (bên tạo dữ liệu) nhanh hơn consumer (bên tiêu thụ), thì những chunk chưa kịp tiêu dồn lại trong vùng đệm - không ai hãm producer thì vùng đệm phình tới khi ăn hết RAM: công cụ sinh ra để tiết kiệm bộ nhớ lại làm tràn bộ nhớ. Tín hiệu hãm có sẵn trong API: write(chunk) trả về boolean. false nghĩa là "vùng đệm đã chạm highWaterMark - khoan ghi tiếp"; producer tử tế sẽ dừng, đợi sự kiện drain (vùng đệm đã rút) rồi mới ghi tiếp. Cơ chế dòng dưới báo ngược lên dòng trên này gọi là backpressure (áp lực dội ngược). Xem điều gì xảy ra khi ta LỜ tín hiệu:

import { createWriteStream } from "node:fs";

const ws = createWriteStream("o-at.bin", { highWaterMark: 16 * 1024 });
const mieng = Buffer.alloc(1024, 97);   // mỗi miếng 1 KiB, toàn byte 97 ("a")

let lanFalse = 0;
for (let i = 0; i < 1000; i++) {
  if (!ws.write(mieng)) lanFalse += 1;   // false = "khoan rót!" - nhưng ta lờ đi
}
ws.end();

console.log("write() trả false:", lanFalse, "/ 1000 lần");

write() trả false: 985 / 1000 lần

Vùng đệm 16 KiB nhận đủ 15 miếng đầu; từ miếng 16 trở đi, cả 985 lần write() đều kêu "khoan!". Chương trình vẫn chạy xong, tệp vẫn đủ 1.024.000 byte - vì false chỉ là lời khuyên - nhưng gần như toàn bộ 1000 KiB đã nằm chờ trong RAM thay vì tối đa ~16 KiB. Phiên bản tử tế (ngoan.mjs) chỉ cần import thêm hàm once từ node:events và đổi vòng lặp:

let lanDoi = 0;
for (let i = 0; i < 1000; i++) {
  if (!ws.write(mieng)) {        // vùng đệm chạm vạch →
    lanDoi += 1;
    await once(ws, "drain");     // ...đứng chờ rút cạn rồi mới rót tiếp
  }
}
ws.end();

console.log("số lần phải dừng chờ drain:", lanDoi);

số lần phải dừng chờ drain: 62

62 lần dừng ấy là nhịp thở của backpressure: ghi - đầy - dừng - rút - ghi tiếp. Nghịch mô phỏng dưới đây để thấy nhịp đó bằng mắt: kéo tốc độ ghi lên, rồi thử tắt backpressure xem vùng đệm đi về đâu.

• ▸write() trả false = vùng đệm đã chạm highWaterMark; là tín hiệu khuyên dừng, không phải lỗi.
• ▸Producer dừng khi gặp false, nghe drain một lần rồi ghi tiếp - đỉnh RAM bị ghìm quanh vạch thay vì phình theo dữ liệu.
• ▸pipe() và pipeline() tự làm trọn vòng này - lý do thật sự để nối stream bằng chúng thay vì tự write trong vòng lặp.

Mảnh ghép cuối: transform - stream hai mặt, vừa writable (nhận chunk vào) vừa readable (nhả chunk ra), đứng giữa dây chuyền để BIẾN ĐỔI dữ liệu chảy qua: nén, mã hoá, đổi định dạng… Node có sẵn một transform kinh điển: createGzip() của node:zlib. Ghép ba mắt xích là thành máy nén tệp không tốn RAM:

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

await pipeline(
  createReadStream("nhat-ky.log"),   // readable: nguồn
  createGzip(),                      // transform: nén từng chunk đi qua
  createWriteStream("nhat-ky.log.gz")// writable: đích
);

console.log("trước khi nén:", (await stat("nhat-ky.log")).size, "byte");
console.log("sau khi nén  :", (await stat("nhat-ky.log.gz")).size, "byte");

trước khi nén: 2588895 byte
sau khi nén  : 262688 byte

Vì sao bài này một mực dùng pipeline() thay vì chuỗi .pipe().pipe()? Vì lỗi. Khi một mắt xích gãy giữa chừng (tệp biến mất, đĩa đầy…), chuỗi .pipe() không tự huỷ các stream còn lại - file descriptor treo, vùng đệm kẹt trong RAM; còn pipeline() huỷ TẤT CẢ mắt xích, dọn tài nguyên, đưa lỗi về đúng một chỗ cho try...catch. Và stream không phải tính năng ngách của module fs: process.stdout là writable; nén, băm, parse từng dòng là transform - hiểu một khái niệm, dùng được khắp Node.

• ▸transform = writable + readable: nhận chunk, biến đổi, nhả ra - createGzip là ví dụ chuẩn mực.
• ▸Tệp thí nghiệm 2.588.895 byte nén còn 262.688 byte (~10%) - văn bản lặp nhiều nén rất sâu.
• ▸pipeline() khi lỗi: huỷ mọi mắt xích + dọn tài nguyên + reject một chỗ - hơn hẳn chuỗi .pipe().