Akışlar ve Kuyruk Farkı

Önceki dersin son ölçümü bir boşluk bıraktı: sonradan bağlanan arama dizini beş ödünç olayının üçünü hiç göremedi, çünkü kopya yalnız yayım anında var olan aboneliklere düşürülür. Aynı boşluğun ikinci yüzü de vardı — işlenmiş kopya silindiği için hiçbir tüketici geriye dönüp yeniden okuyamıyordu.

İki eksiğin de kaynağı tek bir varsayımdır: teslim edilen ileti tüketilmiştir, tüketilen ileti silinir. Bu ders o varsayımı bırakan ikinci taşıma modelini kurar ve iki modeli aynı iş yükü üzerinde karşılaştırır.

Ekleme ve Konum

Akış (stream), iletilerin yalnız sona eklendiği ve okunduğunda yerinde kaldığı sıralı bir kayıttır. Yapının kendisi yeni değil: Veritabanı Yönetimi kursunda yazma öncesi günlük aynı kuralla çalışıyordu — kayıtlar sona eklenir, hiçbir kayıt yerinde değiştirilmez. Akış aynı yapıyı iletileri taşımak için kullanır.

Kuyruktan iki noktada ayrılır. Birincisi, okuma silmez: oku çağrısı günlüğü değiştirmez. İkincisi, tüketicinin nereye kadar geldiği kuyrukta değil tüketicide durur. Bu değere konum (offset) denir ve günlükteki bir kaydın sırasını gösterir.

// gunluk.mjs — yalniz sona ekleme yapilan gunluk: okuma kaydi silmez, konumu tuketici tutar
import { appendFileSync, readFileSync, writeFileSync, existsSync } from "node:fs";

export function sifirla(dosya) { writeFileSync(dosya, ""); }

export function ekle(dosya, kayit) {              // yazma yalniz sona olur
  appendFileSync(dosya, JSON.stringify(kayit) + "\n");
}

export function kayitlar(dosya) {
  if (!existsSync(dosya)) return [];
  const t = readFileSync(dosya, "utf8").trimEnd();
  return t === "" ? [] : t.split("\n").map((s) => JSON.parse(s));
}

// konumdan baslayarak en cok `adet` kayit dondurur; okuma gunlugu degistirmez
export function oku(dosya, konum, adet) {
  return kayitlar(dosya).slice(konum, konum + adet);
}

ekle işlevinin tek yazma biçimi appendFileSync olması bir kolaylık değil, modelin kuralıdır. Sona ekleme, dosyanın önceki baytlarına dokunmaz; bu yüzden aynı anda okuyan bir tüketicinin altındaki veri hiç değişmez ve okuma ile yazma birbirini kilitlemez.

Bir Kopya, Üç Konum

Önceki derste yayma çarpanını ölçmüştük: abone sayısı arttıkça düşürülen kopya sayısı da çarpılıyordu. Akışta kopya sayısı birdir; çarpılan şey kopya değil, tutulan konum değeridir. Aşağıdaki koşum aynı altı olayı iki biçimde saklar ve ikisinin diskte kapladığı yeri karşılaştırır.

// uc-tuketici.mjs — tek gunluk, tuketici basina konum; kopya yerine konum tutulur
import { statSync } from "node:fs";
import { sifirla, ekle, kayitlar, oku } from "./gunluk.mjs";

const OLAYLAR = [
  { tur: "odunc_verildi", oduncId: 1, kitapId: 1, uyeId: 1 },
  { tur: "odunc_verildi", oduncId: 2, kitapId: 2, uyeId: 2 },
  { tur: "iade_alindi", oduncId: 1, kitapId: 1, uyeId: 1 },
  { tur: "odunc_verildi", oduncId: 3, kitapId: 3, uyeId: 3 },
  { tur: "rezervasyon_acildi", kitapId: 1, uyeId: 2 },
  { tur: "iade_alindi", oduncId: 2, kitapId: 2, uyeId: 2 },
];

sifirla("olaylar.log");
for (const o of OLAYLAR) ekle("olaylar.log", o);

const ADLAR = ["bildirim", "stok-ozeti", "rapor"];
for (const ad of ADLAR) {                       // abonelik basina kopya tutan olcut
  sifirla(`kopya-${ad}.log`);
  for (const o of OLAYLAR) ekle(`kopya-${ad}.log`, o);
}
const kopyaBayt = ADLAR.reduce((t, a) => t + statSync(`kopya-${a}.log`).size, 0);
console.log(`tek gunluk       : ${kayitlar("olaylar.log").length} kayit  ${statSync("olaylar.log").size} bayt`);
console.log(`abonelik kopyasi : ${ADLAR.length} dosya  ${kopyaBayt} bayt`);

const konum = { bildirim: 0, "stok-ozeti": 0, rapor: 0 };
const hiz = { bildirim: 6, "stok-ozeti": 4, rapor: 1 };   // turda okunabilen kayit sayisi
for (const ad of ADLAR) konum[ad] += oku("olaylar.log", konum[ad], hiz[ad]).length;
console.log(`bir turdan sonra konumlar: ${ADLAR.map((a) => `${a}=${konum[a]}`).join("  ")}`);
console.log(`gunluk uzunlugu okumadan sonra = ${kayitlar("olaylar.log").length}`);
console.log(`raporun geride kaldigi kayit sayisi = ${kayitlar("olaylar.log").length - konum.rapor}`);

node uc-tuketici.mjs

tek gunluk       : 6 kayit  337 bayt
abonelik kopyasi : 3 dosya  1011 bayt
bir turdan sonra konumlar: bildirim=6  stok-ozeti=4  rapor=1
gunluk uzunlugu okumadan sonra = 6
raporun geride kaldigi kayit sayisi = 5

Üç sayı belirleyici. Diskte tutulan veri 337 bayta karşı 1011 bayt: kopya tutan model, ilgili taraf sayısıyla doğru orantılı yer kaplarken günlük tek kopya taşıyor. Üç tüketici farklı konumlarda — 6, 4 ve 1 — yani birbirlerini beklemiyorlar; yalıtım kopyayla değil konumla sağlanıyor. Ve en önemlisi son iki satır: altı kayıt okumadan sonra hâlâ günlükte duruyor, raporun beş kayıt geride olduğu tek bir çıkarmayla hesaplanıyor. Bu fark, akışta doğrudan ölçülebilen bir gecikme ölçütüdür.

Günlük dosyası satır başına bir kayıt tutar ve içeriği okunabilir durumdadır.

head -3 olaylar.log

{"tur":"odunc_verildi","oduncId":1,"kitapId":1,"uyeId":1}
{"tur":"odunc_verildi","oduncId":2,"kitapId":2,"uyeId":2}
{"tur":"iade_alindi","oduncId":1,"kitapId":1,"uyeId":1}

Yeniden Okuma

İki modelin ayrıldığı yer burasıdır. Aşağıdaki koşum altı ödünç olayını önce onaylananı silen bir kuyruğa, sonra günlüğe koyar; her ikisinde de bir tüketici hepsini işler ve ardından ikinci bir tüketici geçmişi istemeye çalışır.

// yeniden-okuma.mjs — ayni is yuku once kuyrukta sonra gunlukte; gecmis yeniden okunmaya calisilir
import { sifirla, ekle, kayitlar, oku } from "./gunluk.mjs";

const OLAYLAR = [1, 2, 3, 4, 5, 6].map((n) => ({ tur: "odunc_verildi", oduncId: n }));

// Onaylanan iletiyi silen kuyruk: 02. derste kurulan davranisin ozu
class Kuyruk {
  constructor() { this.iletiler = []; }
  yaz(g) { this.iletiler.push(g); }
  al() { return this.iletiler.length === 0 ? null : this.iletiler[0]; }
  onayla() { this.iletiler.shift(); }            // onay: ileti kuyruktan silinir
  get derinlik() { return this.iletiler.length; }
}

const k = new Kuyruk();
for (const o of OLAYLAR) k.yaz(o);
let birinci = 0;
for (let i = k.al(); i !== null; i = k.al()) { birinci += 1; k.onayla(); }
let ikinci = 0;
for (let i = k.al(); i !== null; i = k.al()) { ikinci += 1; k.onayla(); }
console.log(`kuyruk: yazilan=${OLAYLAR.length} birinci tuketici=${birinci} ` +
            `derinlik=${k.derinlik} sonra baglanan tuketici=${ikinci}`);

sifirla("olaylar.log");
for (const o of OLAYLAR) ekle("olaylar.log", o);
let konum = 0;
for (const kayit of oku("olaylar.log", konum, 100)) konum += 1;
const sonraki = oku("olaylar.log", 0, 100).length;      // konumu sifira alan ikinci tuketici
console.log(`akis  : yazilan=${OLAYLAR.length} birinci tuketici=${konum} ` +
            `gunluk uzunlugu=${kayitlar("olaylar.log").length} sonra baglanan tuketici=${sonraki}`);

// Ayni tuketici kendi hatasini duzeltip dorduncu kayittan yeniden isleyebilir mi?
const geriAl = oku("olaylar.log", 3, 100);
console.log(`akis  : konum 3'e geri alindi -> yeniden okunan kayit=${geriAl.length} ` +
            `ilk odunc=${geriAl[0].oduncId}`);

node yeniden-okuma.mjs

kuyruk: yazilan=6 birinci tuketici=6 derinlik=0 sonra baglanan tuketici=0
akis  : yazilan=6 birinci tuketici=6 gunluk uzunlugu=6 sonra baglanan tuketici=6
akis  : konum 3'e geri alindi -> yeniden okunan kayit=3 ilk odunc=4

Kuyrukta ikinci tüketici sıfır kayıt okuyabildi. Bu bir yapılandırma eksikliği değil: onay iletiyi sildiği için okunacak bir şey kalmamıştı. Akışta ikinci tüketici altı kaydın hepsini okudu ve bunun için günlükte hiçbir şey değişmedi.

Üçüncü satır yeniden okumanın (replay) asıl kullanımını gösterir. Rapor tüketicisi dördüncü olaydan itibaren yanlış hesap yaptığını fark ederse, düzeltilmiş kodla konumunu 3 değerine geri alması yeter; günlük ona aynı kayıtları aynı sırayla verir. Kuyrukta bunun karşılığı yoktur, çünkü geri alınacak bir konum yoktur — kuyruğun tuttuğu tek durum, hangi iletinin hâlâ orada olduğudur.

Bunun bedeli de aynı yerdedir. Konum tüketicide olduğu için onu kaydetmek de tüketicinin işidir. Kuyrukta “işlendi” bilgisini kuyruk tutar; akışta tüketici, konumu kendi kalıcı deposuna yazmadıkça yeniden başladığında nereden devam edeceğini bilemez. Konumun ne zaman yazılacağı — işten önce mi sonra mı — bir sonraki dersin konusudur.

Saklama Süresi

Kuyruk tükettikçe boşalır. Günlük boşalmaz; yalnız büyür. Bu yüzden akışta silme kararı tüketime değil bir kurala bağlanır: saklama süresi (retention). Kural süreye, boyuta ya da kayıt sayısına dayanabilir; ortak yanı, tüketicilerin nerede olduğuna bakmamasıdır.

// saklama.mjs — saklama suresi dolan kayitlar budanir; geride kalan tuketici ne kaybeder
import { writeFileSync } from "node:fs";
import { sifirla, ekle, kayitlar, oku } from "./gunluk.mjs";

let ilk = 0;                                  // gunlukte duran en eski kaydin konumu
function buda(dosya, tut) {                   // en yeni `tut` kayit birakilir
  const hepsi = kayitlar(dosya);
  const atilan = Math.max(0, hepsi.length - tut);
  writeFileSync(dosya, hepsi.slice(atilan).map((k) => JSON.stringify(k) + "\n").join(""));
  ilk += atilan;
  return atilan;
}
function okuKonumdan(dosya, konum, adet) {    // mutlak konum, budanmis gunlukte
  if (konum < ilk) return null;               // istenen kayit artik yok
  return oku(dosya, konum - ilk, adet);
}

sifirla("olaylar.log");
for (let n = 1; n <= 10; n++) ekle("olaylar.log", { tur: "odunc_verildi", oduncId: n });

const konum = { bildirim: 10, rapor: 2 };     // rapor sekiz kayit geride
console.log(`budamadan once: gunluk=${kayitlar("olaylar.log").length} kayit  ilk konum=${ilk}`);
console.log(`  rapor konum=${konum.rapor} -> okunabilen=${okuKonumdan("olaylar.log", konum.rapor, 100).length}`);

const atilan = buda("olaylar.log", 4);
console.log(`budama sonrasi: atilan=${atilan}  gunluk=${kayitlar("olaylar.log").length} kayit  ilk konum=${ilk}`);
const raporOkuma = okuKonumdan("olaylar.log", konum.rapor, 100);
console.log(`  bildirim konum=${konum.bildirim} -> okunabilen=${okuKonumdan("olaylar.log", konum.bildirim, 100).length}`);
console.log(`  rapor konum=${konum.rapor} -> ${raporOkuma === null ? "konum budandi, kayit geri alinamaz" : raporOkuma.length}`);
console.log(`raporun kaybettigi kayit sayisi = ${ilk - konum.rapor}`);

node saklama.mjs

budamadan once: gunluk=10 kayit  ilk konum=0
  rapor konum=2 -> okunabilen=8
budama sonrasi: atilan=6  gunluk=4 kayit  ilk konum=6
  bildirim konum=10 -> okunabilen=0
  rapor konum=2 -> konum budandi, kayit geri alinamaz
raporun kaybettigi kayit sayisi = 4

Budamadan önce rapor sekiz kaydı okuyabiliyordu. En yeni dört kayıt bırakılınca günlüğün en eski konumu 6 oldu ve raporun konumu bu değerin gerisinde kaldı; artık okuyabileceği bir yer yok. Kaybettiği kayıt sayısı iki konumun farkıdır: dört.

Bildirim tüketicisi aynı budamadan hiç etkilenmedi, çünkü konumu günlüğün sonundaydı ve okuyacak yeni kaydı zaten yoktu. Kuralın izlenmesi gereken ölçütü buradan çıkar: en geride kalan tüketicinin konumu ile günlüğün en eski konumu arasındaki fark. Bu fark sıfıra yaklaştığında saklama süresi ya uzatılır ya da geride kalan tüketici hızlandırılır.

Hangi Model Neyi Çözer

İki model birbirinin yerine geçmez; farklı soruların yanıtıdır.

Kuyruk, işin bölünmesi için kurulur. Her ileti bir kez bir tüketiciye gider, iş tüketiciler arasında paylaşılır, işlenen iş silinerek yer açılır. Gecikme bildirimi gönderme işi böyledir: kim gönderdiği önemsizdir, gönderildikten sonra iletinin saklanmasının bir değeri yoktur.

Akış, olayın kaydı için kurulur. Her tüketici tüm kayıtları kendi konumundan okur, aynı kaydı birden çok taraf görebilir, geçmiş silinmediği için yeniden okunabilir. Ödünç olaylarından türetilen şube stok özeti böyledir: hesap değişirse özet günlükten yeniden üretilir.

Ayrımı tek bir soruyla yapmak yeterlidir: aynı kaydın ikinci kez okunması anlamlı mı? Yanıt hayırsa kuyruk, evetse akış. Bu ayrım maliyetsiz değildir — akış saklama alanı, konum yönetimi ve budama kuralı ister; kuyruk bunların hiçbirini istemez.

Özet

• Akış, iletilerin yalnız sona eklendiği ve okumanın kaydı silmediği sıralı bir günlüktür; tüketicinin nereye kadar geldiği kuyrukta değil, tüketicideki konum değerinde durur.
• Altı olay tek günlükte 337 bayt, abonelik başına kopyalandığında üç dosyada 1011 bayt yer kapladı; üç tüketici 6, 4 ve 1 konumlarında birbirini beklemeden ilerledi.
• Aynı iş yükünde sonradan bağlanan tüketici kuyrukta sıfır, akışta altı kayıt okuyabildi; konumu 3 değerine geri alan tüketici üç kaydı yeniden okudu.
• Konum tüketicide durduğu için onu kalıcı olarak kaydetmek de tüketicinin işidir; kuyrukta bu bilgiyi kuyruğun kendisi tutar.
• Günlük tüketimle boşalmaz, saklama kuralıyla budanır: en yeni dört kayıt bırakıldığında en eski konum 6 oldu ve konumu 2 olan tüketici dört kaydı geri alamaz duruma düştü.

Sonraki Adım

Her iki modelde de bir soru yanıtsız kaldı: işin yapılması ile “yapıldı” bilgisinin yazılması iki ayrı adım. Kuyrukta onay işten önce mi sonra mı gönderilir, akışta konum işten önce mi sonra mı kaydedilir? Sıralamanın iki seçeneği vardır ve ikisi de bir şey kaybeder — biri iş yapılmadan işlenmiş sayar, öteki aynı işi iki kez yaptırır. Sonraki ders aynı iş yükünü üç kipte koşturur: en çok bir kez, en az bir kez ve tekillik anahtarıyla tekilleştirme. Kayıp sayısı ile yinelenen teslim sayısı ölçülür ve “tam bir kez”in uçtan uca neden bir kip değil, bir sonuç olduğu gösterilir.