Mesaj Kuyrukları ve Akışlar

Önceki dersin veriyolunun ölçülmemiş bir özelliği vardı: yayimla çağrısı, bağlı bütün işleyiciler bitmeden geri dönmüyordu. Dört işleyici aynı çağrı yığıtında sırayla koşuyor, dolayısıyla ücreti kesinleştiren modülün işi en yavaş işleyicinin işi kadar sürüyordu. Arşiv yazımı yavaşlarsa ücret kesinleştirme de yavaşlar; oysa arşivin ne zaman yazıldığı ücretlendirme için önemli değildir. Bu ders araya bir depo koyar: yayımcı iletiyi bırakır ve döner, tüketici onu sonra alır.

Mesaj kuyruğu (message queue) ile akış (stream) Önbellekleme, Kuyruklar ve Eşzamansız İşleme kursunun Mesajlaşma konusunda kurulmuş ve ölçülmüştü: teslim anlamları, sıralama ile bölümleme arasındaki ödünleşim, ölü mektup kuyruğu ve yeniden deneme oraya aittir ve burada yeniden ele alınmaz. Buraya düşen kısım biçemin kendisidir — deponun neyi garanti ettiği, neyi etmediği ve bunun mimariye getirdiği kısıt. Ölçülecek dört şey vardır: bırakma anında koşmuş tüketici işi, bekleyen ileti sayısı, sıranın korunup korunmadığı ve aynı iletinin iki kez işlenmesinin sonucu.

Bağlandığı kalite niteliği güvenilirliktir ve nitelik sorusu şudur: tüketici üreticiden yavaş kaldığında ya da bir ileti ikinci kez teslim edildiğinde kaç iş birikir ve kaç teslimat kaydı iki kez açılır. İkinci bir ölçü başarım verimliliği ailesine girer: yayımcı iletiyi bıraktığında kaç tüketici iş birimi koşmuş olur.

İki Uç ve Bir Depo

Tüketici teslimat operasyonu bağlamındadır: gelen her ileti için teslimat kaydı açar. Her iletinin üç iş birimi tuttuğu varsayılır. Bu bir modeldir, ölçüm değildir; sayılan şey iş birimi adedidir, hiçbir süre ölçülmemiştir. anahtarli seçeneği aynı gönderi kodunu ikinci kez gördüğünde kayıt açmaz; dersin sonunda bu seçeneğin ölçüsü alınır.

mkdir -p esz

// esz/teslimat.mjs — tuketici: her ileti icin teslimat kaydi acar ve 3 is birimi harcar
export const IS_BIRIMI = 3;

export function teslimatci(anahtarli = false) {
  const kayit = [], gorulen = new Set();
  let birim = 0;
  return {
    isle(ileti) {
      birim += IS_BIRIMI;
      if (anahtarli && gorulen.has(ileti.kod)) return;
      gorulen.add(ileti.kod);
      kayit.push(ileti.kod);
    },
    birim: () => birim,
    kayit: () => kayit,
  };
}

Depo iki işlem açar. koy iletiyi sona ekler ve hiçbir tüketici çağırmaz; al en eski iletiyi verir. Bu iki imzanın arasındaki boşluk biçemin tamamıdır: koyan ile alan aynı anda çalışmak zorunda değildir.

// esz/kuyruk.mjs — mesaj kuyrugu: koyan beklemez, alan sirayla alir
export function kuyruk() {
  const bekleyen = [];
  let konan = 0, alinan = 0, tepe = 0;
  return {
    koy(ileti) { bekleyen.push(ileti); konan += 1; tepe = Math.max(tepe, bekleyen.length); },
    al() { if (bekleyen.length === 0) return null; alinan += 1; return bekleyen.shift(); },
    uzunluk: () => bekleyen.length,
    sayac: () => ({ konan, alinan, tepe }),
  };
}

Yayımcı Beklemiyor

İlk betik yirmi iletiyi iki düzende geçirir. Eşzamanlı düzende yayımcı tüketiciyi doğrudan çağırır; kuyruk düzeninde iletiyi bırakır. Bırakma bittiği anda koşmuş tüketici işi sayılır, sonra kuyruk boşaltılır ve iki düzenin aynı kaydı ürettiği doğrulanır. İkinci bölümde üretici tur başına beş, tüketici tur başına üç ileti işler ve üretim dördüncü turdan sonra durur.

// bloke.mjs — yayimci olayi biraktiginda kosmus tuketici is birimi ve kuyrukta bekleyen ileti
import { teslimatci, IS_BIRIMI } from "./esz/teslimat.mjs";
import { kuyruk } from "./esz/kuyruk.mjs";

const ILETI = Array.from({ length: 20 }, (_, i) => ({ kod: `G-${String(i + 1).padStart(2, "0")}` }));

const hemen = teslimatci();
for (const ileti of ILETI) hemen.isle(ileti);

const k = kuyruk(), sonra = teslimatci();
for (const ileti of ILETI) k.koy(ileti);
const birakmaAninda = { birim: sonra.birim(), bekleyen: k.uzunluk() };
let alinan;
while ((alinan = k.al()) !== null) sonra.isle(alinan);

const GENISLIK = [15, 25, 16, 18];
const SATIR = [
  ["eszamanli", ILETI.length, hemen.birim(), 0, hemen.birim()],
  ["kuyruk", ILETI.length, birakmaAninda.birim, birakmaAninda.bekleyen, sonra.birim()],
];
console.log("duzen      yayimci cagrisi  birakma aninda kosan is  bekleyen ileti  sonunda kosan is");
for (const [ad, ...s] of SATIR) console.log(ad.padEnd(11) + s.map((v, i) => String(v).padStart(GENISLIK[i])).join(""));
console.log(`kuyruk sayaci: ${JSON.stringify(k.sayac())}, kayit esit mi = ${JSON.stringify(hemen.kayit()) === JSON.stringify(sonra.kayit())}`);

const URETIM = [5, 5, 5, 5, 0, 0, 0, 0];
const TUKETIM = 3;
const k2 = kuyruk(), t2 = teslimatci();
let sira = 0;
console.log("\ntur  uretilen  tuketilen  bekleyen");
for (const [i, uret] of URETIM.entries()) {
  for (let n = 0; n < uret; n += 1) { sira += 1; k2.koy({ kod: `H-${sira}` }); }
  let tuketilen = 0;
  for (let n = 0; n < TUKETIM; n += 1) {
    const ileti = k2.al();
    if (ileti === null) break;
    t2.isle(ileti);
    tuketilen += 1;
  }
  console.log(`${String(i + 1).padStart(3)}${String(uret).padStart(10)}${String(tuketilen).padStart(11)}${String(k2.uzunluk()).padStart(10)}`);
}
console.log(`tepe bekleyen = ${k2.sayac().tepe}, is birimi = ${t2.birim()}, ileti basina is birimi = ${IS_BIRIMI}`);

node bloke.mjs

duzen      yayimci cagrisi  birakma aninda kosan is  bekleyen ileti  sonunda kosan is
eszamanli               20                       60               0                60
kuyruk                  20                        0              20                60
kuyruk sayaci: {"konan":20,"alinan":20,"tepe":20}, kayit esit mi = true

tur  uretilen  tuketilen  bekleyen
  1         5          3         2
  2         5          3         4
  3         5          3         6
  4         5          3         8
  5         0          3         5
  6         0          3         2
  7         0          2         0
  8         0          0         0
tepe bekleyen = 11, is birimi = 60, ileti basina is birimi = 3

Sayıların Okunması

İlk tablonun son sütunu iki düzende de 60’tır: iş yok olmadı, aynı iş yapıldı ve iki düzen aynı kaydı üretti. Değişen şey işin ne zaman yapıldığıdır. Eşzamanlı düzende yayımcı yirminci çağrıyı bitirdiğinde altmış iş birimi onun zamanından gitmişti; kuyruk düzeninde aynı anda sıfır iş birimi koşmuştu ve depoda yirmi ileti bekliyordu. Bekleyen ileti sayısı, bedelin adıdır: yayımcı beklemekten kurtuldu, ama sistemde artık işlenmemiş iş duruyor ve bu iş bir yerde tutulmak zorunda.

İkinci tablo bu birikimin nasıl davrandığını gösterir. Üretim tüketimden hızlıyken bekleyen ileti her turda ikişer artıyor; dördüncü turun üretiminden sonra kuyruk en yüksek değerine, on bire ulaşıyor. Üretim durduğunda kuyruk erimeye başlıyor ve üç tur sonra boşalıyor. Bu, biçemin temel kısıtıdır: kuyruk hız farkını saklar, ortadan kaldırmaz. Üretim hızı tüketim hızının üstünde kalıcı olarak durursa bekleyen ileti sayısı sınırsız büyür; kuyruğun sağladığı şey farkın geçici olduğu durumda yayımcının beklememesidir.

Sıra ve Tekrar

İkinci betik iki soru sorar. Birincisi sıradır: kuyruğa konma sırası ile işlenme sırası aynı mı? Tek tüketicide ve aynı kuyruğun ucunda çalışan iki tüketicide ayrı ayrı ölçülür. İki tüketicili düzende işçilerin iş birimi maliyeti farklıdır ve boşalan işçi sıradaki iletiyi alır; bitiş sırası buna göre kurulur. Bu bir modeldir, ölçüm değildir — sayılan şey iş birimi ve bitiş sırasıdır, hiçbir süre ölçülmemiştir. İkinci soru tekrardır: bir ileti onaylanmadığı için kuyruğa geri konursa ne olur?

// sira-tekrar.mjs — tek ve iki tuketicide sira, bir ileti yeniden teslim edildiginde acilan kayit
import { teslimatci } from "./esz/teslimat.mjs";
import { kuyruk } from "./esz/kuyruk.mjs";

const ILETI = Array.from({ length: 8 }, (_, i) => ({ kod: `G-${i + 1}` }));
const SIRA = ILETI.map((i) => i.kod);

const k = kuyruk(), tek = teslimatci();
for (const ileti of ILETI) k.koy(ileti);
let alinan;
while ((alinan = k.al()) !== null) tek.isle(alinan);

const tersine = (dizi) => {
  const yer = new Map(SIRA.map((kod, i) => [kod, i]));
  let t = 0;
  for (let a = 0; a < dizi.length; a += 1)
    for (let b = a + 1; b < dizi.length; b += 1) if (yer.get(dizi[a]) > yer.get(dizi[b])) t += 1;
  return t;
};

const MALIYET = [1, 4];
const bos = MALIYET.map(() => 0);
const bitis = [];
const k2 = kuyruk();
for (const ileti of ILETI) k2.koy(ileti);
while ((alinan = k2.al()) !== null) {
  const j = bos[0] <= bos[1] ? 0 : 1;
  bos[j] += MALIYET[j];
  bitis.push([alinan.kod, bos[j]]);
}
const bitisSirasi = [...bitis].sort((a, b) => a[1] - b[1]).map(([kod]) => kod);

console.log(`koyulan sira : ${SIRA.join(" ")}`);
console.log(`tek tuketici : ${tek.kayit().join(" ")}  tersine cift = ${tersine(tek.kayit())}`);
console.log(`iki tuketici : ${bitisSirasi.join(" ")}  tersine cift = ${tersine(bitisSirasi)}`);

for (const [ad, anahtarli] of [["anahtarsiz", false], ["anahtarli", true]]) {
  const k3 = kuyruk(), t3 = teslimatci(anahtarli);
  for (const ileti of ILETI) k3.koy(ileti);
  let sayi = 0;
  while ((alinan = k3.al()) !== null) {
    t3.isle(alinan);
    sayi += 1;
    if (sayi === 3) k3.koy(alinan);
  }
  const s = k3.sayac();
  console.log(`${ad.padEnd(11)} teslim = ${s.alinan}, farkli ileti = ${new Set(ILETI.map((i) => i.kod)).size}, acilan teslimat kaydi = ${t3.kayit().length}, is birimi = ${t3.birim()}`);
}

node sira-tekrar.mjs

koyulan sira : G-1 G-2 G-3 G-4 G-5 G-6 G-7 G-8
tek tuketici : G-1 G-2 G-3 G-4 G-5 G-6 G-7 G-8  tersine cift = 0
iki tuketici : G-1 G-3 G-4 G-2 G-5 G-6 G-8 G-7  tersine cift = 3
anahtarsiz  teslim = 9, farkli ileti = 8, acilan teslimat kaydi = 9, is birimi = 27
anahtarli   teslim = 9, farkli ileti = 8, acilan teslimat kaydi = 8, is birimi = 27

Tek tüketicide tersine çift sayısı sıfırdır: sıra korunmuştur ve bu, deponun al imzasının sonucudur. İkinci tüketici eklendiğinde aynı kuyruk, aynı sırayla dağıtım yapmasına karşın bitiş sırası üç yerde bozuldu. Sıranın korunması tek tüketiciye bağlı bir özelliktir; ikinci tüketici eklemek bir başarım kararı değil, sıra garantisinden vazgeçme kararıdır. Mimari sonucu şudur: sırası önemli olan işler ile paralel işlenebilen işler aynı kuyruktan geçemez.

Son iki satır tekrarın faturasıdır. Sekiz farklı ileti dokuz kez teslim edildi. Anahtarsız tüketicide açılan teslimat kaydı dokuz oldu: aynı gönderi için iki kayıt açıldı. Anahtarlı tüketicide sekiz kaldı, yani ikinci teslim etkisiz geçti — ama iş birimi iki düzende de 27’dir, çünkü ileti yine alındı ve yine işlendi. Buradaki ölçüm biçemin kısıtını verir: depo iletiyi sıraya koyar, tekliği garanti etmez; teklik tüketicinin kendi anahtarıyla sağlanmak zorundadır. Aynı iletinin iki kez ulaşmasına karşı iki tüketici tasarımının nasıl saptığı bu kursun Olay Güdümlü Mimari dersinde ayrıca ölçülür.

Tüketmek ile Silmek

Kuyruğun al işlemi iletiyi depodan çıkarır: tüketmek silmektir. Akış bu iki işi ayırır — kayıt yerinde kalır, her tüketici kendi konumunu tutar. Konum, saklama süresi ve yeniden okuma Mesajlaşma konusundaki Akışlar ve Kuyruk Farkı dersinde ölçülmüştü. Buradaki soru mimaridir: sonradan eklenen bir birim geçmişi kendi başına kurabilir mi?

// esz/gunluk.mjs — akis: tuketmek silmez, her tuketici kendi konumunu tutar
export function gunluk() {
  const kayit = [];
  return { ekle: (ileti) => kayit.push(ileti), oku: (konum) => kayit.slice(konum), uzunluk: () => kayit.length };
}

// gecmis.mjs — kuyruk tuketilince bosalir, akis konumla okunur: sonradan eklenen birimin gecmisi
import { kuyruk } from "./esz/kuyruk.mjs";
import { gunluk } from "./esz/gunluk.mjs";
import { teslimatci } from "./esz/teslimat.mjs";

const ILETI = Array.from({ length: 20 }, (_, i) => ({ kod: `G-${i + 1}` }));

const k = kuyruk(), kuyrukTuketicisi = teslimatci();
for (const ileti of ILETI) k.koy(ileti);
let alinan;
while ((alinan = k.al()) !== null) kuyrukTuketicisi.isle(alinan);

const g = gunluk(), akisTuketicisi = teslimatci();
for (const ileti of ILETI) g.ekle(ileti);
let konum = 0;
for (const ileti of g.oku(konum)) { akisTuketicisi.isle(ileti); konum += 1; }
const konumOnce = konum;

const sonradan = teslimatci();
for (const ileti of g.oku(0)) sonradan.isle(ileti);

console.log(`kuyruk: konan ${k.sayac().konan}, tuketildikten sonra bekleyen ${k.uzunluk()}`);
console.log(`  sonradan eklenen birimin kuyruktan kendi basina okudugu gecmis = ${k.uzunluk()} kayit`);
console.log(`  gecmis ancak tuketicinin kaydi disa acilarak alinir: ${kuyrukTuketicisi.kayit().length} kayit, baska tuketiciye kurulan bag = 1`);
console.log(`akis: gunluk uzunlugu ${g.uzunluk()}, ilk tuketicinin konumu ${konumOnce}`);
console.log(`  sonradan eklenen birimin konum 0'dan okudugu gecmis = ${sonradan.kayit().length} kayit, baska tuketiciye kurulan bag = 0`);
console.log(`  ilk tuketicinin konumu ${konumOnce} -> ${konum}, gunluk uzunlugu ${g.uzunluk()}`);

node gecmis.mjs

kuyruk: konan 20, tuketildikten sonra bekleyen 0
  sonradan eklenen birimin kuyruktan kendi basina okudugu gecmis = 0 kayit
  gecmis ancak tuketicinin kaydi disa acilarak alinir: 20 kayit, baska tuketiciye kurulan bag = 1
akis: gunluk uzunlugu 20, ilk tuketicinin konumu 20
  sonradan eklenen birimin konum 0'dan okudugu gecmis = 20 kayit, baska tuketiciye kurulan bag = 0
  ilk tuketicinin konumu 20 -> 20, gunluk uzunlugu 20

Kuyruk boşaldıktan sonra sonradan eklenen birimin kendi başına okuyabildiği geçmiş sıfır kayıttır. Aynı bilgiyi almanın tek yolu var olan tüketicinin kaydını dışa açması, yani iki tüketici arasında yeni bir bağ kurulmasıdır: yeni birim artık yayımcıya değil, bir başka tüketiciye bağlıdır. Akışta aynı sayı yirmidir ve kurulan bağ sıfır; yeni birim yalnız günlüğü tanır. İkinci okuma ilk tüketicinin konumunu değiştirmedi ve günlüğün uzunluğu artmadı — tüketici sayısı bilme yükümlülüğünü büyütmüyor.

Seçim ölçütü budur: iş bir kez yapılıp bitecekse kuyruk yeterlidir; sonradan eklenen birimlerin geçmişten kendi durumunu kurması gerekiyorsa akış gerekir ve bedeli kaydın saklanmasıdır.

Özet

• Depo yayımcı ile tüketiciyi ayırır: yirmi iletiyi bırakan yayımcı eşzamanlı düzende 60 iş birimini kendi zamanından harcadı, kuyruk düzeninde 0 harcadı ve depoda 20 ileti bekledi.
• Toplam iş iki düzende de 60’tır ve iki düzen aynı kaydı üretti; kuyruk işi azaltmaz, zamanını değiştirir.
• Kuyruk hız farkını saklar: tur başına 5 üretim ile 3 tüketimde bekleyen ileti tepe değeri 11 oldu ve üretim durduktan üç tur sonra kuyruk boşaldı.
• Sıra tek tüketicide korundu (tersine çift 0), ikinci tüketici eklendiğinde bitiş sırası üç yerde bozuldu; ikinci tüketici eklemek sıra garantisinden vazgeçmektir.
• Sekiz ileti dokuz kez teslim edildiğinde anahtarsız tüketici 9, anahtarlı tüketici 8 teslimat kaydı açtı; iş birimi ikisinde de 27’dir — depo sırayı garanti eder, tekliği etmez.
• Kuyrukta tüketmek silmektir: sonradan eklenen birimin okuyabildiği geçmiş 0 kayıt, akışta 20 kayıttır ve akışta başka bir tüketiciye bağ kurulmaz.

Sonraki Adım

Bu konu birimlerin birbiriyle nasıl konuştuğunu ele aldı. Altı biçemde aynı üç şey ölçüldü: birimin karşı tarafı bilme yükümlülüğü, bağlanma yönü ve sınırdan geçen veri. İstemci ile sunucu, eşler, model ile görünüm, boru ile süzgeç, yayımcı ile abone, kuyruğun iki ucu — hepsi bu üç sayıyla karşılaştırıldı. Ancak bütün bu ölçümler bir soruyu sormadan geçti: bu birimler kaç ayrı parça olarak yayınlanıyor? Bu dersteki kuyruk da, önceki dersteki veriyolu da tek bir çalışan programın içindeydi; yayımla–abone ol düzeni için ikinci bir süreç gerekmedi, bir nesne yetti. Etkileşim biçemi ile parça sayısı birbirinden bağımsız iki karardır. Sonraki konu ikincisini sorar: yazılım kaç ayrı dağıtım biriminde yayınlanır ve bu sayı neyi değiştirir.