Redis kurmadan iş kuyruğu: SQLite ile yeniden denenebilir görev kuyruğu tasarımı

SQLite ile iş kuyruğu kurmak, Redis ya da RabbitMQ gibi ayrı bir servis işletmeden, bir otomasyonun işlerini sıraya koyup başarısız olanı yeniden denemenin en hafif yoludur. Küçük ekiplerde tipik ihtiyaç şudur: bir form doldurulduğunda e-posta gidecek, bir dosya düştüğünde dönüştürülecek, her gece bin müşteri için rapor üretilecek. Bu işlerin tamamı tek bir betiğin içinde sırayla çalıştığında, ortadaki bir hata hem kendisinden sonrakileri durdurur hem de nerede kalındığını belirsizleştirir.

Bir kuyruk, işi üreten tarafla yapan tarafı ayırır. Üreten taraf kaydı tabloya yazar ve döner; işçi süreç kayıtları sırayla alır, biten işi işaretler, başarısız olanı bir süre sonra yeniden dener. Bu ayrım için çoğu rehber Redis ve Celery önerir. Tek sunuculu, günde birkaç bin işi geçmeyen bir kurulumda ise bu araçlar hem fazladan bir servis hem de fazladan bir hata noktasıdır.

Bu yazıda SQLite üzerinde çalışan, birden fazla işçinin aynı işi almasını engelleyen, üstel geri çekilmeyle yeniden deneyen ve çökmüş işçinin elinde kalan işi kurtaran bir kuyruk kuruyorum. Örnekler yalnız Python standart kütüphanesini kullanır; ek paket kurulmaz. Kod parçaları aynı modülün bölümleridir ve birbirini çağırır.

İş kuyruğu ne zaman gerekir

İş kuyruğu, bir isteği alan tarafın işi hemen yapmak zorunda kalmadığı ve işin kaybolmadan, sırayla ve yeniden denenebilir biçimde yürütüldüğü yapıdır. Üç belirti kuyruk gerektiğini söyler: bir web isteği ya da form, cevap dönmeden önce yavaş bir işlemi bekliyorsa; bir toplu iş ortasında çöktüğünde baştan başlıyorsa; başarısız olan tek bir kaydı yeniden denemek için bütün betiği elle çalıştırıyorsanız.

SQLite'ın yeterli olduğu sınır, iş hacminden çok işçi sayısıyla ilgilidir. Tek sunucuda çalışan birkaç işçi süreci, WAL kipindeki bir SQLite dosyasını sorunsuz paylaşır. İşçiler farklı makinelere dağılacaksa ya da saniyede yüzlerce iş alınıp bırakılacaksa, ağ üzerinden erişilen bir kuyruk servisine geçmek gerekir. O noktaya kadar SQLite, yedeği tek dosya kopyalamak kadar kolay olan bir kuyruktur.

Bu yaklaşım, SQLite ile küçük ama sağlam veri boru hatları yazısında anlattığım kurulumun devamıdır: veriyi tutan dosya ile işi tutan dosya aynı motoru kullanır, aynı yedek düzenine girer ve aynı araçlarla incelenir.

Tablo tasarımı: durum, deneme sayısı ve zaman damgaları

Kuyruk tablosu, her işi bir satır olarak tutar ve dört bilgiyi kayıt altına alır: işin türü ve yükü, hangi durumda olduğu, kaç kez denendiği ve en erken ne zaman çalışabileceği. Durum alanı dört değer alır: bekliyor, çalışıyor, bitti ve ölü. Zaman damgalarının tamamı UTC ve ISO 8601 biçiminde metin olarak saklanır; böylece karşılaştırma düz metin sıralamasıyla doğru çalışır.

import json
import sqlite3
from datetime import datetime, timezone

SEMA = """
CREATE TABLE IF NOT EXISTS isler (
    id          INTEGER PRIMARY KEY,
    tur         TEXT NOT NULL,
    yuk         TEXT NOT NULL,
    durum       TEXT NOT NULL DEFAULT 'bekliyor',
    deneme      INTEGER NOT NULL DEFAULT 0,
    en_cok      INTEGER NOT NULL DEFAULT 5,
    calistir_at TEXT NOT NULL,
    kilit_at    TEXT,
    isci        TEXT,
    son_hata    TEXT,
    olusturma   TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS isler_bekleyen ON isler (durum, calistir_at);
"""


def simdi():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def baglan(yol="kuyruk.db"):
    con = sqlite3.connect(yol, timeout=30, isolation_level=None)
    con.execute("PRAGMA journal_mode=WAL")
    con.execute("PRAGMA busy_timeout=30000")
    con.executescript(SEMA)
    return con


def ekle(con, tur, yuk, calistir_at=None):
    con.execute(
        "INSERT INTO isler (tur, yuk, calistir_at, olusturma) VALUES (?, ?, ?, ?)",
        (tur, json.dumps(yuk, ensure_ascii=False), calistir_at or simdi(), simdi()),
    )

Yük alanı JSON metnidir; işçinin ihtiyaç duyduğu her şey oradadır ve kuyruk tablosu işin anlamını bilmek zorunda değildir. en_cok alanı iş başına farklı yeniden deneme sınırı vermeye izin verir: e-posta gönderimi beş kez denenebilir, bir ödeme çağrısı ise yalnız bir kez. İndeks, işçinin en sık çalıştırdığı sorguyu, yani çalışmaya hazır en eski işi bulmayı hızlandırır.

Bağlantı WAL kipinde açılır ve busy_timeout ayarlanır. WAL, bir işçi yazarken diğerlerinin okumasına izin verir; busy_timeout ise kilit çakışmasında hemen hata almak yerine kısa bir süre beklemeyi sağlar. Bu iki ayar olmadan birden fazla işçi çalıştırmak rastgele "database is locked" hatalarına yol açar. isolation_level değerinin None olması, işlemleri Python'ın örtük başlatması yerine kodda açıkça yönetmemizi sağlar.

İşi güvenle almak: iki işçi aynı satırı görmesin

İş alma, kuyruğun en kritik adımıdır: iki işçi aynı anda hazır işi sorgularsa ikisi de aynı satırı görüp aynı e-postayı iki kez gönderebilir. SQLite'ta çözüm, seçme ve işaretleme adımlarını BEGIN IMMEDIATE ile açılan tek bir yazma işlemine koymaktır. Komut, işlem daha ilk satırı okumadan yazma kilidini alır; ikinci işçi kilit bırakılana kadar bekler ve ardından güncellenmiş tabloyu görür.

import os
import socket
import time
import traceback

ISCI = "%s-%d" % (socket.gethostname(), os.getpid())


def al(con):
    con.execute("BEGIN IMMEDIATE")
    satir = con.execute(
        "SELECT id, tur, yuk, deneme, en_cok FROM isler "
        "WHERE durum = 'bekliyor' AND calistir_at <= ? "
        "ORDER BY calistir_at LIMIT 1",
        (simdi(),),
    ).fetchone()
    if satir is None:
        con.execute("COMMIT")
        return None
    con.execute(
        "UPDATE isler SET durum = 'calisiyor', kilit_at = ?, isci = ?, "
        "deneme = deneme + 1 WHERE id = ?",
        (simdi(), ISCI, satir[0]),
    )
    con.execute("COMMIT")
    return satir


def calis(con, isleyiciler, bekleme=2):
    while True:
        satir = al(con)
        if satir is None:
            time.sleep(bekleme)
            continue
        is_id, tur, yuk, deneme, en_cok = satir
        deneme += 1
        try:
            isleyiciler[tur](json.loads(yuk))
            tamamla(con, is_id)
        except Exception:
            basarisiz(con, is_id, deneme, en_cok, traceback.format_exc()[-2000:])

İşçi kimliği makine adı ve süreç numarasından üretilir ve alınan her satıra yazılır. Böylece bir iş takıldığında hangi sürecin elinde olduğu görülür. deneme alanı iş alınırken artırılır, bitince değil; işçi işin ortasında çökse bile o deneme sayılmış olur ve sonsuz tekrar döngüsü oluşmaz.

İşleyici fonksiyonlar idempotent yazılmalıdır, çünkü yeniden deneme aynı yükü ikinci kez gönderecektir. E-posta göndermeden önce aynı iş kimliği için gönderim kaydı var mı diye bakmak, dosya üretirken geçici ada yazıp sonra taşımak gibi kurallar zamanlanmış iş tasarımı yazısında ayrıntılıdır ve burada aynen geçerlidir.

Yeniden deneme, geri çekilme ve ölü iş

Yeniden deneme politikası, başarısız bir işin ne kadar sonra ve en çok kaç kez tekrar alınacağını belirler. Üstel geri çekilme, bekleme süresini her denemede ikiye katlar: geçici bir ağ hatası birkaç saniye sonra geçer, ancak çökmüş bir dış servis bir saat boyunca her dakika dövülmez. Sınıra ulaşan iş ölü olarak işaretlenir ve kuyruktan çıkar, ancak silinmez; bir kişi nedenini okuyup elle yeniden sıraya koyabilir.

from datetime import timedelta


def geri_cekilme(deneme):
    return min(3600, 30 * (2 ** (deneme - 1)))


def tamamla(con, is_id):
    con.execute("UPDATE isler SET durum = 'bitti', kilit_at = NULL WHERE id = ?", (is_id,))


def basarisiz(con, is_id, deneme, en_cok, hata):
    if deneme >= en_cok:
        con.execute(
            "UPDATE isler SET durum = 'olu', son_hata = ?, kilit_at = NULL WHERE id = ?",
            (hata, is_id),
        )
        return
    sonra = datetime.now(timezone.utc) + timedelta(seconds=geri_cekilme(deneme))
    con.execute(
        "UPDATE isler SET durum = 'bekliyor', son_hata = ?, kilit_at = NULL, "
        "calistir_at = ? WHERE id = ?",
        (hata, sonra.isoformat(timespec="seconds"), is_id),
    )

Hata metni kısaltılarak son_hata alanına yazılır; tam izleme kaydı loga gider. Bekleme süresi bir saatle sınırlandı, çünkü daha uzun gecikme kuyruğun takıldığı izlenimi verir ve bakım yapan kişiyi yanıltır. Yeniden denemeyi hak etmeyen hatalar da vardır: geçersiz bir e-posta adresi beş kez denenince düzelmez. İşleyici böyle bir hatayı ayrı bir istisna türüyle fırlatırsa, basarisiz fonksiyonu deneme sınırını beklemeden işi doğrudan ölü olarak işaretleyebilir.

İşletme: takılı işi kurtarmak, temizlik ve gözlem

Takılı iş kurtarma, bir işçi süreç işin ortasında öldürüldüğünde çalışıyor durumunda kalan satırları yeniden bekliyor durumuna döndüren bakım adımıdır. Kilit zamanı belirli bir süreyi aşmış her çalışıyor satırı takılı sayılır. Süre, en uzun meşru işin süresinden rahatça uzun seçilmelidir; aksi halde hâlâ çalışan bir iş ikinci kez alınır ve idempotency yoksa iki kez yapılır.

def takili_kurtar(con, sure_dk=30):
    sinir = (datetime.now(timezone.utc) - timedelta(minutes=sure_dk)).isoformat(timespec="seconds")
    return con.execute(
        "UPDATE isler SET durum = 'bekliyor', kilit_at = NULL, isci = NULL "
        "WHERE durum = 'calisiyor' AND kilit_at < ?",
        (sinir,),
    ).rowcount


def temizle(con, gun=7):
    sinir = (datetime.now(timezone.utc) - timedelta(days=gun)).isoformat(timespec="seconds")
    return con.execute(
        "DELETE FROM isler WHERE durum = 'bitti' AND olusturma < ?", (sinir,)
    ).rowcount


def ozet(con):
    return con.execute(
        "SELECT durum, COUNT(*), MIN(calistir_at) FROM isler GROUP BY durum"
    ).fetchall()

Takılı kurtarma ve bitmiş işlerin temizliği, işçinin kendi döngüsünde değil, cron ile saatte bir çalışan ayrı bir bakım komutunda durur. Bitti durumundaki satırları bir hafta sonra silmek tabloyu küçük tutar; ölü satırlar ise biri bakana kadar kalır. Bakım komutunun çıktısı loga yazılır, böylece kaç işin kurtarıldığı sonradan görülebilir.

Kuyruğun sağlığını üç sayı anlatır ve üçü de ozet fonksiyonundaki tek sorgudan çıkar:

• Durum başına satır sayısı: bekliyor durumunun sürekli artması işçi sayısının yetmediğini gösterir.
• En eski bekleyen işin yaşı: kuyruk gecikmesinin en doğrudan ölçüsüdür.
• Ölü iş sayısı: sıfırdan büyükse bir kişiye bildirim gider.

Bu sayıları mevcut heartbeat ya da log hattınıza eklemek, ayrı bir izleme aracı kurmadan kuyruğu görünür kılar. Ölü iş bildirimi gelmeden geçen her gün, kuyruğun değil işleyicilerin sağlam olduğunun kanıtıdır.

Sonuç

SQLite ile iş kuyruğu, tek sunuculu küçük bir otomasyon için yeniden deneme, kilit ve kurtarma davranışlarının tamamını ayrı bir servis işletmeden sağlar. Tasarımın özü dört durumlu bir tablo, BEGIN IMMEDIATE ile alınan iş, üstel geri çekilme ve takılı iş kurtarmadır; geri kalanı işleyicilerin idempotent yazılmasıdır.

İlk adım olarak elinizdeki en uzun toplu betiği seçin ve her kaydı bir iş olarak kuyruğa yazdırın; işçi aynı betik olsun, yalnız döngü kuyruktan okusun. Birkaç günlük logdan sonra hangi işlerin yeniden deneme gerektirdiğini görecek ve politikayı ona göre ayarlayacaksınız. Mevcut otomasyonlarınızı bu düzene taşırken bir yol arkadaşı isterseniz iletişim bölümünden yazabilirsiniz.