Asynchron mit asyncio: Coroutinen, await, gather

Verstehe die kooperative Nebenläufigkeit von asyncio: lazy Coroutinen, Unterbrechungspunkte, gather, TaskGroup und die Fallstricke blockierender Aufrufe.

10 Min. · 3 Fragen

asyncio setzt kooperative Nebenläufigkeit auf einem einzigen Thread um: Eine Event-Loop orchestriert Coroutinen, die bei jedem await freiwillig die Kontrolle abgeben. Ein grundlegender, oft missverstandener Punkt: Eine async def-Funktion aufzurufen führt nichts aus. Es baut ein träges Coroutine-Objekt, dessen Rumpf erst voranschreitet, wenn es mit await erwartet oder als Task auf der Loop eingeplant wird.

import asyncio

async def greet() -> str:
    return "hello"

coro = greet()
print(type(coro).__name__)

async def main() -> None:
    print(await coro)

asyncio.run(main())

await markiert einen Unterbrechungspunkt: Die Coroutine friert dort ein, die Loop nimmt die Kontrolle zurück und bringt einen anderen bereiten Task voran. Zwischen zwei await läuft dein Code ohne Unterbrechung - für die meisten Strukturen sind keine Locks nötig. Um mehrere Coroutinen parallel zu starten (im Sinne von I/O), fasse sie in asyncio.gather zusammen: Drei Wartezeiten von je einer Sekunde enden in insgesamt einer Sekunde, nicht in dreien.

import asyncio
import time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} done"

async def main() -> None:
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("a", 1.0),
        fetch("b", 1.0),
        fetch("c", 1.0),
    )
    elapsed = time.perf_counter() - start
    print(results)
    print(f"total: {elapsed:.1f} s")

asyncio.run(main())

Standardmäßig gibt gather die erste ausgelöste Ausnahme weiter, lässt die anderen Tasks aber im Hintergrund weiterlaufen - selten das, was du willst. Mit return_exceptions=True werden Ausnahmen als Ergebnisse zurückgegeben, um danach sortiert zu werden. Seit Python 3.11 bevorzuge asyncio.TaskGroup: Beim ersten Fehler werden alle Geschwister-Tasks sauber abgebrochen, und etwaige mehrfache Ausnahmen steigen gebündelt in einer ExceptionGroup auf, um mit except* gefangen zu werden.

import asyncio
import time

def blocking_io() -> str:
    time.sleep(1)
    return "report written"

async def main() -> None:
    results = await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1, result="tick"),
    )
    print(results)

asyncio.run(main())

Wissenscheck

Stelle sicher, dass du die wichtigsten Punkte dieser Lektion behalten hast.

1. Was erzeugt der Aufruf fetch('a', 1.0), für sich allein geschrieben, ohne await oder create_task?
   • Ein träges Coroutine-Objekt: Keine einzige Zeile des Funktionsrumpfs läuft
   • Die sofortige, synchrone Ausführung der Coroutine
   • Ein automatisch auf der Event-Loop eingeplanter Task
   • Ein SyntaxError, weil await zwingend ist
2. Drei asyncio.sleep(1), zusammen über asyncio.gather gestartet: welche Gesamtdauer?
   • Etwa 3 Sekunden, da ein einziger Thread den Code ausführt
   • Unmöglich vorherzusagen, ohne die Anzahl der Kerne zu kennen
   • Etwa 1 Sekunde: Die Wartezeiten überlappen sich, die Loop wechselt von einem Task zum anderen
3. Warum ist time.sleep(2) innerhalb einer Coroutine verboten?
   • Es löst in einem async-Kontext eine Ausnahme aus
   • Es blockiert den Thread der Event-Loop: 2 Sekunden lang schreitet kein anderer Task voran
   • Es ist schlicht ungenauer als asyncio.sleep
   • Es verbraucht doppelt so viel Speicher