Node.js stream: Readable, Writable, Duplex, Transform va oqim rejimlari

Qisqacha: Stream (oqim) — ma'lumotni bir yo'la emas, bo'laklab (chunk) qayta ishlash usuli. To'rt turi bor: Readable (o'qiladigan — fayl, so'rov tanasi), Writable (yoziladigan — fayl, process.stdout), Duplex (ikkalasi, mustaqil — TCP soket) va Transform (kirganini o'zgartirib chiqaradi — gzip). Fayl oqimi sukut bo'yicha 64 KB lik bo'laklar beradi: 1 MB fayl — 16 bo'lak. Shuning uchun 10 GB faylni ham bir necha o'n MB xotira bilan qayta ishlash mumkin.

Bu darsda

• Stream nima ekanini va qachon butun faylni o'qishdan foydaliroq ekanini tushuntirasiz.
• To'rt turni farqlaysiz va Node'dagi misollarini aytasiz.
• Bo'lak (chunk), highWaterMark va objectMode nima ekanini bilasiz.
• for await, "data" va "end" hodisalari bilan Readable'ni o'qiysiz, Writable'ga yozasiz.
• Flowing va paused rejimlari farqini va ular qanday almashishini ko'rasiz.

Oldin bilishingiz kerak: Streams API va javobni oqim bilan o'qish, EventEmitter asoslari, Buffer va binar ma'lumot, fs: uch API shakli.

1. Nega bu kerak?

fs: uch API shakli darsida faylni readFile bilan o'qidik: fayl butunligicha xotiraga tushadi, keyin ishlaymiz. 25 KB lik CBU javobi uchun bu ideal. Endi tasavvur qiling: bank sizga 5 yillik kunlik kurslar arxivini beradi — 2 GB lik CSV. readFile bilan:

• Butun 2 GB xotiraga tushishi kerak (va satrga aylantirilsa — yana ko'proq).
• Birinchi qatorni qayta ishlashdan oldin oxirgi baytgacha kutasiz.
• Kichik serverda (512 MB) dastur shunchaki yiqiladi.

Kanon buni o'lchagan (natijalar/tajriba/t35-katta.txt, 41 MB, 1 mln qatorli CSV): butun faylni o'qib split qilish — eng ko'p 194 MB xotira, oqim bilan — 63 MB. Xotira chegarasi 16 MB qilib qo'yilganda butun fayl usuli FATAL heap out of memory bilan yiqildi, oqim esa ishladi. Halol qo'shimcha: oqim tezroq emas — 796 ms va 567 ms. Uning yutug'i xotirada, tezlikda emas. Batafsil — Katta faylni stream bilan qayta ishlash darsida.

O'xshatish: hovuzni to'ldirish. Butun suvni bitta ulkan idishda olib kelish mumkin (readFile) — lekin idish hovuzdan katta bo'lishi kerak. Yoki shlang ulash mumkin (stream): suv oqib keladi, siz uni qabul qilib turasiz, idish kerak emas.

2. Bo'lak: oqimning birligi

2.1 Fayl qanday bo'laklarga bo'linadi

Brauzerda Streams API ni ko'rgansiz: fetch javobi ReadableStream, uni bo'laklab o'qish mumkin edi. Node'ning node:stream moduli — xuddi shu g'oyaning Node'dagi (va undan eskiroq) ko'rinishi. Keling, 1 MB lik faylni oqim bilan o'qib, bo'laklarni sanaymiz:

import { writeFileSync, createReadStream } from "node:fs";

writeFileSync("katta.txt", "a".repeat(1_000_000)); // 1 MB fayl
const sizes = [];
for await (const chunk of createReadStream("katta.txt")) {
  sizes.push(chunk.length);
}
console.log(sizes.length, sizes[0], sizes.at(-1)); // 16 65536 16960

Fayl 16 bo'lakda keldi: 15 tasi 65 536 bayt (64 KB), oxirgisi qolgan 16 960 bayt. Har bo'lak — Buffer. Bir vaqtda xotirada faqat bitta bo'lak — 64 KB, 1 MB emas.

for await (const chunk of stream) — Readable'ni o'qishning eng zamonaviy va eng xavfsiz yo'li (Asinxron iteratsiya). readline va net darslarida aynan shu bilan soket va qatorlarni o'qidik.

2.2 highWaterMark va objectMode

Bo'lak hajmi highWaterMark (yuqori suv belgisi) bilan belgilanadi. U oqim ichki buferining "to'ldi" chegarasi:

Kanon tajribasida (Windows): new Writable().writableHighWaterMark — 16 384. Bu sukut operatsion tizimga bog'liq: Node hujjatiga ko'ra Node 22 dan beri Windows'da 16 KB, boshqa tizimlarda 64 KB (65 536). Linux'da biz sinamadik. createReadStream("f", { highWaterMark: 4 }) desangiz — 4 baytlik bo'laklar keladi.

objectMode (obyekt rejimi) — oqimda bayt emas, istalgan JS qiymati (obyekt, son, satr) oqadi. Har qiymat — bitta bo'lak. Readable.from([...]) sukut bo'yicha obyekt rejimida: kanon tajribasi Readable.from([1]).readableObjectMode — true. valyuta ning tarix buyrug'i aynan shunday ishlaydi: kunma-kun kurs obyektlari oqadi.

Tekshirib ko'ring: 200 KB lik faylni createReadStream sukut sozlamalari bilan o'qisangiz, nechta bo'lak keladi va oxirgisi qancha?

3. To'rt tur

flowchart LR
  R["Readable<br/>fayl, stdin"] --> T["Transform<br/>gzip"]
  T --> W["Writable<br/>fayl, stdout"]
  D["Duplex<br/>TCP soket"] <--> N(("tarmoq"))

Diagrammada ma'lumot chapdan o'ngga oqadi: manbadan (Readable) o'zgartiruvchi (Transform) orqali qabul qiluvchiga (Writable). Duplex esa ikki tomonlama — uning kirishi va chiqishi bir-biridan mustaqil.

Duplex va Transform farqi: soketga yozganingiz (so'rov) va undan o'qiganingiz (javob) — boshqa-boshqa ma'lumot. Gzip'ga yozganingiz va undan o'qiganingiz esa bog'liq — siqilgan ko'rinishi. Hammasi EventEmitter — shuning uchun on("data"), on("error") ishlaydi.

Biz klasslarni tekshirdik (Node 24.21.0):

import net from "node:net";
import { Duplex, Transform, PassThrough } from "node:stream";
import { Writable } from "node:stream";

console.log(new net.Socket() instanceof Duplex); // true
console.log(new PassThrough() instanceof Transform); // true
console.log(process.stdout instanceof Writable); // true

PassThrough — "hech narsani o'zgartirmaydigan" Transform. readline darsida testlarda process.stdout o'rniga shuni berdik.

4. Readable bilan ishlash

4.1 Uch usul

Readable'dan o'qishning uch yo'li bor:

import { Readable } from "node:stream";

// 1) for await — tavsiya etiladi
for await (const dish of Readable.from(["osh", "manti"])) {
  console.log("for await:", dish);
}

// 2) "data" va "end" hodisalari
const menu = Readable.from(["lag'mon", "somsa"]);
menu.on("data", (dish) => console.log("data:", dish));
await new Promise((resolve) => menu.on("end", resolve));
console.log("end");

Konsolda:

for await: osh
for await: manti
data: lag'mon
data: somsa
end

Uchinchi yo'l — pipe/pipeline bilan to'g'ridan boshqa oqimga ulash. U keyingi ikki darsning mavzusi (Backpressure, pipeline).

"end" — Readable'da boshqa ma'lumot yo'q. "close" — oqim resurslari (fayl deskriptori) yopildi. "error" — xato; EventEmitter asoslari dagi qoida bu yerda ham: tinglovchisiz "error" jarayonni yiqitadi. for await esa xatoni o'zi throw qiladi — try/catch bilan ushlanadi. Bu for await ning asosiy ustunligi.

4.2 Matn bo'laklari: setEncoding

Fayl oqimi Buffer beradi. Matn kerak bo'lsa, encoding bering:

import { writeFileSync, createReadStream } from "node:fs";

writeFileSync("menyu.txt", "osh, manti, somsa");
const stream = createReadStream("menyu.txt", {
  encoding: "utf8",
  start: 0,
  end: 2,
});
for await (const chunk of stream) console.log(typeof chunk, chunk);

Konsolda:

string osh

Bo'lak endi satr. Muhimi — Buffer darsidagi bo'lingan belgi muammosi bu yerda hal qilingan. encoding (yoki stream.setEncoding("utf8")) ichida StringDecoder ishlaydi va ʻ kabi ko'p baytli belgi bo'lak chegarasida buzilmaydi. start/end — faylning faqat bir qismini o'qish (bu yerda 0–2 baytlar, chegaralar kiradi).

5. Ikki rejim: flowing va paused

5.1 Rejim nima

Readable ikki rejimdan birida bo'ladi:

• Flowing (oqayotgan) — ma'lumot o'zi keladi: oqim bo'laklarni "data" hodisasi bilan imkon boricha tez chiqaradi.
• Paused (to'xtatilgan) — ma'lumot oqim ichida kutadi, siz uni read() bilan o'zingiz olasiz.

Holat readableFlowing xususiyatida — uchta qiymat:

import { Readable } from "node:stream";

const stream = Readable.from(["a", "b", "c"]);
console.log(stream.readableFlowing); // null
stream.on("data", () => {});
console.log(stream.readableFlowing); // true
stream.pause();
console.log(stream.readableFlowing); // false
stream.resume();
await new Promise((resolve) => stream.on("end", resolve));
console.log("tugadi");

null holati muhim: yangi oqim o'zi oqib ketmaydi. Kimdir "data" tinglovchisini qo'shmaguncha (yoki pipe/for await boshlanmaguncha) ma'lumot yo'qolmaydi — u kutadi. Shuning uchun "data" ni biroz kechroq qo'shish xavfsiz.

5.2 Qachon nima almashtiradi

stateDiagram-v2
  [*] --> kutish: yangi oqim (null)
  kutish --> oqayotgan: on("data") / resume() / pipe
  oqayotgan --> toxtagan: pause()
  toxtagan --> oqayotgan: resume()
  oqayotgan --> [*]: "end"

Holatlar diagrammasida ikki narsaga qarang. Birinchidan, "data" tinglovchisini qo'shishning o'zi oqimni ishga tushiradi. Ikkinchidan, pause() dan keyin ma'lumot yo'qolmaydi — oqim ichida to'planadi va resume() da davom etadi.

Paused rejimda o'qish — "readable" hodisasi va read(): oqim "ichimda ma'lumot bor" deydi, siz read(n) bilan olasiz. Bu past darajadagi usul — protokol tahlilchilari (masalan "avval 4 bayt sarlavha, keyin tana") uchun. Kundalik kodda for await u bilan bir xil ishni qiladi: har bo'lakni olgach, keyingisini faqat siz tayyor bo'lganingizda so'raydi.

Diqqat: Bitta oqimni ikki usul bilan bir vaqtda o'qimang (masalan "data" va for await). Ikkalasi bir xil bo'laklarga da'vo qiladi — readline dagi question va for await aralashmasi kabi.

Tekshirib ko'ring: const s = createReadStream("f.txt") yaratildi, 2 soniyadan keyin s.on("data", …) qo'shildi. Shu 2 soniyada ma'lumot yo'qoldimi?

6. Writable va Transform: qisqa ko'rinish

Writable — write(chunk) bilan yoziladi, end() bilan yopiladi. Hamma ma'lumot haqiqatan yozib bo'lingach — "finish" hodisasi:

import { createWriteStream, readFileSync } from "node:fs";
import { once } from "node:events";

const out = createWriteStream("chek.txt");
out.write("osh 35000\n");
out.write("choy 5000\n");
out.end("jami 40000\n");
await once(out, "finish");
console.log(readFileSync("chek.txt", "utf8").split("\n").length - 1);

Konsolda:

3

once — EventEmitter chuqur dan. "finish" dan oldin faylni o'qisangiz, unda hali hamma narsa bo'lmasligi mumkin — write asinxron. write() ning qaytargan qiymati ham bor (true/false) — u keyingi darsning markazida: unga qaramaslik 98 MB xotira yeydi.

Transform — kirgan bo'lakni o'zgartirib chiqaradi:

import { Transform } from "node:stream";

const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, String(chunk).toUpperCase());
  },
});
upper.on("data", (chunk) => console.log(String(chunk)));
upper.write("osh");
upper.end("manti");
await new Promise((resolve) => upper.on("end", resolve));

Konsolda:

OSH
MANTI

transform funksiyasi har bo'lak uchun chaqiriladi; callback(xato, natija) — tayyor bo'lak. O'z Readable va Transform oqimlaringizni yozishni O'z Readable/Transform stream'ingizni yozish darsida chuqur o'rganamiz.

7. CLI qadami: holat 35 ga tayyorgarlik (hali commit qilmang)

valyuta ning stream qismi — tarix va tahlil buyruqlari — holat 35 da quriladi va bir nechta darsga bo'lingan: bu dars va Backpressure, pipeline, o'z oqimlari, async iterator va Web Streams. Commit — oxirgi darsda, Katta faylni stream bilan qayta ishlash da. Bu darsda hali commit qilmang.

Holat 35 da qaysi oqim turi qayerda ishlaydi — hozirdan xarita (kanon src/tarix.ts va src/tahlil.ts):

Bu darsdagi amaliy qadam — o'lchov. Loyiha papkangizda (loyihaga qo'shmasdan) fixture faylni oqim bilan o'qing va bo'laklarni sanang:

import { createReadStream } from "node:fs";

const file = "fixtures/kurslar-2026-10-23.json";
for (const highWaterMark of [64 * 1024, 4 * 1024]) {
  let count = 0;
  const stream = createReadStream(file, { highWaterMark });
  for await (const _ of stream) count++;
  console.log(`${highWaterMark} bayt: ${count} bo'lak`);
}

Fixture 25.5 KB: sukut 64 KB bilan — 1 bo'lak (oqimdan foyda yo'q), 4 KB bilan — 7 bo'lak. Xulosa: 25 KB lik CBU javobi uchun readFile to'g'ri tanlov va valyuta uni shunday o'qiydi. Oqim faqat tarix (yuzlab kunlar) va tahlil (million qatorli CSV) uchun kerak.

8. Ko'p uchraydigan xatolar

9. Mashqlar

Mashqlar kurs/mashqlar/24/30-stream/ da.

1-mashq (oson): turini ayting

• process.stdin —
• fs.createWriteStream —
• net.Socket —
• zlib.createGzip() —
• Yangi oqimning readableFlowing qiymati —

2-mashq (o'rta): qator sanagich

countLines(stream) funksiyasini yozing: istalgan matn Readable'dan qatorlar sonini qaytarsin, butun matnni xotirada yig'masdan — har bo'lakda \n larni sanang. Oxirgi qator \n siz tugasa, u ham qator. Sinang: Readable.from(["osh\nman", "ti\nsom", "sa"]) va Readable.from(["a\n", "b\n"]).

import { Readable } from "node:stream";

async function countLines(stream) {
  let count = 0;
  let last = "";
  for await (const chunk of stream) {
    const text = String(chunk);
    for (const ch of text) if (ch === "\n") count++;
    if (text.length) last = text.at(-1);
  }
  return last && last !== "\n" ? count + 1 : count;
}

const menu = Readable.from(["osh\nman", "ti\nsom", "sa"]);
console.log(await countLines(menu));
console.log(await countLines(Readable.from(["a\n", "b\n"])));

3
2

3-mashq (qiyin): narx Transform'i

new Transform({ objectMode: true, transform… }) bilan addTax oqimini yozing: kirishga { dish, price } obyektlari keladi, chiqishga "osh: 39200" kabi satr (12 % QQS qo'shilgan, butun songa yaxlitlangan). Narxi 0 dan kichik obyekt kelsa — callback(new Error(...)). Readable.from dan o'qib, natijani for await bilan chiqaring; xatoni try/catch bilan ushlang. Ma'lumot: osh 35 000, choy 5 000, "xato" −1.

import { Readable, Transform } from "node:stream";

const addTax = new Transform({
  objectMode: true,
  transform({ dish, price }, encoding, callback) {
    if (price < 0) {
      return callback(new Error(`Noto'g'ri narx: ${dish}`));
    }
    callback(null, `${dish}: ${Math.round(price * 1.12)}`);
  },
});

const menu = Readable.from([
  { dish: "osh", price: 35000 },
  { dish: "choy", price: 5000 },
  { dish: "xato", price: -1 },
]);
menu.pipe(addTax);

try {
  for await (const line of addTax) console.log(line);
} catch (error) {
  console.log("ushlandi:", error.message);
}

ushlandi: Noto'g'ri narx: xato

4-mashq: CLI qadami — bo'laklarni o'lchang

«CLI qadami» bo'limidagi skriptni o'z valyuta papkangizda ishga tushiring. Keyin 5 MB lik soxta CSV yarating ("USD,2026-10-23,11903.15\n".repeat(220_000)) va uni ham sanang. Natijalarni yozib qo'ying — Katta faylni stream bilan qayta ishlash darsida xotira bilan solishtiramiz. Hali commit qilmang.

10. Real ishda

• HTTP — Node serverida so'rov (req) — Readable, javob (res) — Writable. Express va boshqa freymvorklarning hammasi shu ustida (25-qism).
• Fayl yuklash va yuklab olish — katta fayl serverdan oqim bilan beriladi; butunligicha xotiraga o'qilsa, bir vaqtda 100 ta yuklab olish serverni yiqitadi.
• Log va ETL — gigabaytlik loglarni tahlil, CSV → baza — oqim + Transform zanjirlari.
• Video va audio — ffmpeg chiqishi child_process dagi spawn orqali oqim bo'lib keladi.
• Intervyuda: "Stream'ning to'rt turi?", "Duplex va Transform farqi?", "10 GB faylni qanday o'qiysiz?".

Xulosa

• Stream — ma'lumotni bo'laklab qayta ishlash; yutug'i — xotira (kanon: 63 MB va 194 MB), tezlik emas.
• To'rt tur: Readable, Writable, Duplex (mustaqil ikki tomon), Transform (chiqish kirishdan).
• Bo'lak hajmi — highWaterMark: fayl o'qishda 64 KB, boshqalarda Windows'da 16 KB (Linux/macOS'da 64 KB), objectMode da 16 obyekt.
• O'qish: for await (xatoni throw qiladi) yoki "data"/"end"; matn uchun encoding: "utf8".
• Rejimlar: null (kutadi), flowing ("data", resume), paused (pause, read); ma'lumot yo'qolmaydi.

Keyingi dars: Backpressure — yozuvchi sekin bo'lsa nima bo'ladi: write() ning false javobi, "drain" va 98 MB ga o'sgan bufer.

Manbalar

• Node.js hujjati — Stream (to'rt tur, highWaterMark, objectMode, "Two reading modes", readableFlowing): https://nodejs.org/docs/latest-v24.x/api/stream.html
• Node.js hujjati — fs.createReadStream (highWaterMark 64 KiB, start/end, encoding): https://nodejs.org/docs/latest-v24.x/api/fs.html#fscreatereadstreampath-options
• Kanon: natijalar/tajriba/t35-katta.txt (41 MB CSV: 63 MB va 194 MB), t31-backpressure.txt (sukut highWaterMark, objectMode), valyuta holat 35 (src/tarix.ts, src/tahlil.ts). Bo'lak va rejim misollari — biz ishga tushirdik (Node 24.21.0, Windows 11, 2026-10-23).