Node.js Stream Nedir? Readable, Writable, Duplex ve Transform

Neden stream kullanılır?

Node.js Learn rehberi stream'lerin üç avantajını öne çıkarır. Birincisi bellek verimliliği: 2 GB'lık bir log dosyasını readFile ile okumak 2 GB bellek ister, stream ile ise her an yalnızca birkaç onlarca kilobaytlık bir parça bellekte durur. İkincisi yanıt süresi: ilk parça geldiği anda işlemeye başlarsınız, tüm verinin gelmesini beklemezsiniz. Üçüncüsü ölçeklenebilirlik: sınırlı kaynakla çok miktarda veri işleyebilirsiniz. Rehberin bir de dürüst notu var: veri zaten tamamen bellekteyse stream kullanmak gereksiz karmaşıklık ve ek yük getirebilir. Küçük bir JSON dosyası için readFile çoğu zaman daha doğru seçimdir.

Duplex ve Transform arasındaki fark önemlidir. Duplex stream'in iki tarafı birbirinden bağımsızdır: bir TCP soketine yazdığınız veri ile soketten okuduğunuz veri farklı şeylerdir. Transform'da ise okunan taraf, yazılan verinin işlenmiş halidir; gzip stream'ine düz metin yazarsınız, ondan sıkıştırılmış bayt okursunuz. Belgeler her iki türün de okuma ve yazma için ayrı iç tamponlar tuttuğunu belirtir.

Readable stream: olaylar ve iki okuma modu

Okunabilir bir stream'i tüketmenin klasik yolu 'data' olayını dinlemektir. Her parça geldikçe dinleyici çağrılır, veri bittiğinde 'end', alttaki kaynak kapandığında 'close' olayı gelir. Aşağıdaki örnek bir diziden Readable.from() ile stream üretir:

import { Readable } from 'node:stream';

const kaynak = Readable.from(['elma', 'armut', 'kiraz']);

kaynak.on('data', (parca) => console.log('data:', parca));
kaynak.on('end', () => console.log('end: veri bitti'));
kaynak.on('close', () => console.log('close: kaynak kapandı'));
// çıktı: data: elma
// çıktı: data: armut
// çıktı: data: kiraz
// çıktı: end: veri bitti
// çıktı: close: kaynak kapandı

Belgelere göre her Readable stream paused (duraklatılmış) modda başlar ve şu üç yoldan biriyle flowing (akan) moda geçer: 'data' dinleyicisi eklemek, resume() çağırmak ya da pipe() ile bir hedefe bağlamak. Flowing modda veri olabildiğince hızlı olaylarla iletilir; paused modda ise read() çağırarak veriyi siz çekersiniz. Temel kural şudur: bir Readable, veriyi tüketen ya da yok sayan bir mekanizma olmadan veri üretmez. Tersine, flowing moda geçip hiçbir tüketici yoksa veri kaybolur; örneğin 'data' dinleyicisi olmadan resume() çağırmak veriyi boşa akıtır.

İçeride durum aslında üç değerlidir: readable.readableFlowing başta null'dır (tüketici yok), tüketici eklenince true, pause() veya backpressure ile false olur. false durumundayken yeni bir 'data' dinleyicisi eklemek stream'i yeniden akıtmaz, resume() gerekir. Belgelerin uyarısı açık: aynı stream'i on('data'), on('readable'), pipe() ve async iterator yöntemlerinden birden fazlasıyla tüketmeyin; tek bir yöntem seçin.

for await...of ile okumak (önerilen yol)

Node.js Learn rehberi, stream'lerle çalışmanın standart yolu olarak async iterator'ları önerir. Tüm okunabilir stream'ler async iterable'dır; for await...of ile parçaları sırayla alır, hataları try/catch ile yakalarsınız. Tüketici bir sonraki parçayı await ile beklediği için akış kontrolü de doğal olarak sağlanır. Async iterator ve generator kavramlarını JavaScript generator ve iterator yazısında bulabilirsiniz.

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

await writeFile('ornek.txt', 'satır 1\nsatır 2\nsatır 3\n'.repeat(10000));

const okuyucu = createReadStream('ornek.txt', { encoding: 'utf8' });
let parcaSayisi = 0;
let karakter = 0;
for await (const parca of okuyucu) {
  parcaSayisi++;
  karakter += parca.length;
}
console.log({ parcaSayisi, karakter }); // çıktı: { parcaSayisi: 5, karakter: 240000 }
console.log(okuyucu.readableHighWaterMark); // çıktı: 65536

Dosya 270.000 bayt (Türkçe "ı" iki bayt tuttuğu için karakter sayısından fazla) ve fs.createReadStream'in varsayılan highWaterMark değeri 64 * 1024 bayt olduğu için veri beş parça halinde geldi. Belgelere göre döngü break, return ya da throw ile erken biterse stream yok edilir (destroy). Yani yarım bıraktığınız bir iterasyon dosya tanımlayıcısını açık bırakmaz, ama aynı stream'i daha sonra tekrar okuyamazsınız. encoding seçeneği verildiği için parçalar string geldi; çok baytlı karakterlerin bölünmesi sorununu Node bizim için çözdü (Buffer yazısındaki StringDecoder bölümüne bakın).

Writable stream: write, end ve 'finish'

Yazılabilir stream'e write(parca) ile veri gönderir, işiniz bitince end() çağırırsınız. end() son bir parça da alabilir. Tüm veri alttaki sisteme teslim edildiğinde 'finish', dosya tanımlayıcısı kapandığında 'close' olayı gelir. end()'den sonra write() çağırmak hatadır.

import { createWriteStream } from 'node:fs';
import { readFile } from 'node:fs/promises';

const yazici = createWriteStream('log.txt');
yazici.write('ilk satır\n');
yazici.write('ikinci satır\n');
yazici.end('son satır\n');

yazici.on('finish', () => console.log('finish: tüm veri işletim sistemine verildi'));
yazici.on('close', async () => {
  console.log('close: dosya kapandı');
  console.log(await readFile('log.txt', 'utf8'));
});
// çıktı: finish: tüm veri işletim sistemine verildi
// çıktı: close: dosya kapandı
// çıktı: ilk satır
// çıktı: ikinci satır
// çıktı: son satır

Olaylar her zaman bu sırayla gelir: önce 'finish', ardından dosya tanımlayıcısı kapatılınca 'close'. Dosyayı tekrar okumayı 'close' dinleyicisine koyduk. İlk denememizde okumayı 'finish' içine koymuştuk; readFile asenkron çalıştığı için 'close' mesajının dosya içeriğinden önce mi sonra mı yazılacağı çalıştırmadan çalıştırmaya değişti. Sıranın önemli olduğu yerde, beklediğiniz durumu kesin olarak bildiren olayı dinleyin. write()'ın bir de dönüş değeri vardır: iç tampon highWaterMark eşiğine ulaştığında false döner ve 'drain' olayına kadar yazmayı bırakmanızı ister. Bu mekanizma (backpressure) pipeline ve backpressure yazısında ayrıntılı ve ölçülü örneklerle anlatılıyor. Dosyaya tek seferde metin yazmak için ise writeFile ve appendFile yazısına bakabilirsiniz.

Kendi stream'inizi oluşturmak: Readable, Transform, Writable

Belgelerin "Simplified construction" bölümü, sınıf türetmeden yalnızca yapıcıya fonksiyon vererek stream oluşturmayı gösterir. Readable için read(), Writable için write(chunk, encoding, callback), Transform için transform(chunk, encoding, callback) yazarsınız. Readable'da veriyi this.push() ile kuyruğa koyar, akışın bittiğini this.push(null) ile bildirirsiniz. Writable ve Transform'da her parçayı işledikten sonra callback() çağırmak zorunludur; çağırmazsanız stream bir sonraki parçayı hiç istemez.

import { Readable, Writable, Transform } from 'node:stream';

let sayac = 0;
const sayilar = new Readable({
  objectMode: true,
  read() {
    sayac++;
    this.push(sayac <= 3 ? sayac : null);
  },
});

const kare = new Transform({
  objectMode: true,
  transform(sayi, _enc, callback) {
    callback(null, sayi * sayi);
  },
});

const yazdir = new Writable({
  objectMode: true,
  write(deger, _enc, callback) {
    console.log('gelen:', deger);
    callback();
  },
});

sayilar.pipe(kare).pipe(yazdir).on('finish', () => console.log('bitti'));
// çıktı: gelen: 1
// çıktı: gelen: 4
// çıktı: gelen: 9
// çıktı: bitti

Transform'da callback(null, deger) hem parçanın işlendiğini bildirir hem de sonucu okunur tarafa iter. Belgeler stream uygulayan kodun tüketiciye yönelik genel yöntemleri (write(), end(), destroy()) ezmemesini ve 'data', 'end', 'error' gibi iç olayları emit() ile elle yaymamasını ister; bu, stream'in iç tutarlılığını bozar. Örnekteki pipe() zinciri kısa olduğu için kullanıldı; gerçek projelerde hataları da yöneten pipeline() tercih edilmelidir.

pipe() hedef stream'i döndürdüğü için zincir kurulabilir ve kaynak 'end' yaydığında varsayılan olarak hedefte end() çağrılır. Hedefi açık tutmak isterseniz { end: false } seçeneğini verirsiniz; örneğin bir dosyanın içeriğini aktarıp ardından aynı hedefe kapanış satırı yazmak için. Belgeler bir istisnayı da not eder: process.stdout ve process.stderr, seçenekten bağımsız olarak süreç bitene kadar hiçbir zaman kapatılmaz. Belgelere göre tek bir Readable'a birden fazla Writable bağlamak da mümkündür.

Nesne modu ve highWaterMark

Node'un kendi stream'leri yalnızca string, Buffer, TypedArray ve DataView taşır. Sayı ya da nesne taşımak için stream'i oluştururken objectMode: true vermeniz gerekir; var olan bir stream'i sonradan nesne moduna geçirmek belgelere göre güvenli değildir. null değeri her iki modda da özeldir: akış sonunu belirtir, veri olarak gönderilemez. Readable.from() ise varsayılan olarak nesne modunu açar; ilk örnekteki dizi elemanlarının ayrı ayrı gelmesi bu yüzdendir.

highWaterMark, bir stream'in daha fazla veri istemeyi bırakmadan önce tamponlayacağı miktardır. Normal stream'lerde bayt, nesne modunda nesne sayısı olarak ölçülür. Belgeler bunun katı bir bellek sınırı değil bir eşik olduğunu vurgular. Varsayılan değer stream.getDefaultHighWaterMark() ile okunur: nesne modunda 16; bayt stream'lerinde Windows dışındaki platformlarda 65536 (64 KiB), Windows'ta 16384 (16 KiB). Aşağıdaki çıktı Windows'ta alındı. Aynı örnek, node:stream/consumers modülünün text() ve json() yardımcılarıyla bir stream'i tek satırda tüketmeyi ve nesne modu olmayan stream'e nesne itmenin sonucunu da gösteriyor:

import { Readable, getDefaultHighWaterMark } from 'node:stream';
import { text, json } from 'node:stream/consumers';

console.log(getDefaultHighWaterMark(false), getDefaultHighWaterMark(true)); // çıktı: 16384 16

const metin = await text(Readable.from(['Mer', 'haba']));
console.log(metin); // çıktı: Merhaba

const nesne = await json(Readable.from(['{"ad":', '"Ayşe"}']));
console.log(nesne); // çıktı: { ad: 'Ayşe' }

const r = new Readable({ read() {} });
r.on('error', (err) => console.log(err.name, err.code)); // çıktı: TypeError ERR_INVALID_ARG_TYPE
r.push({ a: 1 });

Nesne modu kapalı stream'e nesne itmek bir istisna fırlatmadı; stream bir 'error' olayı yaydı. Bu olayı dinlemeseydik süreç "Unhandled 'error' event" ile çökecekti. stream/consumers ailesinde text, json, buffer, bytes, arrayBuffer ve blob bulunur; tüm gövdeyi belleğe aldıkları için yalnızca boyutu makul verilerde kullanın.

Hatalar ve Web Streams ile dönüşüm

Bir stream'de hata, 'error' olayıyla bildirilir. for await döngüsü kullanıyorsanız bu hata döngüden istisna olarak fırlar ve try/catch ile yakalanır; belgeler async iterator'ın stream'e kalıcı bir hata dinleyicisi eklediğini de belirtir. Kendi Readable'ınızda hata bildirmek için this.destroy(err) çağırırsınız.

Node'da ikinci bir stream uygulaması daha vardır: WHATWG standardındaki Web Streams (ReadableStream, WritableStream, TransformStream). Learn rehberi ikisinin kavramsal olarak benzer ama API olarak doğrudan uyumlu olmadığını hatırlatır. fetch() yanıt gövdesi bir Web ReadableStream'dir; Node stream bekleyen bir API'ye vermek için Readable.fromWeb(), tersi için Readable.toWeb() kullanılır:

import { Readable } from 'node:stream';

const bozuk = new Readable({
  read() {
    this.destroy(new Error('disk okunamadı'));
  },
});

try {
  for await (const parca of bozuk) {
    console.log(parca);
  }
} catch (err) {
  console.log('yakalandı:', err.message); // çıktı: yakalandı: disk okunamadı
}
console.log(bozuk.destroyed); // çıktı: true

const web = Readable.toWeb(Readable.from(['a', 'b']));
console.log(web.constructor.name); // çıktı: ReadableStream
const geri = Readable.fromWeb(web, { objectMode: true });
for await (const x of geri) console.log(x);
// çıktı: a
// çıktı: b

Belgelerde stream.Readable sayfasında Stability: 1 - Experimental işaretli bazı yöntemler (ör. map, filter, toArray gibi yardımcılar) bulunur; bunlar deneyseldir ve davranışları değişebilir. Bu yazıdaki örnekler yalnızca kararlı API'leri kullanır. Seride kapsam dışı bırakılan deneysel stream/iter modülü de ayrı bir konudur.

Uygulamalı mini örnek: büyük bir log dosyasında satır saymak

Birkaç gigabaytlık bir erişim logunda kaç satır olduğunu bulmak istediğinizi düşünün. readFile ile tüm dosyayı okumak bellek tüketir. Bunun yerine createReadStream ile dosyayı açar, for await ile her parçayı alır ve parçadaki satır sonu baytlarını (0x0a) sayarsınız. Parçaları Buffer olarak bırakmak (encoding vermemek) burada daha verimlidir, çünkü satır sonu tek baytlık bir karakterdir ve string'e çevirmeye gerek yoktur. Satır satır anlamlı işlem yapacaksanız (ör. her satırı ayrıştırmak) parçalar satır ortasından bölündüğü için kendi tamponunuzu tutmanız ya da readline modülünü kullanmanız gerekir.

Sık yapılan hatalar

• 'error' olayını dinlememek: Dinleyicisi olmayan bir 'error' olayı süreci çökertir. Ya dinleyici ekleyin ya da hataları kendisi yöneten pipeline() veya for await kullanın.
• Aynı stream'i birden fazla yöntemle tüketmek: on('data') ile for await ya da pipe() karışınca veri kaybı ve beklenmedik sıralar görülür.
• Transform ya da Writable'da callback çağırmayı unutmak: Stream sonraki parçayı istemez ve akış sessizce takılır.
• 'end' olayının gelmesini beklemek ama veriyi tüketmemek: Belgelere göre 'end' ancak veri tamamen tüketildiğinde gelir.
• Küçük, bellekteki veri için stream kurmak: Learn rehberinin notuyla, veri zaten bellekteyse stream ek yük getirir.
• write() dönüş değerini yok saymak: Hızlı üretip yavaş hedefe yazarken bellek şişer; ayrıntı backpressure yazısında.

Sık Sorulan Sorular

'finish' ile 'close' arasındaki fark nedir?

'finish', end() çağrıldıktan sonra tüm verinin alttaki sisteme teslim edildiğini bildirir. 'close' ise stream ve kullandığı kaynak (ör. dosya tanımlayıcısı) kapandığında gelir ve artık başka olay olmayacağını gösterir.

Stream okumak için 'data' mı for await mi kullanmalıyım?

Yeni kodda Node.js Learn rehberi async iterator'ları önerir: okunması kolaydır, hatalar try/catch ile yakalanır ve tüketici hızında ilerler. 'data' olayı eski kodda ve olay tabanlı yapılarda hâlâ yaygındır.

highWaterMark değerini büyütmek hızlandırır mı?

Belgeler bunu bir eşik olarak tanımlar, performans garantisi vermez. Daha büyük değer daha az ama daha büyük parça demektir ve bellek kullanımını artırır. Ölçmeden değiştirmeyin; performans ölçümü için Node.js performans optimizasyonu yazısına bakabilirsiniz.

Node stream ile Web ReadableStream aynı şey mi?

Hayır. Kavram benzer ama API farklıdır. fetch ve Blob.stream() Web stream döndürür; fs, http, zlib Node stream kullanır. Readable.fromWeb() ve Readable.toWeb() aradaki köprüdür.

Marka adları sahiplerine aittir. Bu yazı bağımsız bir teknik rehberdir; adı geçen kuruluşlarla bayilik veya ortaklık ilişkisi bulunduğu anlamına gelmez.

İlgili içerikler: Node.js öğrenme rehberi · stream.pipeline ve backpressure · Node.js Buffer kullanımı · EventEmitter kullanımı · Node.js dosya okuma · zlib ile sıkıştırma