asyncio: eşzamansız kod, coroutine, await ve gather

Derinleş: yapılandırılmış eşzamanlılık, iptal, kuyruk ve dayanıklılık

gather, create_task, Semaphore ve wait_for ile birden çok işi aynı anda yürütebiliyorsun. Üretimdeki bir async servis ise daha fazlasını ister: görevlerin hata durumunda birlikte iptal edilmesi, ilk gelen cevabı alıp diğerlerini durdurmak, iptal edilen işin arkasını temizlemesi, sabit sayıda işçiyle kuyruk işlemek, paylaşılan durumu korumak ve geçici hatalarda yeniden denemek. Bu bölüm bu kalıpları tamamlıyor.

Kendi bilgisayarında: kurulum ve sık hatalar

asyncio standart kütüphanededir, kurulum gerekmez. Gerçek HTTP istekleri için async destekli bir istemci kurarsın; requests senkrondur ve olay döngüsünü bloklar.

python -m pip install httpx          # ya da: aiohttp

import asyncio, httpx

async def main():
    async with httpx.AsyncClient(timeout=5) as istemci:
        cevaplar = await asyncio.gather(
            *(istemci.get(f"https://api.ornek.com/musteri/{i}") for i in range(10)))
    print([c.status_code for c in cevaplar])

asyncio.run(main())        # betikte standart giriş noktası

TaskGroup: birlikte başla, birlikte bit

async with asyncio.TaskGroup() as tg: bloğunda tg.create_task(...) ile açılan tüm görevler blok bitmeden beklenir. Bir görev hata verirse grup diğerlerini iptal eder ve hataları ExceptionGroup içinde fırlatır; except* ile türüne göre yakalanır. gather'dan farkı, "biri patladı ama diğerleri arka planda çalışmaya devam ediyor" durumunun olmamasıdır.

import asyncio

async def sorgu(sistem: str, sure: float, hata: bool = False) -> str:
    await asyncio.sleep(sure)
    if hata:
        raise ConnectionError(f"{sistem} cevap vermedi")
    return f"{sistem}: tamam"

async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(sorgu("kimlik", 0.02))
        t2 = tg.create_task(sorgu("skor", 0.01))
    print(t1.result(), "|", t2.result())

    try:
        async with asyncio.TaskGroup() as tg:
            yavas = tg.create_task(sorgu("kara_liste", 0.5))
            tg.create_task(sorgu("skor", 0.01, hata=True))
    except* ConnectionError as eg:
        print("hata:", eg.exceptions[0])
    print("yavaş görev iptal edildi mi?", yavas.cancelled())

loop = asyncio.new_event_loop()
loop.run_until_complete(main())
loop.close()

İlk biteni almak: as_completed ve wait

asyncio.as_completed(gorevler) sonuçları bitiş sırasına göre verir; uzun süren bir listede ilk gelenleri ekrana basmak için idealdir. asyncio.wait(gorevler, return_when=asyncio.FIRST_COMPLETED) ise (bitenler, bekleyenler) kümelerini döndürür; aynı veriyi üç yedek sunucudan isteyip en hızlısını almak ve diğerlerini iptal etmek tipik kullanımdır.

import asyncio

async def kur(kaynak: str, sure: float) -> str:
    await asyncio.sleep(sure)
    return kaynak

async def main():
    for gelecek in asyncio.as_completed([kur("A", 0.05), kur("B", 0.01), kur("C", 0.03)]):
        print("geldi:", await gelecek)

    gorevler = [asyncio.create_task(kur(k, s)) for k, s in [("ana", 0.2), ("yedek", 0.02)]]
    biten, bekleyen = await asyncio.wait(gorevler, return_when=asyncio.FIRST_COMPLETED)
    print("en hızlı:", biten.pop().result(), "| bekleyen:", len(bekleyen))
    for g in bekleyen:
        g.cancel()
    await asyncio.gather(*bekleyen, return_exceptions=True)   # iptalin bitmesini bekle

loop = asyncio.new_event_loop()
loop.run_until_complete(main())
loop.close()

İptal (cancellation) ve temizlik

task.cancel() görevin beklediği noktaya CancelledError fırlatır. Görev try/finally ile bağlantıyı kapatmak, geçici dosyayı silmek gibi temizliği yapabilir. CancelledError'ı yakalayıp yutma: except Exception onu yakalamaz (BaseException alt sınıfıdır) ama çıplak except: yakalar ve iptal sessizce bozulur. Yakalarsan temizlikten sonra raise ile yeniden fırlat.

Kuyruk ve sabit işçi havuzu

10.000 işi tek seferde gather'a vermek 10.000 görev nesnesi demektir. asyncio.Queue ile N işçi kurup işleri kuyruktan çektirmek bellek ve eşzamanlılığı birlikte sınırlar. Her iş bittiğinde task_done(), üretici tarafında await kuyruk.join() "tüm işler işlendi" anını verir; sonra sonsuz döngüdeki işçiler iptal edilir.

import asyncio

async def isci(ad: str, kuyruk: asyncio.Queue, sonuc: list):
    while True:
        siparis = await kuyruk.get()
        try:
            await asyncio.sleep(0.01)               # kargo API çağrısı
            sonuc.append((siparis, ad))
        finally:
            kuyruk.task_done()

async def main():
    kuyruk = asyncio.Queue()
    for no in range(1, 9):
        kuyruk.put_nowait(no)
    sonuc = []
    iscilar = [asyncio.create_task(isci(f"i{k}", kuyruk, sonuc)) for k in range(3)]
    await kuyruk.join()                             # hepsi işlendi
    for t in iscilar:
        t.cancel()
    await asyncio.gather(*iscilar, return_exceptions=True)
    print(len(sonuc), "sipariş,", len({ad for _, ad in sonuc}), "işçi kullandı")

loop = asyncio.new_event_loop()
loop.run_until_complete(main())
loop.close()

Tek iş parçacığında da yarış olur: asyncio.Lock

asyncio tek iş parçacığında çalışır, ama her await bir geçiş noktasıdır. Bakiyeyi okuyup await edip sonra yazarsan, araya giren başka bir görev aynı eski değeri okur ve bir güncelleme kaybolur. Okuma-bekleme-yazma bloğunu async with kilit: içine al. Bir görevin diğerine "hazırım" demesi için asyncio.Event kullanılır.

Yeniden deneme ve üstel geri çekilme

Ağ hataları çoğu zaman geçicidir. Doğru tepki, hemen tekrar saldırmak değil, giderek artan aralıklarla (0.1, 0.2, 0.4 s; üstel geri çekilme, exponential backoff) birkaç kez denemek ve sonunda hatayı yukarı iletmektir. Yalnızca geçici hatalarda (bağlantı, zaman aşımı) yeniden dene; "yetkisiz" ya da "geçersiz istek" gibi kalıcı hataları tekrarlamak işe yaramaz.

import asyncio

deneme_sayisi = 0

async def kararsiz_servis() -> str:
    global deneme_sayisi
    deneme_sayisi += 1
    await asyncio.sleep(0)
    if deneme_sayisi < 3:
        raise ConnectionError("geçici hata")
    return "skor: 712"

async def tekrar_dene(fabrika, deneme=4, bekleme=0.01):
    for i in range(deneme):
        try:
            return await fabrika()
        except ConnectionError as e:
            if i == deneme - 1:
                raise
            sure = bekleme * 2 ** i
            print(f"{i + 1}. deneme başarısız ({e}), {sure:.2f} s bekleniyor")
            await asyncio.sleep(sure)

async def main():
    print(await tekrar_dene(kararsiz_servis))

loop = asyncio.new_event_loop()
loop.run_until_complete(main())
loop.close()