Redis, önbellek ve arka plan işleri (Celery/RQ)

Neden önemli?

Lezzet Kurye'nin uygulamasında restoran menüsü saniyede yüzlerce kez okunur ama günde birkaç kez değişir; her okumada veritabanına gitmek gereksiz yüktür. Sipariş verildiğinde müşteriye SMS, restorana bildirim, kuryeye atama gider; bunları HTTP isteğinin içinde yapmak yanıtı saniyelerce geciktirir ve SMS sağlayıcısı çökünce sipariş de çöker. Bir müşteri dakikada 200 istek atınca diğerlerini etkilememelidir. Bu üç sorunun ortak aracı Redis'tir: bellek içi, çok hızlı bir anahtar-değer veri yapısı sunucusu. Önbellek (cache), hız sınırlayıcı, oturum deposu, dağıtık kilit ve görev kuyruğu aracısı (broker) olarak kullanılır.

Redis veri türleri

# docker run -d -p 6379:6379 redis:7     ;     pip install redis
import json, redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
r.set("menu:5", json.dumps([{"ad": "İskender", "fiyat": 295}]), ex=300)   # 5 dk TTL
print(r.get("menu:5"), r.ttl("menu:5"))
r.hset("kurye:47", mapping={"enlem": 41.01, "boylam": 29.03, "durum": "yolda"})
r.zincrby("lider:2024-10", 1, "kurye:47")
print(r.zrevrange("lider:2024-10", 0, 2, withscores=True))

Redis tek iş parçacıklı bir olay döngüsüyle komutları sırayla işler; bu yüzden INCR gibi tekil komutlar atomiktir. Ama "oku, Python'da karar ver, yaz" dizisi atomik değildir; araya başka istemci girebilir. Bunun için ya tek komut (SET NX, INCR), ya MULTI/EXEC işlemi, ya da sunucuda çalışan Lua betiği kullanılır.

FakeRedis: sözlük + TTL + enjekte edilen saat

Gerçek Redis'in temel sözleşmesini küçük bir sınıfla taklit edelim. Her anahtarın bir bitiş anı olabilir; süresi dolan anahtar okunduğunda silinir (Redis de bunu "tembel silme" ve arka plan örneklemesiyle yapar). Saat bir fonksiyon olarak verilir: testte saat.ilerle(301) demek, 5 dakika beklemekle aynıdır.

class Saat:
    def __init__(self, t=0.0):
        self.t = t
    def __call__(self):
        return self.t
    def ilerle(self, s):
        self.t += s

class MiniRedis:
    def __init__(self, saat):
        self.saat, self.v, self.bitis = saat, {}, {}
    def _canli(self, k):
        if k in self.bitis and self.saat() >= self.bitis[k]:
            del self.v[k], self.bitis[k]
        return k in self.v
    def set(self, k, deger, ex=None):
        self.v[k] = deger
        self.bitis.pop(k, None)
        if ex is not None:
            self.bitis[k] = self.saat() + ex
    def get(self, k):
        return self.v[k] if self._canli(k) else None
    def ttl(self, k):
        if not self._canli(k):
            return -2
        return -1 if k not in self.bitis else int(self.bitis[k] - self.saat())

saat = Saat()
r = MiniRedis(saat)
r.set("menu:5", "[...]", ex=300)
r.set("ayar:komisyon", "18")
print(r.get("menu:5"), r.ttl("menu:5"), r.ttl("ayar:komisyon"), r.ttl("yok"))
saat.ilerle(300)
print(r.get("menu:5"), r.ttl("menu:5"))

Önbellek desenleri

Geçersizleştirme (invalidation) için iki araç birlikte kullanılır: veri değişince anahtarı silmek ve her anahtara TTL vermek (silmeyi unuttuğun yol için emniyet ağı). TTL'lere küçük rastgele bir pay (jitter) eklemek, aynı anda doldurulan binlerce anahtarın aynı anda düşmesini önler.

import json
import veri

con = veri.baglan("lezzet_kurye", tablolar=["lk_menu_items"])
r = MiniRedis(saat)
db_sorgu = 0

def menu_getir(restoran_id):
    global db_sorgu
    anahtar = f"menu:{restoran_id}"
    onbellek = r.get(anahtar)
    if onbellek is not None:
        return json.loads(onbellek)                     # isabet (hit)
    db_sorgu += 1                                       # ıskalama (miss)
    satirlar = con.execute("SELECT name, price FROM lk_menu_items WHERE restaurant_id = ? ORDER BY item_id",
                           (restoran_id,)).fetchall()
    r.set(anahtar, json.dumps(satirlar), ex=300)
    return satirlar

def fiyat_guncelle(item_id, fiyat, restoran_id):
    con.execute("UPDATE lk_menu_items SET price = ? WHERE item_id = ?", (fiyat, item_id))
    r.v.pop(f"menu:{restoran_id}", None)                # DEL: bir sonraki okuma tazeler

for _ in range(1000):
    menu = menu_getir(5)
print("1000 okuma, veritabanı sorgusu:", db_sorgu, "| İskender:", menu[2])
fiyat_guncelle(26, 310.0, 5)
print("güncelleme sonrası:", menu_getir(5)[2], "| sorgu:", db_sorgu)

Cache stampede (önbellek hücumu)

Çok okunan bir anahtarın süresi dolduğu anda yüzlerce istek aynı anda ıskalar ve hepsi veritabanına gider; veritabanı yavaşlar, yavaşladıkça daha çok istek birikir. Çözümler: tek uçuş (single-flight; anahtar başına yalnız bir yükleyici, diğerleri onun sonucunu bekler; dağıtık ortamda kısa bir Redis kilidiyle), erken tazeleme (TTL bitmeden, olasılıksal olarak bir isteğin yenilemesi) ve bayat veriyi kısa süre sunmak (stale-while-revalidate).

Hız sınırlama: token bucket ve kayan pencere

Token bucket'ta kova kapasite kadar jeton alır ve saniyede hiz jeton dolar. Her istek bir jeton harcar; jeton yoksa istek 429 alır. Durum yalnız iki sayıdır; dolum, istek geldiğinde geçen süreyle hesaplanır.

def token_bucket(durum, simdi, kapasite=5, hiz=1.0):
    jeton, son = durum.get("jeton", kapasite), durum.get("son", simdi)
    jeton = min(kapasite, jeton + (simdi - son) * hiz)    # geçen süre kadar doldur
    izin = jeton >= 1
    durum.update(jeton=jeton - 1 if izin else jeton, son=simdi)
    return izin

durum = {}
zamanlar = [0, 0, 0, 0, 0, 0, 0, 0.5, 1.0, 3.0]
print([token_bucket(durum, t) for t in zamanlar])

İlk 5 istek kovayı boşaltır, 6. ve 7. reddedilir; 1. saniyede bir jeton dolduğu için izin çıkar, 3. saniyede ise iki jeton birikmiştir. Redis'te bu hesap birden çok sunucu arasında paylaşılır ve yarışı önlemek için Lua betiği olarak tek adımda çalıştırılır.

Dağıtık kilit ve oturum deposu

"Bu siparişe aynı anda iki kurye atanmasın" gibi durumlarda birden çok sunucu arasında kilit gerekir. Doğru desen: SET kilit:siparis:42 <rastgele_token> NX PX 30000. NX anahtar yoksa yazar (kilidi alan tek istemci olur), PX süre verir (süreç çökerse kilit sonsuza kalmaz), token ise bırakırken "bu kilit hâlâ benim mi?" kontrolü içindir; bırakma bu yüzden karşılaştır-ve-sil yapan bir Lua betiğiyle yapılır.

Oturum deposu olarak Redis, oturum kimliğini anahtar, kullanıcı verisini hash olarak tutar ve her istekte EXPIRE ile süreyi uzatır (kayan oturum süresi). Django'da SESSION_ENGINE = "django.contrib.sessions.backends.cache" ve Redis önbellek ayarı bunu yapar; birden çok uygulama sunucusu aynı oturumları görür.

Görev kuyrukları: üretici, aracı, işçi

Web isteği (üretici) "SMS gönder" işini bir mesaj olarak aracıya (broker: Redis ya da RabbitMQ) bırakır ve hemen 201 döner. Ayrı süreçlerde çalışan işçiler (worker) kuyruktan mesaj alıp işi yapar. Sonucu saklamak gerekiyorsa sonuç deposu (result backend) kullanılır. Periyodik işler (her gece rapor) bir zamanlayıcı (Celery beat) tarafından kuyruğa atılır; Flower ise işçileri ve görevleri izleyen web arayüzüdür.

# pip install "celery[redis]"      ->   celery -A gorevler worker -l info     /     celery -A gorevler beat
from celery import Celery
from celery.schedules import crontab

app = Celery("lezzet", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1")
app.conf.task_acks_late = True              # iş BİTİNCE onayla; işçi çökerse mesaj geri döner
app.conf.worker_prefetch_multiplier = 1

@app.task(bind=True, autoretry_for=(ConnectionError,), retry_backoff=2, retry_backoff_max=300,
          retry_jitter=True, max_retries=5)
def sms_gonder(self, siparis_id):
    if Bildirim.objects.filter(siparis_id=siparis_id, tur="sms").exists():   # idempotent
        return "zaten gönderildi"
    saglayici.gonder(...)
    Bildirim.objects.create(siparis_id=siparis_id, tur="sms")

app.conf.beat_schedule = {
    "gece-raporu": {"task": "gorevler.gunluk_rapor", "schedule": crontab(hour=3, minute=0)},
}
# Django view içinde:  transaction.on_commit(lambda: sms_gonder.delay(siparis.id))

# RQ: daha sade bir alternatif   ->  pip install rq   ;   rq worker bildirim
from redis import Redis
from rq import Queue, Retry
q = Queue("bildirim", connection=Redis())
is_ = q.enqueue(sms_gonder, 42, retry=Retry(max=3, interval=[10, 30, 60]))

Aynı model tarayıcıda asyncio.Queue ve sabit sayıda işçi göreviyle kurulabilir. Aşağıda üç işçi, Lezzet Kurye'nin ilk 12 siparişi için bildirim kuyruğunu boşaltıyor.

import asyncio
import veri

siparisler = veri.kayitlar("lezzet_kurye", "lk_orders")[:12]

async def isci(ad, kuyruk, kayit):
    while True:
        s = await kuyruk.get()
        try:
            await asyncio.sleep(0.001)                  # SMS servisine gidiyormuş gibi
            kayit.append((ad, s["order_id"], s["status"]))
        finally:
            kuyruk.task_done()

async def main():
    kuyruk, kayit = asyncio.Queue(), []
    for s in siparisler:
        kuyruk.put_nowait(s)                            # üretici
    iscilar = [asyncio.create_task(isci(f"i{n}", kuyruk, kayit)) for n in range(3)]
    await kuyruk.join()                                 # tüm işler bitene kadar bekle
    for t in iscilar:
        t.cancel()
    await asyncio.gather(*iscilar, return_exceptions=True)
    return kayit

kayit = asyncio.run(main())
print(len(kayit), "bildirim;", sorted({ad for ad, _, _ in kayit}))

Yeniden deneme, idempotentlik ve ölü mektup kuyruğu

Kuyruk sistemleri genelde "en az bir kez" (at-least-once) teslim eder: işçi işi bitirip onay göndermeden çökerse mesaj başka işçiye tekrar verilir. Bu yüzden görevler idempotent olmalıdır: iki kez çalışması bir kez çalışmasıyla aynı sonucu vermeli (aynı müşteriye iki SMS, iki kez iade olmamalı). Yöntem: bir idempotency anahtarı (siparis:42:sms) ile işi önce SET NX ya da benzersiz kısıtlı bir tabloyla "kaydet", sonra yap.

Geçici hatalar (zaman aşımı, 503) üstel geri çekilme (exponential backoff) ile yeniden denenir: bekleme = min(tavan, taban × 2^deneme). Tüm işçilerin aynı anda yeniden denemesini önlemek için jitter eklenir (tam jitter: random.uniform(0, bekleme)). Kalıcı hatalar (geçersiz telefon) ya da deneme hakkı biten işler ölü mektup kuyruğuna (dead letter queue, DLQ) gider; orada bir insan inceler, düzeltir ve gerekirse yeniden kuyruğa atar.

import random

def bekleme_plani(deneme_sayisi, taban=1, tavan=60, jitter=False, rng=random.Random(7)):
    plan = []
    for n in range(deneme_sayisi):
        b = min(tavan, taban * 2 ** n)
        plan.append(round(rng.uniform(0, b), 2) if jitter else b)
    return plan

print("üstel      :", bekleme_plani(8))
print("tam jitter :", bekleme_plani(8, jitter=True))

Pub/sub ile basit bildirim

PUBLISH siparis:durum "42 yolda" ile gönderilen mesaj, o anda SUBSCRIBE olmuş herkese iletilir; abone o an bağlı değilse mesaj kaybolur (at-most-once). Canlı panel güncellemesi, önbellek geçersizleştirme duyurusu gibi kaybı tolere edilen işler için uygundur. Kalıcılık ve yeniden okuma gerekiyorsa Redis Streams ya da Kafka kullan (akış işleme dersinde).