Yayın ve Kanal Yönetimi

Bir önceki ders çift yönlü bağlantıyı kurdu ve bir uyarıyla bitti: WebSocket bağlantısı bir sunucu örneğine bağlanır ve bağlantı süresince orada kalır. Aynı sonuç sunucu gönderimli olaylar için de geçerlidir; açık tutulan her akış belirli bir örneğin belleğinde durur.

Tek örnekli bir kurulumda bu görünmez. İki örneğe dağıtılmış bir serviste ise sonucu şudur: şube stok sayacını izleyen üç görevliden ikisi birinci örneğe, biri ikinciye bağlanmışsa, birinci örneğe gelen bir güncelleme üçüncü görevliye ulaşmaz. Bu ders o boşluğu ölçer ve kapatır.

Kanal

Önce adlandırma. Bir istemcinin ilgilendiği mesaj kümesine kanal (channel) denir; kütüphane servisinde sube-3, odunc-1204 ya da rapor-2026-01 gibi. Kanal, mesajlaşma konusunda tanıtılan kuyruktan iki noktada ayrılır.

Kanalın belleği yoktur: yayın anında dinlemeyen bir aboneye mesaj ulaşmaz ve bir yerde birikmez. Ve kanal çok abonelidir: aynı mesajın kopyası her aboneye gider, ilk alan mesajı tüketmiş olmaz. Yani kanal, mesajlaşma konusundaki konunun ve yayımla–abone ol modelinin bağlantı düzeyindeki karşılığıdır.

Bu iki özellik gerçek zamanlı veri için genellikle doğrudur — şube stok sayacının üç saniye önceki değeri kimseyi ilgilendirmez. Kalıcılık gerekiyorsa kanal yetmez; mesajın ayrıca dayanıklı bir kuyruğa ya da tabloya yazılması gerekir.

Uygulama Örneği

Aşağıdaki örnek iki uç sunuyor. /abone istemciye bir olay akışı açar ve onu kanal kümesine ekler; /yayinla bir kanala mesaj basar. İki kipte çalışıyor: yayını yalnız kendi abonelerine veren --yerel kipi ve yayını aktarıcıya gönderen --aktarici kipi.

// ornek.mjs — uygulama ornegi: istemcilere kanal aboneligi verir, yayin uretir
//   node ornek.mjs <nokta> --yerel      yayin yalniz kendi abonelerine gider
//   node ornek.mjs <nokta> --aktarici   yayin aktariciya gider, aktarici herkese dagitir
import { createServer } from "node:http";

const NOKTA = Number(process.argv[2] ?? 8392);   // varsayilan: birinci ornek
const AKTARICILI = process.argv.includes("--aktarici");
const AKTARICI = "http://127.0.0.1:8391";

const kanallar = new Map();                        // kanal -> Set<istemci yaniti>
const sayac = { abonelik: 0, yerel_teslim: 0, aktaricidan: 0 };

const govdeOku = (istek) => new Promise((c) => {
  let v = ""; istek.on("data", (p) => (v += p)); istek.on("end", () => c(v));
});

function yerelDagit(kanal, govde) {
  for (const y of kanallar.get(kanal) ?? []) {
    y.write(`data: ${govde}\n\n`);
    sayac.yerel_teslim++;
  }
}

// Aktariciyi dinleyen taraf: gelen her yayini kendi yerel abonelerine dagitir.
async function aktariciyiDinle() {
  const yanit = await fetch(`${AKTARICI}/dinle`);
  let tampon = "";
  for await (const parca of yanit.body) {
    tampon += Buffer.from(parca).toString();
    let k;
    while ((k = tampon.indexOf("\n\n")) >= 0) {
      const blok = tampon.slice(0, k); tampon = tampon.slice(k + 2);
      if (!blok.startsWith("data: ")) continue;    // yorum satiri: yok say
      const ileti = JSON.parse(blok.slice(6));
      sayac.aktaricidan++;
      yerelDagit(ileti.kanal, ileti.veri);
    }
  }
}

createServer(async (istek, yanit) => {
  yanit.sendDate = false;
  const adres = new URL(istek.url, "http://yerel");
  const kanal = adres.searchParams.get("kanal");

  if (adres.pathname === "/abone") {
    yanit.writeHead(200, { "content-type": "text/event-stream; charset=utf-8",
                           "cache-control": "no-store" });
    yanit.write(`: abone olundu kanal=${kanal} ornek=${NOKTA}\n\n`);
    if (!kanallar.has(kanal)) kanallar.set(kanal, new Set());
    kanallar.get(kanal).add(yanit);
    sayac.abonelik++;
    return void istek.on("close", () => kanallar.get(kanal).delete(yanit));
  }

  if (adres.pathname === "/yayinla") {
    const veri = await govdeOku(istek);
    if (AKTARICILI) await fetch(`${AKTARICI}/yayin`, {
      method: "POST", body: JSON.stringify({ kanal, veri }),
    });
    else yerelDagit(kanal, veri);
    return yanit.writeHead(202).end();
  }

  if (adres.pathname === "/olcum") {
    yanit.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    return yanit.end(JSON.stringify({ ornek: NOKTA, ...sayac }) + "\n");
  }
  yanit.writeHead(404).end();
}).listen(NOKTA, "127.0.0.1", () => {
  if (AKTARICILI) aktariciyiDinle();
  console.log(`ornek: 127.0.0.1:${NOKTA}`);
});

Aboneliğin bağlantı kapandığında silinmesi kritik bir ayrıntıdır. istek.on("close") işleyicisi olmadan kanal kümesi kapanmış yanıt nesnelerini tutmaya devam eder; bunlara yazma denemesi hata üretir ve küme sınırsız büyür. Uzun ömürlü bağlantı tutan her yapıda en sık görülen sızıntı budur.

Örnekler Arası Yayın Kanalı

Örneğin --aktarici kipinde konuştuğu yapı, tek işi mesajı bütün örneklere ulaştırmak olan bir ara katmandır. Her örnek ona bir akışla bağlanır; bir örnekten gelen yayın, bağlı olan tüm örneklere yazılır — yayını gönderen örnek de dâhil.

// aktarici.mjs — ornekler arasi yayin kanali; her uygulama ornegine bir akis acik tutar
import { createServer } from "node:http";

const ornekler = new Set();                        // aktariciyi dinleyen ornek yanitlari
const sayac = { yayin: 0, dagitim: 0 };

const govdeOku = (istek) => new Promise((c) => {
  let v = ""; istek.on("data", (p) => (v += p)); istek.on("end", () => c(v));
});

createServer(async (istek, yanit) => {
  yanit.sendDate = false;
  const adres = new URL(istek.url, "http://yerel");

  if (adres.pathname === "/dinle") {                // ornek aktariciya baglanir
    yanit.writeHead(200, { "content-type": "text/event-stream; charset=utf-8",
                           "cache-control": "no-store" });
    yanit.write(": aktariciya baglanildi\n\n");
    ornekler.add(yanit);
    return void istek.on("close", () => ornekler.delete(yanit));
  }

  if (adres.pathname === "/yayin") {                // bir ornekten gelen yayin: herkese
    const govde = await govdeOku(istek);
    sayac.yayin++;
    for (const o of ornekler) { o.write(`data: ${govde}\n\n`); sayac.dagitim++; }
    return yanit.writeHead(202).end();
  }

  if (adres.pathname === "/olcum") {
    yanit.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    return yanit.end(JSON.stringify({ ...sayac, ornek: ornekler.size }) + "\n");
  }
  yanit.writeHead(404).end();
}).listen(8391, "127.0.0.1", () => console.log("aktarici: 127.0.0.1:8391"));

Yayını gönderen örneğin de aktarıcıdan geri alması bir kusur değil, sadeleştirmedir: yerel dağıtım tek bir yerde, aktarıcıdan gelen yolda yapılır. Aksi hâlde mesaj kendi örneğinde iki kez teslim edilir.

Ölçüm

Aşağıdaki betik iki uygulama örneği ile bir aktarıcı çalıştırıyor. Abone ikinci örneğe bağlanıyor, yayın ise birinci örneğe geliyor. Bağlantı noktaları 8391–8393 keyfîdir ve boş olmalıdır.

#!/usr/bin/env bash
# Iki uygulama ornegi + bir aktarici; abone ikinci ornekte, yayin birinci ornege gelir.
olc() {
  node aktarici.mjs > /dev/null & a=$!
  node ornek.mjs 8392 "$1" > /dev/null & o1=$!
  node ornek.mjs 8393 "$1" > /dev/null & o2=$!
  sleep 1

  curl -sN --max-time 1.2 "http://127.0.0.1:8393/abone?kanal=sube-3" > abone.txt &
  dinleyen=$!
  sleep 0.4
  curl -sS -o /dev/null -X POST -d '{"sube":3,"stok":41}' \
       "http://127.0.0.1:8392/yayinla?kanal=sube-3"
  wait $dinleyen

  echo "--- $2 ---"
  echo "ikinci ornekteki abonenin gordugu:"
  sed '/^$/d; s/^/  /' abone.txt
  echo "sayaclar:"
  curl -sS http://127.0.0.1:8392/olcum | sed 's/^/  /'
  curl -sS http://127.0.0.1:8393/olcum | sed 's/^/  /'
  curl -sS http://127.0.0.1:8391/olcum | sed 's/^/  aktarici /'
  kill $a $o1 $o2 2>/dev/null
  sleep 0.5
}

olc --yerel    "aktarici yok"
olc --aktarici "aktarici var"

--- aktarici yok ---
ikinci ornekteki abonenin gordugu:
  : abone olundu kanal=sube-3 ornek=8393
sayaclar:
  {"ornek":8392,"abonelik":0,"yerel_teslim":0,"aktaricidan":0}
  {"ornek":8393,"abonelik":1,"yerel_teslim":0,"aktaricidan":0}
  aktarici {"yayin":0,"dagitim":0,"ornek":0}
--- aktarici var ---
ikinci ornekteki abonenin gordugu:
  : abone olundu kanal=sube-3 ornek=8393
  data: {"sube":3,"stok":41}
sayaclar:
  {"ornek":8392,"abonelik":0,"yerel_teslim":0,"aktaricidan":1}
  {"ornek":8393,"abonelik":1,"yerel_teslim":1,"aktaricidan":1}
  aktarici {"yayin":1,"dagitim":2,"ornek":2}

Birinci koşumda abone yalnız abonelik onayını gördü; yayın hiçbir yere gitmedi. Birinci örneğin yerel_teslim sayacı sıfır, çünkü o örnekte abone yok. Mesaj kaybolmadı, hata da üretmedi — sessizce hiçbir şey olmadı. Çok örnekli kurulumlarda bu kusurun tehlikeli yanı budur: tek örnekle geliştirilirken hiçbir belirti vermez.

İkinci koşumda abone mesajı aldı. Sayaçlar yolu adım adım gösteriyor: aktarıcıya bir yayın geldi, iki örneğe dağıtıldı, her iki örnek de birer mesaj aldı, ama yalnız aboneye sahip olan örnek yerel teslim yaptı.

Yayma Maliyeti

Aktarıcının dagitim sayacı yayının değil, kopyaların sayısıdır. Bir yayın, örnek sayısı kadar kopyaya dönüşür. $N$ örnek ve saniyede $M$ yayın için aktarıcının yazdığı mesaj sayısı $N \times M$’dir ve bu, abonesi olmayan örnekler için tamamen boşa giden bir iştir. Ölçümde birinci örnek mesajı aldı ve hiçbir yere teslim etmedi.

İki azaltma yolu vardır. Abonelik kaydı tutmakta aktarıcı her örneğin hangi kanalları dinlediğini bilir ve yayını yalnız ilgili örneklere gönderir. Kazanç, kanal sayısı örnek sayısından büyük olduğunda belirgindir; bedeli, her abone açılıp kapandığında aktarıcıya bir kayıt mesajı gitmesidir. Kanal bölümlemede ise kanallar örneklere önceden paylaştırılır ve istemci ilgilendiği kanalı barındıran örneğe yönlendirilir; dağıtım tamamen ortadan kalkar ama esneklik de kalkar, çünkü bir istemci birden çok kanalı dinliyorsa birden çok bağlantı açması gerekir.

Ölçek büyüdükçe üçüncü bir sınır belirir: aktarıcının kendisi tek bir noktadır. Gerçek kurulumlarda bu iş, bu kursun mesajlaşma konusunda tanıtılan yayımla–abone ol altyapısına devredilir; aktarıcı o altyapının önündeki ince bir uyarlama katmanına dönüşür.

Yapışkan Yönlendirme Yeterli mi

Sık başvurulan bir kestirme, aynı istemcinin hep aynı örneğe yönlendirilmesidir; buna yapışkan yönlendirme (sticky routing) denir. Bağlantı koptuğunda istemcinin aynı örneğe dönmesini sağladığı için yararlıdır, ama buradaki sorunu çözmez: aynı kanalı dinleyen iki farklı istemci yine farklı örneklere düşebilir.

Yapışkanlık ancak kanal tabanlı olduğunda çözüm olur — yani yönlendirme kararı istemciye değil kanala bakarak verildiğinde. Bu da yukarıdaki kanal bölümleme yaklaşımının ta kendisidir ve yük dağıtıcının uygulama kavramlarını bilmesini gerektirir.

Özet

• Kanal, istemcinin ilgilendiği mesaj kümesidir; belleği yoktur ve çok abonelidir, bu yüzden kalıcılık gerektiren veriler ayrıca yazılmalıdır.
• Açık bağlantı belirli bir örneğin belleğinde durduğu için, yerel dağıtım çok örnekli kurulumda mesajı diğer örneklerin abonelerine ulaştırmaz ve bu kusur hiçbir hata üretmez.
• Örnekler arası bir yayın kanalı boşluğu kapatır; yayını gönderen örneğin de mesajı geri alması, yerel dağıtımı tek bir yolda toplar.
• Abonelik bağlantı kapandığında silinmezse kanal kümesi kapanmış yanıtları tutar ve sınırsız büyür.
• Yayma maliyeti örnek sayısıyla çarpılır; abonelik kaydı ya da kanal bölümleme bunu azaltır, yapışkan yönlendirme ise tek başına çözmez.

Sonraki Adım

Bu konudaki dört yaklaşım da sunucudan istemciye olay taşıdı: durum değişti, sıra ilerledi, stok güncellendi. Geriye tek bir durum kalıyor. Bazen taşınacak şey bir olay değil, yanıtın kendisidir — otuz bin ödünç kaydını tarayan raporun çıktısı, üretildiği hızla istemciye akıtılabilir. Bu, yeni bir protokol gerektirmez; HTTP yanıt gövdesinin parça parça yazılmasıyla yapılır. Sonraki ders bu son biçimi kurar ve kursu kapatır.