Python asyncio: async, await, TaskGroup ve Zaman Aşımı

Coroutine nedir, nasıl çalıştırılır?

async def ile tanımlanan fonksiyona coroutine fonksiyonu, onu çağırınca dönen nesneye coroutine nesnesi denir. Resmî belge de bu iki kavramın aynı adla anıldığını not eder. Kritik nokta şudur: coroutine fonksiyonunu çağırmak gövdeyi çalıştırmaz, yalnızca bir coroutine nesnesi üretir. Onu çalıştırmanın üç yolu vardır: en üst düzeyde asyncio.run() ile, başka bir coroutine içinde await ile ya da asyncio.create_task() veya TaskGroup ile görev olarak zamanlayarak.

import asyncio

async def selam():
    print('merhaba')
    await asyncio.sleep(0.1)
    print('dünya')
    return 42

c = selam()
print(type(c).__name__)
c.close()
print(asyncio.run(selam()))

# çıktı: coroutine
# çıktı: merhaba
# çıktı: dünya
# çıktı: 42

asyncio.run() verilen awaitable'ı bir olay döngüsünde çalıştırır ve sonucunu döndürür; döngüyü oluşturma, asenkron generator'ları sonlandırma ve executor'ı kapatma işlerini de kendisi üstlenir. Aynı iş parçacığında başka bir olay döngüsü zaten çalışıyorsa çağrılamaz. Genellikle programın tek giriş noktası olan bir main() coroutine'i için bir kez kullanılır. Örnekteki c.close() çağrısı, hiç çalıştırılmayan coroutine için uyarı üretilmesini engellemek içindir.

Bir coroutine'i çağırıp await etmeyi unutursanız ne olur? asyncio geliştirme belgesine göre, ne beklenen ne de görev olarak zamanlanan bir coroutine için RuntimeWarning yayımlanır. Aşağıdaki örnekte test() gövdesi hiç çalışmıyor; stderr'de görünen uyarıyı (dosya yolu kısaltılarak) yorum olarak gösterdik.

import asyncio

async def test():
    print('hiç çalışmadı')

async def main():
    test()

asyncio.run(main())

# stderr: ornek.py:9: RuntimeWarning: coroutine 'test' was never awaited
# stderr:   test()
# stderr: RuntimeWarning: Enable tracemalloc to get the object allocation traceback

Olay döngüsü: işbirlikçi çoklu görev

Resmî kavramsal genel bakış belgesi olay döngüsünü bir orkestra şefine benzetir: döngü çalıştırılacak işlerden oluşan bir koleksiyon tutar, birini seçer ve kontrolü ona verir; iş duraklayınca ya da bitince kontrol döngüye döner ve sıradaki iş seçilir. Bekleyen iş yoksa döngü boşuna işlemci harcamaz, bir G/Ç işlemi tamamlanana ya da bir zamanlayıcı dolana kadar bekler.

Bu model işbirliğine dayanır. Görevler yalnızca await noktalarında kontrolü bırakır. Belgede de vurgulandığı gibi, kontrolü bırakmayan "açgözlü" bir iş diğer işleri aç bırakır ve yaklaşımın bütün faydasını ortadan kaldırır. asyncio geliştirme belgesi de aynı uyarıyı yapar: olay döngüsü bir iş parçacığında çalışır ve bir görev çalışırken aynı iş parçacığındaki başka hiçbir görev çalışamaz. İşlemciyi bir saniye meşgul eden bir hesaplama, bütün görevleri ve G/Ç işlemlerini bir saniye geciktirir.

Sıralı ve eşzamanlı bekleme

await ile bir coroutine'i doğrudan beklerseniz işler sırayla yürür: ikinci bekleme, birincisi bitmeden başlamaz. Eşzamanlılık için coroutine'leri görev (Task) olarak zamanlamanız gerekir. Aşağıdaki örnekte aynı iki bekleme önce sırayla, sonra bir TaskGroup içinde yapılıyor. Sıralı sürüm yaklaşık 0.5 + 1.0 = 1.5 saniye, eşzamanlı sürüm ise en uzun bekleme kadar, yani yaklaşık 1.0 saniye sürüyor.

import asyncio
import time

async def bekle(sure, ad):
    await asyncio.sleep(sure)
    return ad

async def sirali():
    await bekle(0.5, 'a')
    await bekle(1.0, 'b')

async def es_zamanli():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(bekle(0.5, 'a'))
        tg.create_task(bekle(1.0, 'b'))

for ana in (sirali, es_zamanli):
    t0 = time.perf_counter()
    asyncio.run(ana())
    print(ana.__name__, round(time.perf_counter() - t0, 1))

# çıktı: sirali 1.5
# çıktı: es_zamanli 1.0

Burada iş parçacığı yok; iki bekleme aynı iş parçacığında örtüşüyor. Bu yüzden asyncio, işin büyük kısmı beklemekten oluşan durumlarda (HTTP istekleri, veritabanı sorguları, soket iletişimi) faydalıdır. İşlemci yoğun hesaplamaları hızlandırmaz; bunun için Python threading ve multiprocessing yazısında anlatılan süreç havuzları daha uygundur.

Awaitable türleri: coroutine, Task ve Future

Belgeye göre await ifadesinde kullanılabilen nesnelere awaitable denir ve üç ana türü vardır. Coroutine'ler doğrudan beklenebilir. Task'lar, coroutine'leri eşzamanlı çalışmak üzere zamanlar; create_task() ile oluşturulan görev "kısa süre içinde" çalışmaya başlar. Future ise bir asenkron işlemin nihai sonucunu temsil eden düşük seviyeli bir nesnedir; geri çağırma (callback) tabanlı kodu async/await ile kullanabilmek için vardır ve belgeye göre uygulama kodunda genellikle Future oluşturmanız gerekmez.

gather(): sonuçları sırayla toplamak

asyncio.gather() verilen awaitable'ları eşzamanlı çalıştırır; coroutine'leri otomatik olarak görev yapar. Hepsi başarılı olursa sonuç, sonuçların verilme sırasıyla dizildiği bir listedir; tamamlanma sırası önemli değildir. Aşağıdaki örnekte görevler b, c, a sırasıyla bitiyor ama liste a, b, c sırasında dönüyor.

import asyncio

async def getir(ad, sure):
    await asyncio.sleep(sure)
    print(f'{ad} hazır')
    return ad.upper()

async def main():
    sonuc = await asyncio.gather(getir('a', 0.3), getir('b', 0.1), getir('c', 0.2))
    print(sonuc)

asyncio.run(main())

# çıktı: b hazır
# çıktı: c hazır
# çıktı: a hazır
# çıktı: ['A', 'B', 'C']

gather() hata durumunda dikkat ister. return_exceptions=False (varsayılan) iken ilk istisna hemen bekleyen göreve iletilir, ama belgenin kalın harflerle vurguladığı gibi diğer awaitable'lar iptal edilmez ve çalışmaya devam eder. return_exceptions=True verilirse istisnalar da başarılı sonuçlar gibi listeye konur. Belge, daha güçlü güvenlik garantileri için TaskGroup'u önerir: bir görev ya da alt görev hata verdiğinde TaskGroup kalan görevleri iptal eder, gather etmez.

TaskGroup: yapılandırılmış eşzamanlılık (3.11+)

asyncio.TaskGroup, Python 3.11'de eklenen bir asenkron context manager'dır. Görevleri tg.create_task() ile gruba eklersiniz ve async with bloğundan çıkılırken hepsi beklenir; ayrıca await yazmanız gerekmez. Görevlerden biri CancelledError dışında bir istisnayla başarısız olursa grubun kalan görevleri iptal edilir. Bütün görevler bittikten sonra hatalar bir ExceptionGroup içinde toplanıp yükseltilir; bunları except* ile türüne göre yakalayabilirsiniz. İstisna grupları Python raise ve özel istisna yazısında anlatılıyor.

import asyncio

async def gorev(ad, sure, hata=False):
    try:
        await asyncio.sleep(sure)
        if hata:
            raise ValueError(f'{ad} başarısız')
        print(f'{ad} tamam')
    except asyncio.CancelledError:
        print(f'{ad} iptal edildi')
        raise

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(gorev('kısa', 0.1))
            tg.create_task(gorev('hatalı', 0.2, hata=True))
            tg.create_task(gorev('uzun', 1))
    except* ValueError as eg:
        for e in eg.exceptions:
            print('yakalandı:', e)

asyncio.run(main())

# çıktı: kısa tamam
# çıktı: uzun iptal edildi
# çıktı: yakalandı: hatalı başarısız

Çıktıyı adım adım okuyalım: "kısa" 0.1 saniyede biter; "hatalı" 0.2 saniyede ValueError yükseltir; bunun üzerine grup, henüz bitmemiş "uzun" görevini iptal eder ve görevin except asyncio.CancelledError bloğu çalışır. Son olarak hata ExceptionGroup içinde dışarı çıkar. Belgeye göre KeyboardInterrupt ve SystemExit özel durumlardır: grup yine kalan görevleri iptal eder ama istisna grubu yerine orijinal istisnayı yeniden yükseltir.

Python 3.15: TaskGroup.cancel() ile erken çıkış

Python 3.15 TaskGroup nesnesine cancel() yöntemini ekledi. What's New belgesine göre bu yöntem, görevlerin amacına ulaşıldığında grubu erkenden ve hata üretmeden sonlandırmak için tasarlandı; önceden bunun için özel bir istisna yükseltip yutmak gibi dolambaçlı yollar gerekiyordu. Yöntem bitmemiş tüm görevleri ve grubun gövdesini iptal eder, ancak async with bloğu CancelledError yükseltmeden çıkar. Aşağıdaki örnekte üç yansıdan ilk yanıt veren kazanıyor, diğerleri iptal ediliyor. Bu örnek 3.14 ve öncesinde çalışmaz.

import asyncio

async def ara(ad, sure, tg, bulunan):
    await asyncio.sleep(sure)
    print(f'{ad} yanıt verdi')
    bulunan.append(ad)
    tg.cancel()

async def main():
    bulunan = []
    async with asyncio.TaskGroup() as tg:
        for ad, sure in [('ayna1', 0.3), ('ayna2', 0.1), ('ayna3', 0.2)]:
            tg.create_task(ara(ad, sure, tg, bulunan))
    print('kazanan:', bulunan)

asyncio.run(main())

# çıktı: ayna2 yanıt verdi
# çıktı: kazanan: ['ayna2']

Zaman aşımı: timeout() ve wait_for()

Bir işlemin ne kadar bekleneceğini sınırlamak için Python 3.11'de eklenen asyncio.timeout() asenkron context manager'ını kullanabilirsiniz. Süre dolarsa geçerli görev iptal edilir ve oluşan CancelledError, context manager tarafından TimeoutError'a dönüştürülür. Belgenin altını çizdiği ayrıntı şudur: bu dönüşümü context manager yaptığı için TimeoutError yalnızca async with bloğunun dışında yakalanabilir. Tek bir awaitable için asyncio.wait_for(aw, timeout=...) da aynı işi görür; 3.11'den beri asyncio.TimeoutError yerine yerleşik TimeoutError yükseltir.

import asyncio

async def yavas():
    await asyncio.sleep(5)
    return 'bitti'

async def main():
    try:
        async with asyncio.timeout(0.2):
            await yavas()
    except TimeoutError:
        print('timeout() ile zaman aşımı')

    try:
        await asyncio.wait_for(yavas(), timeout=0.2)
    except TimeoutError:
        print('wait_for() ile zaman aşımı')

asyncio.run(main())

# çıktı: timeout() ile zaman aşımı
# çıktı: wait_for() ile zaman aşımı

İptal (cancellation) kuralları

Bir görev iptal edildiğinde, ilk fırsatta (genellikle bir sonraki await noktasında) içinde asyncio.CancelledError yükseltilir. Belge, temizlik işlerinin try/finally ile yapılmasını önerir. CancelledError'ı açıkça yakalarsanız, temizlikten sonra genellikle yeniden yükseltmelisiniz; a4 örneğindeki raise satırı bu yüzden oradadır. Bu istisna doğrudan BaseException'dan türediği için sıradan except Exception blokları onu yakalamaz.

Bu kural neden bu kadar önemli? TaskGroup ve timeout() gibi yapılandırılmış eşzamanlılık araçları içeride iptal mekanizmasıyla çalışır. Belgeye göre CancelledError'ı yutan bir coroutine bu araçların yanlış davranmasına neden olabilir. İptali gerçekten bastırmanız gerekiyorsa, iptal durumunu tamamen temizlemek için uncancel() çağırmanız da gerekir; ama bu, uygulama kodunda nadiren doğru tercihtir.

Engelleyen kod ve asyncio.to_thread()

Olay döngüsünün en büyük düşmanı engelleyen (blocking) çağrılardır. Bir coroutine içinde time.sleep(), senkron bir HTTP kütüphanesi ya da büyük bir dosyanın senkron okunması, o süre boyunca bütün döngüyü durdurur. Bu tür çağrılar için Python 3.9'da eklenen asyncio.to_thread() fonksiyonu, verilen fonksiyonu ayrı bir iş parçacığında çalıştırır ve sonucunu beklenebilir bir coroutine olarak döndürür. Aşağıdaki örnekte engelleyen 0.5 saniyelik iş ile 0.5 saniyelik asenkron bekleme örtüşüyor ve toplam süre yaklaşık 0.5 saniye oluyor.

import asyncio
import time

def engelleyen_is():
    time.sleep(0.5)
    return 'dosya okundu'

async def main():
    t0 = time.perf_counter()
    sonuc = await asyncio.gather(asyncio.to_thread(engelleyen_is), asyncio.sleep(0.5, 'zamanlayıcı'))
    print(sonuc, round(time.perf_counter() - t0, 1))

asyncio.run(main())

# çıktı: ['dosya okundu', 'zamanlayıcı'] 0.5

Belge önemli bir sınırı da not eder: GIL nedeniyle to_thread() genellikle yalnızca G/Ç ağırlıklı fonksiyonları engellemesiz yapmak için kullanılabilir. GIL'i serbest bırakan eklenti modüllerinde ya da GIL'siz Python yapılarında işlemci ağırlıklı fonksiyonlar için de kullanılabilir. İşlemci yoğun işler için loop.run_in_executor() ile bir ProcessPoolExecutor kullanmak bir seçenektir.

Eşzamanlılığı sınırlamak: Semaphore

Yüzlerce isteği aynı anda başlatmak hem karşı sunucuyu hem kendi programınızı zorlayabilir. asyncio.Semaphore(n) içindeki sayaç her girişte azalır, her çıkışta artar; sıfıra ulaştığında yeni giriş, bir görev çıkana kadar bekler. async with sinir: bloğu bu işlemleri otomatik yapar. Belge bu nesnenin iş parçacığı açısından güvenli olmadığını belirtir; aynı olay döngüsündeki görevler arasında kullanılmak üzere tasarlanmıştır.

import asyncio

sinir = asyncio.Semaphore(2)
aktif = 0
en_fazla = 0

async def indir(i):
    global aktif, en_fazla
    async with sinir:
        aktif += 1
        en_fazla = max(en_fazla, aktif)
        await asyncio.sleep(0.1)
        aktif -= 1
    return i * 10

async def main():
    sonuc = await asyncio.gather(*(indir(i) for i in range(6)))
    print(sonuc)
    print('aynı anda en fazla:', en_fazla)

asyncio.run(main())

# çıktı: [0, 10, 20, 30, 40, 50]
# çıktı: aynı anda en fazla: 2

Üretici ve tüketici: asyncio.Queue

Görevler arasında veri aktarmak için asyncio.Queue kullanılır. await kuyruk.put() ve await kuyruk.get() çağrıları gerektiğinde bekler, böylece üretici ve tüketici birbirine uyum sağlar. Aşağıdaki örnekte üretici işin bittiğini None göndererek bildiriyor; tüketici bu işareti görünce döngüden çıkıyor.

import asyncio

async def uretici(kuyruk):
    for i in range(3):
        await kuyruk.put(f'iş-{i}')
    await kuyruk.put(None)

async def tuketici(kuyruk):
    while (oge := await kuyruk.get()) is not None:
        print("işlendi:", oge)

async def main():
    kuyruk = asyncio.Queue()
    async with asyncio.TaskGroup() as tg:
        tg.create_task(uretici(kuyruk))
        tg.create_task(tuketici(kuyruk))

asyncio.run(main())

# çıktı: işlendi: iş-0
# çıktı: işlendi: iş-1
# çıktı: işlendi: iş-2

Asenkron generator ve async for

async def gövdesinde yield kullanırsanız asenkron generator elde edersiniz. Bunlar async for döngüsüyle ve asenkron comprehension ile tüketilir. Language Reference, asenkron generator break, iptal ya da başka bir istisna nedeniyle erken bırakılırsa temizlik kodunun beklenmedik bir bağlamda çalışabileceğini, bunu önlemek için aclose() yönteminin açıkça çağrılması gerektiğini belirtir. Senkron generator'lar Python generator ve yield yazısında anlatılıyor.

import asyncio

async def sayac(n):
    for i in range(n):
        await asyncio.sleep(0)
        yield i

async def main():
    print([i async for i in sayac(3)])
    async for i in sayac(2):
        print('değer', i)

asyncio.run(main())

# çıktı: [0, 1, 2]
# çıktı: değer 0
# çıktı: değer 1

Python 3.14 ve 3.15'teki değişiklikler

Python 3.14 What's New belgesine göre asyncio.get_event_loop() artık geçerli bir olay döngüsü yoksa örtük olarak yeni döngü oluşturmuyor, RuntimeError yükseltiyor. Belge bu fonksiyonu kullanan desenlerin çoğunun asyncio.run() ile değiştirilebileceğini söyler. Aynı sürümde olay döngüsü politikası sistemi (get_event_loop_policy(), set_event_loop_policy() ve politika sınıfları) kullanımdan kaldırıldı; belirli bir döngü uygulaması gerekiyorsa asyncio.run() ya da asyncio.Runner ile loop_factory argümanı kullanılmalı.

import asyncio

try:
    asyncio.get_event_loop()
except RuntimeError as e:
    print(type(e).__name__ + ':', e)

# çıktı: RuntimeError: There is no current event loop in thread 'MainThread'.

3.14 ayrıca çalışan bir süreçteki asyncio görevlerini incelemek için python -m asyncio ps PID ve python -m asyncio pstree PID komutlarını, çağrı grafiğini yazdıran capture_call_graph() ve print_call_graph() fonksiyonlarını ekledi; create_task() artık tüm anahtar kelime argümanlarını görev fabrikasına iletiyor. Python 3.15'te ise yukarıda gördüğünüz TaskGroup.cancel() geldi. Ayrıntılar için Python 3.15 yenilikleri yazısına bakabilirsiniz.

Hata ayıklama modu

Geliştirme sırasında asyncio'nun hata ayıklama modunu açmak birçok sorunu erkenden gösterir. Belgeye göre modu PYTHONASYNCIODEBUG=1 ortam değişkeniyle, Python Geliştirme Modu (-X dev) ile, asyncio.run(main(), debug=True) ile ya da loop.set_debug() ile açabilirsiniz. Bu modda yanlış iş parçacığından çağrılan güvensiz API'ler istisna yükseltir, uzun süren G/Ç seçici çağrıları ve 100 milisaniyeden uzun süren geri çağırmalar loglanır. asyncio logger'ının seviyesini DEBUG yapmak için Python logging modülü yazısına bakabilirsiniz.

PYTHONASYNCIODEBUG=1 python main.py
python -X dev main.py

Sık yapılan hatalar

• await'i unutmak: veri = getir() bir coroutine nesnesi verir, sonucu değil; gövde hiç çalışmaz ve RuntimeWarning görürsünüz.
• Coroutine içinde time.sleep() kullanmak: Bütün olay döngüsünü durdurur. await asyncio.sleep() ya da engelleyen işler için asyncio.to_thread() kullanın.
• asyncio.run() içinden asyncio.run() çağırmak: Çalışan bir döngü varken çağrılamaz; iç coroutine'i await edin.
• create_task() sonucunu saklamamak: Görev bitmeden çöp toplanabilir; TaskGroup kullanın veya referansı tutun.
• CancelledError'ı yutmak: TaskGroup ve timeout() doğru çalışmaz; temizlikten sonra yeniden yükseltin.
• TimeoutError'ı timeout() bloğunun içinde yakalamaya çalışmak: Dönüşüm bloktan çıkarken yapılır; try bloğunu async with'in dışına koyun.
• get_event_loop() ile döngü oluşturmayı beklemek: 3.14'ten beri RuntimeError verir; asyncio.run() kullanın.

Sık Sorulan Sorular

asyncio mu, threading mi?

İkisi de G/Ç ağırlıklı işlerde eşzamanlılık sağlar. asyncio tek iş parçacığında çalışır ve kontrolün nerede bırakılacağı await ile açıkça görünür; ancak kullandığınız kütüphanelerin de asenkron olması gerekir. threading mevcut senkron kodla doğrudan çalışır ama paylaşılan veriyi kilitlerle korumanızı gerektirir. threading belgesi de asyncio'yu, birden çok işletim sistemi iş parçacığı kullanmadan görev düzeyinde eşzamanlılık sağlayan alternatif bir yaklaşım olarak anar.

Senkron bir fonksiyondan coroutine çağırabilir miyim?

await yalnızca async def içinde kullanılabilir. Senkron koddan bir coroutine'i çalıştırmanın yolu asyncio.run()'dır. Başka bir iş parçacığından, çalışan bir döngüye coroutine göndermek için asyncio.run_coroutine_threadsafe() kullanılır ve bir concurrent.futures.Future döner.

Jupyter veya çalışan bir döngü içindeyken ne yapmalıyım?

Bir olay döngüsü zaten çalışıyorsa asyncio.run() çağrılamaz. Bu ortamlarda en üst düzeyde doğrudan await yazmanıza izin veriliyorsa onu kullanın. Python'un kendi asenkron REPL'i python -m asyncio ile açılır ve en üst düzeyde await kullanılmasına izin verir.

Web çatıları asyncio kullanıyor mu?

Bazı çatılar asyncio üzerine kuruludur; bunlar üçüncü taraf paketlerdir ve kendi belgeleri vardır. Örneğin FastAPI uygulamasını Uvicorn ve Nginx ile yayına alma ve discord.py ile bot yapımı yazılarında asenkron uygulamaların sunucuda nasıl çalıştırıldığını görebilirsiniz.

Marka adları sahiplerine aittir. Bu yazı bağımsız bir teknik rehberdir; adı geçen kuruluşlarla bayilik veya ortaklık ilişkisi bulunduğu anlamına gelmez.

İlgili içerikler: Python öğrenme rehberi · threading ve multiprocessing · Generator ve yield · raise, ExceptionGroup ve except* · with ve context manager · Python 3.15 yenilikleri