Worker Sharding: İşi Bölmenin Doğru Yolu

Önce dur: sharding mi, message broker mı?

Message broker: bütün worker’lar aynı kuyruktan çeker, kim önce alırsa onun olur (RabbitMQ tek queue + N consumer, Redis BRPOP, SQS standart kuyruk). Kafka bu listede yok: partition key seçtiğin an zaten sharding yapıyorsun. Çoğu durumda broker yetiyor ve gereksiz yere sharding yazılıyor:

Kritik cümle şu: sharding’in asıl amacı hız değil, çakışmayı ortadan kaldırmak. Kilit yaklaşımında herkes aynı kapıya koşup sırayla bekliyor; sharding’de zaten farklı kapılardan giriyorlar.

Başlangıç noktası: Modulo (herkes buradan başlar, kimse burada kalmamalı)

Anahtarın hash’ini worker sayısına böl, kalan (7 % 4 = 3) hangi worker’a gideceğini söyler:

shard = hash(kullanici_id) % WORKER_SAYISI

# worker 2 sadece sunu isler (4 worker, ikinin kuvveti -> & 3):
SELECT * FROM isler
 WHERE durum = 'bekliyor'
   AND (hashtext(kullanici_id::text) & 3) = 2   -- her seferinde tam tarama; asagiya bak
 ORDER BY olusma
 LIMIT 100;

-- shard yazma aninda hesaplanir, kolonda durur, indekslenir
ALTER TABLE isler ADD COLUMN shard smallint NOT NULL;      -- uygulama: murmur3(kullanici_id) & 63
CREATE INDEX isler_bekleyen ON isler (shard, olusma) WHERE durum = 'bekliyor';

SELECT * FROM isler WHERE durum = 'bekliyor' AND shard IN (16,17,...,31)   -- benim shard'larim
 ORDER BY olusma LIMIT 100 FOR UPDATE SKIP LOCKED;
-- not: SKIP LOCKED + LIMIT istedigin sayidan az satir dondurebilir (kilitliler limit'ten sonra atilir)

Modulo’nun kusuru anahtar → worker eşlemesinin worker sayısına bağlı olması. İki yöntem de bu bağı koparıyor; farklı yerlerden.

Yöntem 1: Sabit Shard (Static Partitioning)

İşleri worker sayısına değil, önceden belirlenmiş sabit bir shard sayısına bölersin: 64, 128 ya da 256. İki ayrı eşleme var ve ikisini karıştırmamak bütün mesele:

• Anahtar → Shard: murmur3(kullanici_id) & 63. Bu eşleme sistemin ömrü boyunca sabittir, asla değişmez — o yüzden hash de sabit olmalı (yukarıdaki kutu).
• Shard → Worker: 64 shard’ı o an ayakta olan worker’lara paylaştırırsın; 4 worker varsa her birine 16. Değişen yalnızca bu.
• Dağıtım nasıl yapılır: ring’e gerek yok. Arkada bir coordinator durur — etcd, Consul, ZooKeeper ya da basit bir lease tablosu — ve worker geldiğinde boşta kalan shard’ları oradan zimmetler.
• Nerede kullanılır: Kafka (partition mantığı bu — bir fark: partition sayısını artırınca anahtar → partition eşlemesi bozulur, azaltmak ise mümkün değil; yani Kafka’da partition sayısını ilk günden bol seçmek zorundasın), Redis Cluster (16384 hash slot), veritabanı shard’lama.

SHARD_SAYISI = 64          # asla degismez
shard = hash(kullanici_id) & 63

# 4 worker varsa:
worker 0 -> shard 0..15
worker 1 -> shard 16..31
worker 2 -> shard 32..47
worker 3 -> shard 48..63

# 5. worker eklendiginde shard'lar yeniden bolusulur,
# ama bir anahtarin SHARD'i degismez. Sadece sahibi degisir.

Anahtar → shard sabit kaldığı için shard başına tutulan durum, cache ve ilerleme taşınabilir hale geliyor. Shard sayısını seçerken tek kural: ulaşmayı hayal ettiğin worker sayısının birkaç katı olsun — çünkü paralelliğin tavanı shard sayısıdır: 64 shard varken 100 worker açarsan 36’sı boş oturur. Ve “asla değişmez”in bir kaçağı var: ikinin kuvveti seçtiysen 64 → 128 zararsız, bir anahtarın yeni shard’ı ya eskisi ya da eski+64 olur; veri en fazla bir kez ve öngörülebilir biçimde bölünür.

Shard’ı worker’a kim verir: coordinator

“Shard 17 şu an kimde?” sorusuna tek bir yer cevap vermeli; her worker kendi hesaplarsa geçiş anında iki worker aynı shard’ı kendine ait sanır. En basit coordinator, zaten elindeki veritabanı. Tasarımın zor kısmı sahiplenme değil, her worker’ın payını bilmesi: ilk açılan worker 64’ün hepsini kapmasın diye önce canlı worker sayısı, sonra pay.

CREATE TABLE shard_sahipligi (
  shard          int PRIMARY KEY,   -- 0..63
  sahip          text,
  lease_bitis    timestamptz,
  claim_version  bigint NOT NULL DEFAULT 0   -- fencing token, asagiya bak
);

-- 1) payimi hesapla: canli worker = son 30 sn'de heartbeat atanlar
SELECT ceil(64.0 / count(DISTINCT sahip)) AS pay
  FROM shard_sahipligi WHERE lease_bitis > clock_timestamp();   -- + kendim
-- clock_timestamp(): now() islem basindaki saati verir, uzun islemde lease'i yanlis okursun

-- 2) elimde pay'dan az varsa: sahipsiz ya da lease'i dolmus shard'lari kap
UPDATE shard_sahipligi
   SET sahip = :ben,
       lease_bitis = now() + interval '30 seconds',
       claim_version = claim_version + 1
 WHERE shard IN (
       SELECT shard FROM shard_sahipligi
        WHERE sahip IS NULL OR lease_bitis < now()
        LIMIT :pay - :elimdeki
        FOR UPDATE SKIP LOCKED           -- iki worker ayni satira takilmasin
 )
RETURNING shard, claim_version;          -- her shard'in KENDI surumu

-- 3) her 10 sn: elimdeki (shard, surum) ciftlerinin lease'ini tazele (heartbeat)
UPDATE shard_sahipligi SET lease_bitis = clock_timestamp() + interval '30 seconds'
 WHERE sahip = :ben
   AND (shard, claim_version) IN ((17, 7), (18, 7), (31, 9))    -- elimdekiler
RETURNING shard;
-- 16 gonderdim, 12 dondu -> 4 shard'i KAYBETTIM: o shard'larin islerini HEMEN birak (self-fencing)
-- DB'ye hic ulasamiyorsam da ayni: lease'i tazeleyemeyen worker calismayi durdurur

-- 4) elimde pay'dan cok varsa fazlasini birak (yeni worker gelmis demek)

-- isler tablosunda: fence bigint NOT NULL DEFAULT 0
UPDATE isler
   SET durum = 'tamamlandi', sonuc = :sonuc, fence = :bendeki_surum
 WHERE id = :id
   AND fence <= :bendeki_surum;   -- A v7 ile gelir, B v8'i coktan yazmistir -> 0 satir, GERI AL

Sıra garantisi gerekmiyorsa shard katmanına hiç girme: tek tabloda FOR UPDATE SKIP LOCKED ile her worker işi tek tek kapar, yeniden dağıtım diye bir şey olmaz, autoscale bedavaya gelir. Deneme sayısı, backoff ve dead-letter da oranın konusu; o ayrı bir yazı.

Yöntem 2: Ring Hashing (Consistent Hashing)

Arada “sabit 64 shard” gibi bir katman yoktur. Anahtarı da worker’ı da aynı sanal ring’in (0’dan 2^32’ye bir daire) üstüne dizersin:

• Anahtar → Worker: hash(kullanici_id) ring’de bir noktaya düşer; saat yönünde ilerleyip karşılaştığı ilk worker işi alır. Kayıttan okunmaz, hesaplanır.
• Dinamik mimaridir: worker sayısı 4’ten 5’e çıktığında ring’e yeni bir nokta eklenir; yalnızca o noktanın komşusundaki anahtarlar (≈ 1/yeni worker sayısı) yeni worker’a kayar. Arada mantıksal shard katmanı da coordinator yönetimi de yoktur.
• Nerede kullanılır: Memcached, DynamoDB, Envoy/Nginx load balancer, streaming ve WebSocket node dağıtımları.

Ring (0 .. 2^32):

   worker-A ●          ● worker-B
       ↘  anahtar 7,12,31  ↙
            ●
        worker-C

# Yeni worker-D eklendiginde:
# Sadece D'nin ring'deki KOMSUSUNUN anahtarlari tasiniyor.
# Digerleri hic dokunulmadan kaliyor.

sahip(anahtar) = ring'de anahtarin hash'inden sonraki ilk worker

# Tek noktali basitlestirme. Gercekte her worker 100-200 sanal noktada durur;
# D eklenince ring'in 150 yerine girer ve HER worker'dan orantili anahtar alir.
# Toplam yine ~1/N, ama "tek komsu etkilenir" yalnizca bu cizimde dogru.

Pratik detay: her worker’ı ring’e tek noktada değil, 100–200 sanal noktada yerleştir; tek nokta koyarsan dağılım şansa kalır ve bir worker’a diğerinin üç katı düşebilir. Kütüphanelerin çoğunda virtual node zaten varsayılan.

Neden önemli: reconnect storm

Rebalancing: kademeli devret

Ring taşınacak anahtar sayısını azaltıyor ama sıfırlamıyor. Taşınanları hepsini bir anda değil, parça parça devret — sabit shard için de aynen:

for (parca : tasinacakParcalar) {          // shard ya da anahtar araligi
    eskiSahip.durdur(parca);        // yeni is almayi birak, elindekini bitir
    eskiSahip.onaylaBitti(parca);   // ← beklenecek
    yeniSahip.basla(parca);
    Thread.sleep(200);              // ← sonraki parcaya gecmeden once nefes
}

Hangisi: Sabit Shard mı, Ring Hashing mi?

İkisini karıştırma — biz bir süre karıştırdık: sabit shard’ların üstüne bir de ring kurmuştuk. Shard katmanın varsa ring’e gerek yok; 64 shard’ı bir tabloyla dağıtmak hem daha basit hem de görünür. Ring, shard katmanı olmayan sistemlerin cevabı. Biz üretimde sabit shard’la kaldık.

Sahadan: tek müşteri, dört worker, hiç işe yaramayan hash

Bildirim gönderim sistemi musteri_id % 4 ile bölünmüştü. Kurumsal müşterilerden biri tek başına günlük bildirimlerin yaklaşık %40’ını üretiyordu. Sonuç: o müşterinin düştüğü worker sürekli %100 CPU, diğer üçü boşta. Worker sayısını 8’e çıkardık — hiçbir şey değişmedi, çünkü o müşteri hâlâ tek bir worker’a düşüyordu.

-- once
shard = hash(musteri_id) & 63

-- sonra: butun shard'lara yay
shard = hash(musteri_id || ':' || bildirim_id) & 63

-- ara yol (salting): 8 shard'a yay, kismi locality'yi koru
shard = hash(musteri_id || ':' || (bildirim_id % 8)) & 63

Sıranın önemli olduğu bir sistemde tersi başımıza geldi: sipariş durumları (hazırlanıyor → kargoda → teslim edildi) iki worker’a bölününce bazı siparişler “teslim edildi”den “kargoda”ya geri döndü. Sipariş kimliğine göre shard’ladık ve veritabanına da kendini koruttuk: UPDATE ... WHERE durum_sirasi < :yeni_sira — geç gelen eski mesaj yazamıyor. Fencing token ile aynı fikir: doğruluğu zamanlamaya değil, sıraya emanet etmek.

Ölçülmesi gereken dört şey

Üçüncü satır önemli: ortalamaya bakma. Dört worker’dan üçü boştaysa ortalama gecikme harika görünür; sıkışan shard’daki müşteri ise bildirimlerini yarım saat sonra alır.

Kontrol listesi

Sonuç

İki yöntem var ve seçim basit: shard katmanı kurabiliyorsan sabit shard + bir tablo; kuramıyorsan ring. Baştaki tabloya dönersek: o üç boş worker’ın sebebi kapasite değil, anahtar seçimiydi — ve doğru anahtar her zaman verinin gerçek dağılımına bakılarak seçiliyor, tahminle değil.