Python asyncio: async und await

Pythons asyncio-Modul ermöglicht das Schreiben von nebenläufigem Code in einem einzigen Thread mithilfe der Schlüsselwörter async und await. Anstatt zu blockieren, während auf Netzwerkantworten oder Dateilesevorgänge gewartet wird, unterbricht ein asyncio-Programm die wartende Task und wechselt sofort zu anderer Arbeit — es setzt fort, sobald das Ergebnis bereit ist. Damit ist es das richtige Werkzeug für I/O-gebundene Programme wie Web-Scraper, API-Clients und Chat-Server.

Dieses Kapitel behandelt:

• Was async-Funktionen (Coroutinen) sind und wie sie sich von regulären Funktionen unterscheiden
• Den Event Loop und wie asyncio Arbeit plant
• Ergebnisse erwarten, Tasks gleichzeitig ausführen mit asyncio.gather und asyncio.create_task
• Ausnahmen und Timeouts innerhalb von asynchronem Code behandeln
• Die asyncio.Queue für Producer-Consumer-Muster
• Wann asyncio zu verwenden ist und wann stattdessen auf threading zurückgegriffen werden sollte

Warum asyncio existiert

Betrachten Sie ein Programm, das zwei APIs nacheinander aufruft:

import time

def fetch(name, delay):
    time.sleep(delay)          # blocks the whole program
    return f'data from {name}'

start = time.perf_counter()
r1 = fetch('API A', 1)
r2 = fetch('API B', 1)
print(f'Done in {time.perf_counter() - start:.1f}s')
# Done in 2.0s

Beide Aufrufe laufen sequenziell — insgesamt 2 Sekunden, obwohl jeder Aufruf nur 1 Sekunde des Wartens benötigt. Mit asyncio pausiert das Programm fetch('API A', ...) während des Wartens, startet sofort fetch('API B', ...), und beide sind in etwa 1 Sekunde fertig:

import asyncio
import time

async def fetch(name, delay):
    await asyncio.sleep(delay)   # suspends only this coroutine
    return f'data from {name}'

async def main():
    start = time.perf_counter()
    r1, r2 = await asyncio.gather(fetch('API A', 1), fetch('API B', 1))
    print(f'Done in {time.perf_counter() - start:.1f}s')
    # Done in 1.0s

asyncio.run(main())

Coroutinen: async def und await

Eine mit async def definierte Funktion wird als Coroutinen-Funktion bezeichnet. Der Aufruf führt den Rumpf nicht sofort aus — sie gibt ein Coroutinen-Objekt zurück, das vom Event Loop gesteuert werden muss.

async def greet(name):
    print(f'Hello, {name}!')

# Calling it returns a coroutine object, nothing is printed yet
coro = greet('World')
print(type(coro))   # <class 'coroutine'>

# Run it properly
import asyncio
asyncio.run(greet('World'))
# Hello, World!

Innerhalb einer Coroutine unterbricht await die Ausführung, bis das Awaitable (eine weitere Coroutine, ein Task oder ein Future) ein Ergebnis liefert. Der Event Loop kann andere Coroutinen ausführen, während eine unterbrochen ist.

import asyncio

async def step_one():
    print('Step 1: start')
    await asyncio.sleep(1)     # suspend for 1 second
    print('Step 1: end')
    return 'result-1'

async def main():
    value = await step_one()   # wait for step_one to finish
    print(value)

asyncio.run(main())
# Step 1: start
# Step 1: end
# result-1

Was man erwarten kann

• Eine weitere async def-Coroutine
• Einen asyncio.Task (erstellt mit asyncio.create_task)
• Einen asyncio.Future
• Jedes Objekt mit einer __await__-Methode

Man kann await nicht außerhalb einer async def-Funktion verwenden.

Der Event Loop

Der Event Loop ist asyncios Scheduler. Er hält eine Warteschlange mit Coroutinen und Tasks, führt jede aus, bis sie auf ein await trifft, und wechselt dann zum nächsten bereiten Element. Typischerweise gibt es einen Event Loop pro Thread.

asyncio.run(coro) ist der standardmäßige Einstiegspunkt für asyncio-Programme. Er erstellt einen neuen Event Loop, führt die angegebene Coroutine bis zum Abschluss aus, schließt den Loop und gibt das Ergebnis zurück:

import asyncio

async def compute():
    await asyncio.sleep(0)   # yield control once
    return 6 * 7

result = asyncio.run(compute())
print(result)   # 42

Für die meisten Anwendungen muss der Loop nie direkt verwaltet werden — asyncio.run übernimmt Erstellung und Abbau.

Tasks gleichzeitig ausführen

asyncio.gather

asyncio.gather(*coroutines) plant alle angegebenen Coroutinen zur gleichzeitigen Ausführung und gibt ihre Ergebnisse in derselben Reihenfolge zurück:

import asyncio

async def fetch_data(name, delay):
    print(f'Start fetching {name}')
    await asyncio.sleep(delay)
    print(f'Done fetching {name}')
    return f'data from {name}'

async def main():
    results = await asyncio.gather(
        fetch_data('API A', 1),
        fetch_data('API B', 2),
        fetch_data('API C', 1),
    )
    print(results)

asyncio.run(main())
# Start fetching API A
# Start fetching API B
# Start fetching API C
# Done fetching API A
# Done fetching API C
# Done fetching API B
# ['data from API A', 'data from API B', 'data from API C']

Alle drei Coroutinen starten sofort. Die gesamte abgelaufene Zeit entspricht der langsamsten Coroutine (2 s), nicht der Summe (4 s).

asyncio.create_task

asyncio.create_task(coro) kapselt eine Coroutine in einem Task und plant ihre baldige Ausführung. Anders als gather startet das Erstellen eines Tasks ihn im Hintergrund, während die aktuelle Coroutine weiterläuft:

import asyncio

async def background_job(name, delay):
    print(f'{name}: start')
    await asyncio.sleep(delay)
    print(f'{name}: end')
    return f'{name} done'

async def main():
    t1 = asyncio.create_task(background_job('Task A', 1))
    t2 = asyncio.create_task(background_job('Task B', 2))

    # Both tasks are already scheduled; await collects their results
    result1 = await t1
    result2 = await t2
    print(result1, result2)

asyncio.run(main())
# Task A: start
# Task B: start
# Task A: end
# Task B: end
# Task A done Task B done

Verwenden Sie create_task, wenn eine Task sofort starten soll und Sie planen, ihr Ergebnis später abzurufen (oder sie abzubrechen). Verwenden Sie gather, wenn Sie eine feste Gruppe von Coroutinen starten und auf alle gemeinsam warten möchten.

Verschachtelte Ausgabe

Eine nützliche Methode, den Event Loop in Aktion zu beobachten, ist zu sehen, wie Tasks sich verschachteln:

import asyncio

async def count_down(name, seconds):
    for i in range(seconds, 0, -1):
        print(f'{name}: {i}')
        await asyncio.sleep(1)
    print(f'{name}: done!')

async def main():
    await asyncio.gather(
        count_down('Task A', 3),
        count_down('Task B', 2),
    )

asyncio.run(main())
# Task A: 3
# Task B: 2
# Task A: 2
# Task B: 1
# Task A: 1
# Task B: done!
# Task A: done!

Beide Tasks teilen sich einen Thread; der Event Loop wechselt bei jedem await asyncio.sleep(1) zwischen ihnen.

Ausnahmen behandeln

Ausnahmen, die innerhalb einer Coroutine ausgelöst werden, propagieren durch await genau wie in synchronem Code. Verwenden Sie einen regulären try/except-Block:

import asyncio

async def risky_task():
    await asyncio.sleep(0.1)
    raise ValueError('something went wrong')

async def main():
    try:
        await risky_task()
    except ValueError as e:
        print(f'Caught: {e}')

asyncio.run(main())
# Caught: something went wrong

Wenn bei asyncio.gather eine Coroutine eine Ausnahme auslöst, werden die anderen standardmäßig nicht abgebrochen, aber die Ausnahme wird erneut ausgelöst, wenn Sie den gather-Aufruf awaiten. Übergeben Sie return_exceptions=True, um Ausnahmen stattdessen als Rückgabewerte zu sammeln:

import asyncio

async def good():
    return 'ok'

async def bad():
    raise RuntimeError('oops')

async def main():
    results = await asyncio.gather(good(), bad(), return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f'Error: {r}')
        else:
            print(f'Result: {r}')

asyncio.run(main())
# Result: ok
# Error: oops

Timeouts mit asyncio.wait_for

asyncio.wait_for(coro, timeout) führt eine Coroutine aus und bricht sie ab, wenn sie nicht innerhalb der angegebenen Anzahl von Sekunden abgeschlossen wird, und löst asyncio.TimeoutError aus:

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return 42

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=1.0)
        print(result)
    except asyncio.TimeoutError:
        print('Timed out — operation cancelled')

asyncio.run(main())
# Timed out — operation cancelled

Dies ist wichtig für Produktions-Netzwerkcode, bei dem ein hängender Server sonst eine Task auf unbestimmte Zeit blockieren würde.

asyncio.Queue für Producer-Consumer-Muster

asyncio.Queue ist eine thread-sichere, async-bewusste Warteschlange. Sie eignet sich ideal zur Entkopplung von Produzenten (Code, der Arbeit generiert) und Konsumenten (Code, der sie verarbeitet):

import asyncio

async def producer(queue):
    for i in range(1, 4):
        print(f'Produced item {i}')
        await queue.put(i)
        await asyncio.sleep(0.1)
    await queue.put(None)   # sentinel to signal consumers to stop

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f'Consumed item {item}')

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())
# Produced item 1
# Consumed item 1
# Produced item 2
# Consumed item 2
# Produced item 3
# Consumed item 3

Verwenden Sie für mehrere Konsumenten queue.task_done() und queue.join(), um zu wissen, wann alle Elemente verarbeitet wurden.

asyncio vs. threading

Sowohl asyncio als auch das Python-threading-Modul ermöglichen es, Arbeit gleichzeitig fortzusetzen, tun dies aber auf unterschiedliche Weise:

Verwenden Sie asyncio, wenn Sie den I/O-Code steuern und async-kompatible Bibliotheken nutzen können (z. B. aiohttp, asyncpg). Verwenden Sie threading, wenn Sie auf Drittanbieter-Blocking-Bibliotheken angewiesen sind, die nicht asynchron gemacht werden können.

Für echte CPU-Parallelität greifen Sie stattdessen auf multiprocessing oder concurrent.futures.ProcessPoolExecutor zurück.

Häufige Fallstricke

await vergessen: Der Aufruf einer async-Funktion ohne await gibt ein Coroutinen-Objekt zurück und tut nichts. Python gibt eine RuntimeWarning: coroutine '...' was never awaited aus, um dabei zu helfen, dies zu erkennen.

async def main():
    asyncio.sleep(1)   # BUG: returns a coroutine, does not sleep
    await asyncio.sleep(1)   # correct

Den Event Loop blockieren: Das Ausführen von langsamem synchronem Code (eine enge Schleife, ein blockierender Netzwerkaufruf, time.sleep) innerhalb einer Coroutine friert den gesamten Event Loop ein. Umschließen Sie blockierende Aufrufe mit asyncio.to_thread (Python 3.9+), um sie in einem Thread-Pool auszuführen, ohne zu blockieren:

import asyncio
import time

def blocking_task():
    time.sleep(2)   # simulates a slow blocking operation
    return 'done'

async def main():
    result = await asyncio.to_thread(blocking_task)
    print(result)

asyncio.run(main())
# done

asyncio.run innerhalb eines laufenden Loops verwenden: Jupyter-Notebooks führen bereits einen Event Loop aus. Verwenden Sie await coro direkt in Notebook-Zellen, oder installieren Sie nest_asyncio, um verschachtelte Loops zu erlauben.

Kurzreferenz

Übungen