Queue Pathologies

1. Poison message: kuyruğu kilitleyen tek satır

Bir emir mesajı geldi ve ayrıştırılamıyor. Belki bir alan eksik, belki volume alanında sayı yerine boş string var, belki karakter kodlaması bozuk. Consumer hata alıyor, ack göndermiyor, mesaj kuyrukta kalıyor ve yeniden teslim ediliyor.

Sonra aynı şey oluyor. Ve tekrar. Ve tekrar. Bu, saniyede yüzlerce kez dönen bir döngü ve arkadaki 40 bin emir sırada bekliyor.

İlaç: dead letter queue (DLQ)

Kural basit: bir mesaj N kez retry edildiyse artık denenmez, kenara alınır.

void isle(Mesaj m) {
    try {
        emriIsle(m);
        ack(m);
    } catch (KaliciHata e) {          // ayristirma, dogrulama, is kurali
        dlqYaz(m, e);                 // ← hic tekrar deneme, direkt kenara
        ack(m);
    } catch (GeciciHata e) {          // ag, zaman asimi, 503
        if (m.denemeSayisi >= 4) {          // 4 deneme yapildi: 100,200,400,800 ms
            dlqYaz(m, e);
            ack(m);
        } else {
            m.denemeSayisi++;
            gecikmeliTekrarKuyruguna(m, bekleme(m.denemeSayisi));
            ack(m);                   // asil kuyruktan CIKAR
        }
    }
}

İki ayrım kritik ve çoğu ekip birincisini atlıyor:

• Kalıcı hata hiç denenmez. Ayrıştırılamayan bir mesaj beşinci denemede de ayrıştırılamayacak. Tekrar denemek sadece zaman ve kaynak yakıyor. Sunucudan 400 aldıysan tekrar deneme; 503 aldıysan dene.
• Tekrar denenecek mesaj asıl kuyruktan çıkarılır. Ayrı bir delay queue’ya taşınır. Yerinde bırakırsan tekrar denemek ile sıradakileri bekletmek aynı şey olur. Bedeli var: kuyruktan çıkan mesaj sırasını da kaybeder — 4. bölümdeki sıra bozulmasının sebeplerinden biri tam olarak bu, ve orada cevabı var.

2. Exponential backoff — ve jitter’ın önemi

Geçici hatalarda hemen tekrar denemek işi kötüleştirir: karşı taraf zaten zorlanıyorsa, ikinci istek onu daha da zorlar. Klasik cevap her denemede beklemeyi katlamak:

100 ms → 200 ms → 400 ms → 800 ms → DLQ

Doğru ama eksik. Şunu düşün: köprü bir saniyeliğine düştü ve 800 emir aynı anda hata aldı. Hepsi 100 ms bekleyip aynı anda tekrar deniyor. Hepsi yine hata alıyor, hepsi 200 ms bekleyip yine aynı anda deniyor.

Yani rastgele bir yük yerine, senkronize dalgalar üretiyorsun. Köprü toparlanmaya çalışırken düzenli aralıklarla 800’lük tokat yiyor.

long bekleme(int deneme) {
    // deneme 1..4 -> 100, 200, 400, 800 ms (semayla ayni)
    long taban = Math.min(TABAN_MS * (1L << (deneme - 1)), TAVAN_MS);
    // "full jitter": 0 ile taban arasinda rastgele
    return ThreadLocalRandom.current().nextLong(taban + 1);
}

3. Head-of-line blocking: büyük emir küçükleri bekletir

Tek bir Redis stream var, bütün emirler oradan akıyor. Kurumsal bir müşteri 1000 lotluk bir emir gönderiyor; bu emir 40 parçaya bölünüp köprülere dağıtılıyor ve işlenmesi 2 saniye sürüyor.

O 2 saniye boyunca arkadaki 0,01 lotluk emirler bekliyor. Grafikte gecikme ortalaması iyi görünüyor ama %99’luk dilim tavan yapıyor.

Bizim yaptığımız birinci ve dördüncünün karışımıydı: 100 lot üstü emirler ayrı lane’e alındı ve orada parçalara bölünüp işlendi. Küçük emirlerin gecikmesi %99’luk dilimde 2100 ms’den 40 ms’ye indi — ortalama zaten iyiydi, sorun hep uçlardaydı (tail latency).

4. Message reordering: #3 önce, #1 sonra

Kısmi fill’ler (partial fill) sırayla gönderiliyor: #1 (10 lot), #2 (15 lot), #3 (5 lot). Sana geliş sırası: #3, #1, #2. Kümülatif hesap anlık olarak yanlış çıkıyor ve arayüzde müşteri tuhaf sayılar görüyor.

Neden olur?

• Mesajlar farklı partition’lara düştü, farklı consumer’lar işledi.
• #1 bir kez hata alıp tekrar denendi, bu sırada #2 ve #3 geçti.
• Ağda farklı yollar, farklı gecikmeler.

Üç çözüm, en zayıftan en sağlama

// Beklenen: 1. Gelen: 3.
// al() ve zamanAsimi() ayni durumu (beklenenSira, bekleyenler) degistirir: ya ikisi de
// synchronized, ya da timer mesajlari isleyen thread'de calisir (tek thread'li executor).
// Aksi halde ayni mesaj iki kez islenir ya da sayac bozulur.
synchronized void al(Mesaj m) {
    if (m.sira < beklenenSira) return;   // gec gelen eski mesaj: ya islendi ya atlandi, sirayi KAYDIRMA
    if (m.sira > beklenenSira) {
        bekleyenler.put(m.sira, m);       // kenarda tut
        return;
    }
    isle(m);
    beklenenSira++;
    bekleyendekileriBosalt();             // 1 geldiyse kenardaki 2 ve 3 de islenir
}

// AYRI bir timer, orn. her 50 ms. Mesaj akisi durursa yalnizca burasi calisir;
// zaman asimini "yeni mesaj gelince" kontrol edersen akis kesildiginde hic tetiklenmez.
synchronized void zamanAsimi() {
    if (!bekleyenler.isEmpty() && bekleyenler.enEskiYasi() > 200) {   // 200 ms'den fazla bekledi
        uyar("eksik mesaj: " + beklenenSira);
        beklenenSira = bekleyenler.ilkAnahtar();   // atla, kilitlenme
        bekleyendekileriBosalt();                  // atladiktan sonra kenardakiler islenir
    }
}

// Kirilgan — sira sart, delta gonderiyor
{ "order_id": 991, "fill_no": 2, "eklenen_lot": 15 }

// Saglam — sira onemsiz, mutlak deger gonderiyor
{ "order_id": 991, "fill_no": 2,
  "kumulatif_lot": 25, "kumulatif_tutar": 27512.50,
  "kalan_lot": 5 }

UPDATE pozisyonlar
   SET kumulatif_lot = :kumulatif_lot,
       son_fill_no   = :fill_no
 WHERE order_id = :order_id
   AND COALESCE(son_fill_no, 0) < :fill_no;   -- ← eski mesaj yazamaz; NULL < x false doner, o yuzden COALESCE

Aradaki fark tasarım seviyesinde: delta gönderen sistem her mesajın gelmesine ve sırasına muhtaç; mutlak değer gönderen sistem tek bir mesajla kendini toparlıyor. Yeni bir mesaj şeması tasarlarken sorulacak soru şu: “Bu mesajı tek başına alsam, doğru duruma varabilir miyim?”

5. Backpressure: 50 bin tick, yetişemeyen consumer

Piyasa hareketlendi. Saniyede 50 bin tick geliyor, streaming-engine 30 bin işleyebiliyor. Aradaki 20 bin nereye gidiyor? Dört seçeneğin var ve sonuncusu bir seçenek değil:

Sonuncusu en çok yapılanı, çünkü kod yazarken en görünmezi. Sınırsız bir kuyruk oluşturmak tek satır ve o satır, sistemin ne zaman öleceğini belirsizleştiriyor. Her kuyruğa bir üst sınır koy ve sınır dolduğunda ne olacağını bilinçli seç.

Fiyat için: drop değil, conflation

Fiyat tarafında güzel bir numara var. EURUSD için 40 tick birikmişse, 40’ını da işlemenin bir anlamı yok — müşteri sadece sonuncuyu görecek. Bu yüzden kuyrukta biriktirmek yerine sembol başına son değeri tutuyorsun:

// Kuyruk degil, sembol basina tek slot
ConcurrentHashMap<String, Kotasyon> sonFiyat = new ConcurrentHashMap<>();

// Uretici: uzerine yaz, biriktirme
sonFiyat.put(k.sembol, k);

// Tuketici: yetisebildigi hizda oku; okudugunu CIKAR,
// yoksa yeni tick gelmese de her turda ayni fiyati yeniden yayinlarsin
for (String s : sonFiyat.keySet()) {
    Kotasyon k = sonFiyat.remove(s);
    if (k != null) yayinla(k);
}

Emir için: bounded queue ve dürüst ret

Emir düşürülemez. Ama sınırsız da biriktirilemez. Doğru davranış, kuyruk dolduğunda yeni emri açıkça reddetmek:

BlockingQueue<Emir> kuyruk = new ArrayBlockingQueue<>(10_000);

if (!kuyruk.offer(emir)) {                 // dolu
    metrik.artir("emir_reddedildi_kuyruk_dolu");
    return hata("SYSTEM_BUSY, lutfen tekrar deneyin");
}

Ne izlemeli?

Kontrol listesi

Sonuç

Kuyruk, sistemler arasına konan bir yastık. Ama yastık aynı zamanda bir gizleyici: arkasındaki yavaşlığı bir süre saklıyor, sonra hepsini birden önüne koyuyor.

Baştaki 40 bin mesajlık kuyruk, tek bir bozuk emirden çıkmıştı. Düzeltmesi on beş satırdı. Asıl ders, o on beş satırın ilk günden yazılması gerektiğiydi — çünkü zehirli mesaj bir ihtimal değil, bir kesinlik.