[NODE JS] Redis Pub/Sub ile Mesajlaşma

Silverhand · Pazartesi saat 10:43'de

[NODE JS] Redis Pub/Sub ile Mesajlaşma

Redis Pub/Sub (Publish/Subscribe), gerçek zamanlı mesajlaşma sistemi kurmanın en basit ve etkili yollarından biridir. Bir mikroservis mimarisinde servisler arasında iletişimi sağlamak için ilk tercihim genellikle Redis Pub/Sub olur. Kurulumu kolay, performansı yüksek ve zaten Redis kullanıyorsanız ekstra bir altyapıya ihtiyaç duymazsınız.

Pub/Sub modelinin temel konsepti çok basittir: yayıncılar (publishers) belirli kanallara mesaj gönderir, aboneler (subscribers) ise ilgilendikleri kanalları dinler. Yayıncı ve abone birbirinden tamamen bağımsızdır. Bu gevşek bağlılık (loose coupling), sistemin ölçeklenebilirliğini ve bakım kolaylığını artırır. Yeni bir servis eklemek istediğinizde, mevcut servisleri değiştirmenize gerek kalmaz.

Ancak Redis Pub/Sub'ın önemli bir kısıtlaması vardır: mesajlar kalıcı değildir. Eğer bir abone çevrimdışıyken mesaj gönderilirse, o mesajı kaçırır. Bu yüzden kritik mesajlar için Redis Streams veya RabbitMQ gibi kalıcı mesaj kuyrukları tercih edilmelidir. Pub/Sub daha çok anlık bildirimler, cache invalidation ve gerçek zamanlı güncellemeler için uygundur.

Bir sosyal medya projesinde anlık bildirim sistemi kurarken Redis Pub/Sub kullandım. Kullanıcı bir gönderiyi beğendiğinde, bir mesaj yayınlanıyor ve bildirim servisi bu mesajı alıp ilgili kullanıcıya WebSocket üzerinden iletiyordu. Gecikme neredeyse sıfırdı ve sistem binlerce eşzamanlı kullanıcıyı rahatça kaldırabiliyordu.

Pattern matching özelliği sayesinde, belirli bir desene uyan tüm kanalları dinleyebilirsiniz. Örneğin "siparis:*" deseniyle tüm sipariş kanallarını dinleyebilirsiniz. Bu, özellikle loglama ve izleme servisleri için son derece kullanışlıdır. Tek bir abonelikle tüm sipariş olaylarını yakalayabilirsiniz.

Redis Pub/Sub kullanırken dikkat edilmesi gereken bir diğer nokta, abone bağlantısının özel olmasıdır. Bir Redis bağlantısı SUBSCRIBE moduna geçtiğinde, o bağlantı üzerinden başka komut çalıştıramazsınız. Bu yüzden abone işlemleri için ayrı bir Redis bağlantısı oluşturmanız gerekir. ioredis bu durumu otomatik yönetebilir, ancak bağlantı sayısını göz önünde bulundurmanız önemlidir.

Üretim ortamında Pub/Sub kullanırken, mesaj formatını standartlaştırmanızı öneririm. JSON formatında, zaman damgası, kaynak servis adı ve mesaj tipi bilgilerini içeren bir yapı kullanmak, hata ayıklamayı ve izlemeyi kolaylaştırır. Ayrıca büyük boyutlu mesajlardan kaçının; Redis Pub/Sub, küçük ve hızlı mesajlar için optimize edilmiştir.

JavaScript:

// Redis Pub/Sub ile Mesajlaşma Sistemi
const Redis = require('ioredis');

// Yayıncı ve abone için ayrı bağlantılar gereklidir
const yayinci = new Redis({ host: '127.0.0.1', port: 6379 });
const abone = new Redis({ host: '127.0.0.1', port: 6379 });
const desenAbone = new Redis({ host: '127.0.0.1', port: 6379 });

// ==========================================
// Mesaj Yönetici Sınıfı
// ==========================================
class MesajYonetici {
  constructor(yayinciIstemci, aboneIstemci) {
    this.yayinci = yayinciIstemci;
    this.abone = aboneIstemci;
    this.dinleyiciler = new Map();
  }

  // Mesaj formatını standartlaştırma
  mesajOlustur(tip, veri, kaynak) {
    return JSON.stringify({
      id: Date.now().toString(36) + Math.random().toString(36).substr(2, 5),
      tip: tip,
      veri: veri,
      kaynak: kaynak || 'bilinmeyen',
      zamanDamgasi: new Date().toISOString()
    });
  }

  // Kanala mesaj yayınlama
  async yayinla(kanal, tip, veri, kaynak) {
    const mesaj = this.mesajOlustur(tip, veri, kaynak);
    const aboneSayisi = await this.yayinci.publish(kanal, mesaj);
    console.log('[YAYIN] Kanal:', kanal, '- Abone sayısı:', aboneSayisi);
    return aboneSayisi;
  }

  // Kanala abone olma
  async kanalaDinle(kanal, isleyici) {
    await this.abone.subscribe(kanal);
    this.dinleyiciler.set(kanal, isleyici);
    console.log('[ABONE] Kanal dinleniyor:', kanal);
  }

  // Mesaj dinleme olayını başlatma
  dinlemeyeBasla() {
    this.abone.on('message', (kanal, mesajStr) => {
      try {
        const mesaj = JSON.parse(mesajStr);
        console.log('[MESAJ ALINDI] Kanal:', kanal, '- Tip:', mesaj.tip);

        const isleyici = this.dinleyiciler.get(kanal);
        if (isleyici) {
          isleyici(mesaj, kanal);
        }
      } catch (hata) {
        console.error('Mesaj işleme hatası:', hata.message);
      }
    });
  }
}

// ==========================================
// Desen Tabanlı Abonelik (Pattern Subscribe)
// ==========================================
class DesenDinleyici {
  constructor(aboneIstemci) {
    this.abone = aboneIstemci;
    this.dinleyiciler = new Map();
  }

  async desenDinle(desen, isleyici) {
    await this.abone.psubscribe(desen);
    this.dinleyiciler.set(desen, isleyici);
    console.log('[DESEN ABONE] Desen dinleniyor:', desen);
  }

  dinlemeyeBasla() {
    this.abone.on('pmessage', (desen, kanal, mesajStr) => {
      try {
        const mesaj = JSON.parse(mesajStr);
        console.log('[DESEN MESAJ] Desen:', desen, '- Kanal:', kanal);

        const isleyici = this.dinleyiciler.get(desen);
        if (isleyici) {
          isleyici(mesaj, kanal, desen);
        }
      } catch (hata) {
        console.error('Desen mesaj hatası:', hata.message);
      }
    });
  }
}

// ==========================================
// Bildirim Servisi Örneği
// ==========================================
class BildirimServisi {
  constructor(mesajYonetici) {
    this.yonetici = mesajYonetici;
  }

  async basla() {
    // Farklı bildirim kanallarını dinle
    await this.yonetici.kanalaDinle('bildirim:siparis', (mesaj) => {
      console.log('--- Sipariş Bildirimi ---');
      console.log('Sipariş No:', mesaj.veri.siparisNo);
      console.log('Durum:', mesaj.veri.durum);
      console.log('Müşteri:', mesaj.veri.musteriAdi);
      // Burada e-posta, SMS veya push bildirim gönderilebilir
    });

    await this.yonetici.kanalaDinle('bildirim:stok', (mesaj) => {
      console.log('--- Stok Uyarısı ---');
      console.log('Ürün:', mesaj.veri.urunAdi);
      console.log('Kalan stok:', mesaj.veri.kalanStok);
      if (mesaj.veri.kalanStok < 10) {
        console.log('DİKKAT: Stok kritik seviyede!');
      }
    });

    await this.yonetici.kanalaDinle('bildirim:kullanici', (mesaj) => {
      console.log('--- Kullanıcı Bildirimi ---');
      console.log('İşlem:', mesaj.veri.islem);
      console.log('Kullanıcı:', mesaj.veri.kullaniciAdi);
    });

    this.yonetici.dinlemeyeBasla();
    console.log('Bildirim servisi başlatıldı');
  }
}

// ==========================================
// Sipariş Servisi Örneği (Yayıncı)
// ==========================================
class SiparisServisi {
  constructor(mesajYonetici) {
    this.yonetici = mesajYonetici;
  }

  async yeniSiparis(siparisDetay) {
    console.log('Yeni sipariş oluşturuluyor...');

    // Veritabanına kaydet (simülasyon)
    const siparisNo = 'SIP-' + Date.now();

    // Sipariş bildirimi yayınla
    await this.yonetici.yayinla('bildirim:siparis', 'yeni_siparis', {
      siparisNo: siparisNo,
      durum: 'olusturuldu',
      musteriAdi: siparisDetay.musteriAdi,
      toplamTutar: siparisDetay.toplamTutar,
      urunSayisi: siparisDetay.urunler.length
    }, 'siparis-servisi');

    // Stok güncelleme bildirimi
    for (const urun of siparisDetay.urunler) {
      await this.yonetici.yayinla('bildirim:stok', 'stok_azaldi', {
        urunId: urun.id,
        urunAdi: urun.isim,
        kalanStok: urun.stok - urun.adet
      }, 'siparis-servisi');
    }

    return siparisNo;
  }
}

// ==========================================
// Loglama Servisi (Desen Tabanlı Dinleme)
// ==========================================
class LoglamaServisi {
  constructor() {
    this.dinleyici = new DesenDinleyici(desenAbone);
  }

  async basla() {
    // Tüm bildirim kanallarını dinle
    await this.dinleyici.desenDinle('bildirim:*', (mesaj, kanal) => {
      const logSatiri = '[' + mesaj.zamanDamgasi + '] '
        + '[' + kanal + '] '
        + mesaj.tip + ' - '
        + JSON.stringify(mesaj.veri);
      console.log('LOG:', logSatiri);
      // Dosyaya veya log servisine yazılabilir
    });

    this.dinleyici.dinlemeyeBasla();
    console.log('Loglama servisi başlatıldı - tüm bildirimler izleniyor');
  }
}

// ==========================================
// Ana Uygulama
// ==========================================
async function uygulamayiBaslat() {
  const mesajYonetici = new MesajYonetici(yayinci, abone);
  const bildirimServisi = new BildirimServisi(mesajYonetici);
  const siparisServisi = new SiparisServisi(mesajYonetici);
  const loglamaServisi = new LoglamaServisi();

  // Servisleri başlat
  await bildirimServisi.basla();
  await loglamaServisi.basla();

  // Kısa bir gecikme sonrası test mesajları gönder
  setTimeout(async () => {
    // Yeni sipariş oluştur
    await siparisServisi.yeniSiparis({
      musteriAdi: 'Ahmet Yılmaz',
      toplamTutar: 2500,
      urunler: [
        { id: 1, isim: 'Mekanik Klavye', stok: 25, adet: 1 },
        { id: 2, isim: 'Oyuncu Faresi', stok: 8, adet: 2 }
      ]
    });

    // Kullanıcı bildirimi
    await mesajYonetici.yayinla('bildirim:kullanici', 'giris', {
      islem: 'Giriş yapıldı',
      kullaniciAdi: 'mehmet_dev',
      ipAdresi: '192.168.1.100'
    }, 'auth-servisi');
  }, 1000);
}

uygulamayiBaslat().catch(console.error);

Redis Pub/Sub, gerçek zamanlı iletişim gerektiren senaryolarda güçlü bir araçtır. Ancak mesaj kaybını göze alamayacağınız durumlarda Redis Streams veya özel mesaj kuyrukları kullanmanızı öneririm. Doğru aracı doğru yerde kullanmak, sağlam bir sistem tasarlamanın temelidir.

Best Studio · Çarşamba saat 16:29'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] Redis Pub/Sub ile Mesajlaşma

Silverhand