[NODE JS] Writable Stream ile Veri Yazma

Silverhand · 13 Tem 2026

[NODE JS] Writable Stream ile Veri Yazma

Writable Stream, verilerin bir hedefe yazılmasını sağlayan stream türüdür. Dosya yazma, HTTP yanıtları, veritabanı kayıtları ve ağ soketi gibi pek çok hedef Writable Stream olarak modellenir. Readable Stream'in okuma tarafındaki avantajları, Writable Stream yazma tarafında da aynen geçerlidir: büyük verileri parça parça yazabilir, bellek kullanımını kontrol altında tutabilir ve backpressure mekanizmasını kullanabilirsiniz.

Writable Stream'lerle ilgili en sık yapılan hata, backpressure sinyallerini görmezden gelmektir. write() metodu false döndüğünde, bu "şu an yazma kuyruğum dolu, biraz bekle" demektir. Bu sinyali dikkate almazsanız, bellek kullanımı kontrolsüz şekilde artar. Özellikle büyük dosyalar yazarken veya hızlı veri kaynaklarından yavaş hedeflere yazarken bu durum kritik hale gelir.

Bir log sistemi geliştirirken, yüksek trafikli bir uygulamadan gelen log mesajlarını dosyaya yazıyordum. Başlangıçta backpressure'ı görmezden geldiğim için, yoğun trafik anlarında bellek kullanımı gigabyte'larca artıyordu. drain olayını dinleyerek bu sorunu çözdüm. drain olayı, yazma kuyruğu boşaldığında tetiklenir ve güvenli şekilde yazmaya devam edebileceğinizi bildirir.

Writable Stream'lerin yaşam döngüsünü iyi anlamak önemlidir. write() ile veri yazarsınız, end() ile stream'i sonlandırırsınız ve finish olayı tüm veriler hedefe yazıldığında tetiklenir. end() çağrıldıktan sonra artık write() çağıramazsınız. close olayı ise altta yatan kaynağın kapatıldığını bildirir. Bu sırayı takip etmek, veri kaybını önler.

Kendi Writable Stream'inizi oluşturmak için _write, _writev ve _final metodlarını implement edebilirsiniz. _write tek bir parça için, _writev birden fazla parça için ve _final stream sonlandırılırken çağrılır. Bu özelleştirme, veritabanı, mesaj kuyruğu veya özel formatlara yazmak istediğinizde işinize yarar.

Üretim ortamında Writable Stream kullanırken, hata yönetimi ihmal edilmemesi gereken bir konudur. error olayını mutlaka dinleyin. Dosya sistemindeki izin sorunları, disk dolması, ağ hataları gibi pek çok senaryo yazma sırasında hata üretebilir. Hataları yakalamadan bırakırsanız, uygulamanız sessizce veri kaybedebilir.

JavaScript:

// Writable Stream ile Veri Yazma
const { Writable } = require('stream');
const fs = require('fs');
const path = require('path');

// ==========================================
// 1. Dosyaya Stream ile Yazma
// ==========================================
async function dosyayaYaz() {
  console.log('--- Dosyaya Stream Yazma ---');

  const dosyaYolu = path.join(__dirname, 'cikti.log');
  const yazmaAkisi = fs.createWriteStream(dosyaYolu, {
    encoding: 'utf-8',
    highWaterMark: 16 * 1024, // 16KB tampon
    flags: 'a' // Dosyanın sonuna ekle
  });

  // Hata olayını dinle
  yazmaAkisi.on('error', (hata) => {
    console.error('Yazma hatası:', hata.message);
  });

  // Stream bittiğinde
  yazmaAkisi.on('finish', () => {
    console.log('Dosyaya yazma tamamlandı');
  });

  // Büyük miktarda veri yazma (backpressure yönetimi ile)
  const toplamSatir = 10000;
  let yazilan = 0;

  function satirYaz() {
    let devamEt = true;

    while (yazilan < toplamSatir && devamEt) {
      yazilan++;
      const tarih = new Date().toISOString();
      const seviye = ['BİLGİ', 'UYARI', 'HATA', 'DEBUG'][Math.floor(Math.random() * 4)];
      const mesaj = '[' + tarih + '] [' + seviye + '] İşlem #' + yazilan
        + ' gerçekleştirildi - Kullanıcı: user_' + Math.floor(Math.random() * 100) + '
';

      if (yazilan === toplamSatir) {
        // Son satır - end() ile yaz
        yazmaAkisi.end(mesaj);
      } else {
        // write() false dönerse backpressure var
        devamEt = yazmaAkisi.write(mesaj);
      }
    }

    if (yazilan < toplamSatir) {
      // Tampon doldu, drain olayını bekle
      console.log('Tampon doldu, bekleniyor... (yazılan: ' + yazilan + ')');
      yazmaAkisi.once('drain', () => {
        console.log('Tampon boşaldı, devam ediliyor...');
        satirYaz();
      });
    }
  }

  satirYaz();
}

// ==========================================
// 2. Özel Writable Stream - Veritabanı Yazıcı
// ==========================================
class VeritabaniYazici extends Writable {
  constructor(secenekler = {}) {
    super({
      objectMode: true,
      highWaterMark: secenekler.tamponBoyutu || 50
    });
    this.tablo = secenekler.tablo || 'kayitlar';
    this.topluEkleBoyutu = secenekler.topluEkleBoyutu || 100;
    this.tampon = [];
    this.toplamYazilan = 0;
  }

  // Tek bir kayıt için çağrılır
  async _write(kayit, kodlama, geriCagir) {
    this.tampon.push(kayit);

    // Tampon dolduğunda toplu ekleme yap
    if (this.tampon.length >= this.topluEkleBoyutu) {
      try {
        await this.topluEkle();
        geriCagir(); // Başarılı
      } catch (hata) {
        geriCagir(hata); // Hata durumu
      }
    } else {
      geriCagir(); // Tampona eklendi, devam et
    }
  }

  // Birden fazla kayıt için çağrılır (performans optimizasyonu)
  async _writev(parcalar, geriCagir) {
    for (const parca of parcalar) {
      this.tampon.push(parca.chunk);
    }

    try {
      if (this.tampon.length >= this.topluEkleBoyutu) {
        await this.topluEkle();
      }
      geriCagir();
    } catch (hata) {
      geriCagir(hata);
    }
  }

  // Stream sonlandırılırken kalan verileri yaz
  async _final(geriCagir) {
    try {
      if (this.tampon.length > 0) {
        await this.topluEkle();
      }
      console.log('[' + this.tablo + '] Toplam ' + this.toplamYazilan + ' kayıt yazıldı');
      geriCagir();
    } catch (hata) {
      geriCagir(hata);
    }
  }

  // Toplu ekleme simülasyonu
  async topluEkle() {
    const kayitlar = this.tampon.splice(0); // Tamponu boşalt

    // Gerçek projede burada veritabanı INSERT işlemi olur
    console.log('[' + this.tablo + '] ' + kayitlar.length + ' kayıt toplu ekleniyor...');

    // Veritabanı gecikmesi simülasyonu
    await new Promise(resolve => setTimeout(resolve, 100));

    this.toplamYazilan += kayitlar.length;
  }
}

// ==========================================
// 3. Log Yazıcı Stream
// ==========================================
class LogYazici extends Writable {
  constructor(secenekler = {}) {
    super({ objectMode: true });
    this.dosyaYolu = secenekler.dosyaYolu || 'uygulama.log';
    this.konsoldaGoster = secenekler.konsoldaGoster !== false;
    this.dosyaAkisi = fs.createWriteStream(this.dosyaYolu, { flags: 'a' });
    this.formatci = secenekler.formatci || this.varsayilanFormat;
  }

  varsayilanFormat(kayit) {
    const zaman = new Date().toISOString();
    const seviye = (kayit.seviye || 'BİLGİ').toUpperCase().padEnd(5);
    return '[' + zaman + '] [' + seviye + '] ' + kayit.mesaj;
  }

  _write(kayit, kodlama, geriCagir) {
    const formatliMesaj = this.formatci(kayit) + '
';

    if (this.konsoldaGoster) {
      const renkler = {
        'HATA': '[31m',  // Kırmızı
        'UYARI': '[33m', // Sarı
        'BİLGİ': '[36m', // Mavi
        'DEBUG': '[90m'  // Gri
      };
      const renk = renkler[(kayit.seviye || 'BİLGİ').toUpperCase()] || '[0m';
      process.stdout.write(renk + formatliMesaj + '[0m');
    }

    this.dosyaAkisi.write(formatliMesaj, geriCagir);
  }

  _final(geriCagir) {
    this.dosyaAkisi.end(geriCagir);
  }
}

// ==========================================
// 4. HTTP Yanıt Stream'i (Express Örneği)
// ==========================================
function httpYanitOrnegi() {
  const express = require('express');
  const app = express();

  // Büyük veri setini stream olarak gönderme
  app.get('/api/buyuk-veri', (req, res) => {
    // Yanıt başlıklarını ayarla
    res.setHeader('Content-Type', 'application/json');
    res.setHeader('Transfer-Encoding', 'chunked');

    // JSON array başlangıcı
    res.write('[
');

    let sayac = 0;
    const toplamKayit = 10000;

    function parcaGonder() {
      let devamEt = true;

      while (sayac < toplamKayit && devamEt) {
        sayac++;
        const kayit = JSON.stringify({
          id: sayac,
          isim: 'Ürün ' + sayac,
          fiyat: (Math.random() * 1000).toFixed(2),
          tarih: new Date().toISOString()
        });

        const ayrac = sayac < toplamKayit ? ',
' : '
';
        devamEt = res.write(kayit + ayrac);
      }

      if (sayac >= toplamKayit) {
        res.end(']');
      } else {
        res.once('drain', parcaGonder);
      }
    }

    parcaGonder();
  });

  return app;
}

// ==========================================
// 5. Kullanım Örnekleri
// ==========================================
async function basla() {
  // Veritabanı yazıcı kullanımı
  console.log('--- Veritabanı Yazıcı ---');
  const dbYazici = new VeritabaniYazici({
    tablo: 'kullanicilar',
    topluEkleBoyutu: 25
  });

  // 100 kayıt yazma
  for (let i = 0; i < 100; i++) {
    const yazildi = dbYazici.write({
      id: i + 1,
      ad: 'Kullanıcı ' + (i + 1),
      eposta: 'kullanici' + (i + 1) + '@ornek.com',
      kayitTarihi: new Date().toISOString()
    });

    // Backpressure kontrolü
    if (!yazildi) {
      await new Promise(resolve => dbYazici.once('drain', resolve));
    }
  }

  // Stream'i sonlandır
  await new Promise((resolve, reject) => {
    dbYazici.end(() => resolve());
    dbYazici.on('error', reject);
  });

  // Log yazıcı kullanımı
  console.log('\n--- Log Yazıcı ---');
  const logger = new LogYazici({
    dosyaYolu: 'uygulama.log',
    konsoldaGoster: true
  });

  logger.write({ seviye: 'BİLGİ', mesaj: 'Uygulama başlatıldı' });
  logger.write({ seviye: 'DEBUG', mesaj: 'Veritabanı bağlantısı kontrol ediliyor' });
  logger.write({ seviye: 'UYARI', mesaj: 'Bellek kullanımı %80 üzerine çıktı' });
  logger.write({ seviye: 'HATA', mesaj: 'Ödeme servisi yanıt vermedi' });
  logger.write({ seviye: 'BİLGİ', mesaj: 'Görev kuyruğu işleniyor - 15 görev bekliyor' });

  logger.end(() => {
    console.log('\nLog yazıcı kapatıldı');
  });
}

basla().catch(console.error);

Writable Stream'ler, veriyi hedeflere etkin şekilde yazmak için güçlü bir abstraksiyon sunar. Backpressure yönetimi, hata kontrolü ve yaşam döngüsü olaylarını iyi anladığınızda, yüksek performanslı ve güvenilir yazma işlemleri gerçekleştirebilirsiniz. Bir sonraki konumuzda Transform Stream ile veri dönüştürmeyi ele alacağız.

Best Studio · Çarşamba saat 16:17'de

Emeğiniz ve katkınız için teşekkür ederiz!
Konunuz gerekli incelemelerden geçerek onaylandı. Topluluğumuza katkılarınızın devamını bekliyoruz.
İyi forumlar!

[NODE JS] Writable Stream ile Veri Yazma

Silverhand