Vollständiger Python-Asyncio-Leitfaden: Parallelität im Jahr 2026 meistern

Was Asyncio tatsächlich löst

Asyncio existiert für einen Job: das gleichzeitige Ausführen vieler E/A-gebundener Vorgänge in einem einzelnen Thread, ohne den Overhead von Threads oder Prozessen. Wenn Ihr Programm die meiste Zeit damit verbringt, zu warten – auf eine Datenbankantwort, eine HTTP-Antwort, eine gelesene Datei – lässt Asyncio zu, dass es während dieser Wartezeit zu anderen Aufgaben übergeht, anstatt untätig herumzusitzen.

Dies unterscheidet sich grundlegend von der CPU-gebundenen Parallelität. asyncio gibt Ihnen nicht mehr CPU-Kerne; Dadurch können Sie einen einzelnen Kern besser ausnutzen, während Sie auf E/A warten. Für CPU-intensive Arbeiten (Bildverarbeitung, Zahlenverarbeitung) benötigen Sie weiterhinmultiprocessingoder eine native Erweiterung.

Die Ereignisschleife, erklärt

Im Zentrum jedes Asyncio-Programms steht die Ereignisschleife – ein Single-Threaded-Scheduler, der Coroutinen ausführt, die Kontrolle abgibt, wenn man auf etwas wartet, und sie fortsetzt, wenn das erwartete Ergebnis bereit ist.

import asyncio

async def say_hello():
    print("start")
    await asyncio.sleep(1)
    print("end")

asyncio.run(say_hello())

asyncio.run()Erstellt eine neue Ereignisschleife, führt die Coroutine vollständig aus und schließt dann die Schleife. Es ist der richtige Einstiegspunkt der obersten Ebene für jedes Asyncio-Skript – vermeiden Sie das ältereget_event_loop().run_until_complete()Muster, es sei denn, Sie haben einen bestimmten Grund, die Schleife manuell zu verwalten.

Coroutinen vs. Aufgaben

Ein Coroutine-Objekt (erstellt durch den Aufruf einer ||||-Funktion) macht selbst nichts – es ist ein angehaltenes, generatorähnliches Objekt, das darauf wartet, gesteuert zu werden. In Erwartung wird es nacheinander inline ausgeführt. Einasync defAufgabeUmschließt eine Coroutine und plant sie so, dass sie sofort gleichzeitig in der Ereignisschleife ausgeführt wird.task.py

import asyncio
import time

async def fetch(n):
    await asyncio.sleep(1)
    return n * 2

async def sequential():
    start = time.perf_counter()
    results = [await fetch(i) for i in range(5)]
    print("sequential:", time.perf_counter() - start)
    return results

async def concurrent():
    start = time.perf_counter()
    tasks = [asyncio.create_task(fetch(i)) for i in range(5)]
    results = await asyncio.gather(*tasks)
    print("concurrent:", time.perf_counter() - start)
    return results

asyncio.run(sequential())   # ~5 seconds
asyncio.run(concurrent())   # ~1 second

Gather() vs. TaskGroup

ist seit Python 3.4 die Standardmethode, um mehrere Coroutinen gleichzeitig auszuführen, hat aber einen scharfen Vorteil: Wenn eine Aufgabe ausgelöst wird, laufen die anderen im Hintergrund weiter, es sei denn, Sie übergeben

asyncio.gather()und Fehler manuell behandeln. Dies ist eine häufige Ursache für verwaiste Aufgaben und nicht behandelte Ausnahmewarnungen in Produktionsprotokollen.return_exceptions=TruePython 3.11 eingeführt

, ein strukturiertes Parallelitätsprimitiv, das dieses Problem ordnungsgemäß behebt: Wenn eine untergeordnete Aufgabe fehlschlägt, bricht die Gruppe alle verbleibenden Geschwisteraufgaben ab und löst einasyncio.TaskGroupaus jeden Fehler enthalten.ExceptionGrouptaskgroup.py

import asyncio

async def risky(n):
    if n == 2:
        raise ValueError("boom at " + str(n))
    await asyncio.sleep(1)
    return n

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            results = [tg.create_task(risky(i)) for i in range(4)]
    except* ValueError as eg:
        for exc in eg.exceptions:
            print("caught:", exc)

asyncio.run(main())

Syntax-Handlesexcept*Instanzen, sodass Sie jeden Fehler verarbeiten können, der in der gesamten Gruppe aufgetreten ist, und nicht nur den ersten. Wenn Sie auf Python 3.11 oder höher abzielen, bevorzugen SieExceptionGroupüber rohTaskGroupfür alles, was über die triviale Fire-and-Forget-Parallelität hinausgeht.gather()Asynchrone Kontextmanager und Sperren

Der gemeinsam genutzte veränderliche Status über gleichzeitige Aufgaben hinweg muss auch in einem einzelnen Thread weiterhin geschützt werden, da ein

Der Punkt kann die Steuerung während des Vorgangs an eine andere Aufgabe übergeben.awaitschützt kritische Abschnitte auf die gleiche Weiseasyncio.Lockfunktioniert für Threads.threading.Locklock_example.py

import asyncio

counter = 0
lock = asyncio.Lock()

async def increment():
    global counter
    async with lock:
        current = counter
        await asyncio.sleep(0)  # simulate a yield point
        counter = current + 1

async def main():
    await asyncio.gather(*(increment() for _ in range(100)))
    print(counter)  # always 100, never less

asyncio.run(main())

Der Yield-Punkt würde eine weitere Aufgabe zwischen Lesen und Schreibenawait asyncio.sleep(0)ermöglichen , was zu verlorenen Updates führt – die klassische Rennbedingung, nur in einem Thread statt in vielen.counterDie blockierende Anruffalle

Dies ist die größte Einzelquelle für Fehlerberichte „Meine Asyncio-App ist langsam“. Die Ereignisschleife ist Single-Threaded. Jeder synchrone Aufruf, der blockiert –

, eine Blockierungtime.sleep(), eine CPU-lastige Schleife, eine ungepufferte Datei, die auf einer langsamen Festplatte gelesen wird – friert die gesamte Schleife ein, einschließlich aller anderen darauf wartenden Aufgaben.requests.get()bad_vs_good.py

# BAD: blocks the whole event loop for 2 seconds
import time
async def bad():
    time.sleep(2)

# GOOD: yields control back to the loop
import asyncio
async def good():
    await asyncio.sleep(2)

# GOOD: offload a real blocking call to a thread
async def good_blocking_lib():
    result = await asyncio.to_thread(some_blocking_function, arg1, arg2)
    return result

(3.9+) zum Blockieren von E/A von Bibliotheken von Drittanbietern, die kein asynchrones Äquivalent haben. Für CPU-gebundene Arbeit verwenden Sieasyncio.to_thread()mit einemloop.run_in_executor()Stattdessen helfen Threads aufgrund der GIL nicht bei CPU-gebundener Arbeit.ProcessPoolExecutorZeitüberschreitungen und Stornierung

Der Produktionscode benötigt bei jedem externen Aufruf Zeitüberschreitungen. Python 3.11 hinzugefügt

Production code needs timeouts on every external call. Python 3.11 addedasyncio.timeout()als sauberer Kontextmanager-Ersatz für den älterenasyncio.wait_for().

import asyncio

async def slow_call():
    await asyncio.sleep(10)

async def main():
    try:
        async with asyncio.timeout(2):
            await slow_call()
    except TimeoutError:
        print("operation timed out after 2s")

asyncio.run(main())

Wenn ein Timeout ausgelöst wird, bricht Asyncio die innere Aufgabe ab, indem esasyncio.CancelledErrorauslöst drin. Wenn Ihre Coroutine Ressourcen umschließt, die bereinigt werden müssen (offene Verbindungen, Dateihandles), verwenden Sietry/finallyoder ein asynchroner Kontextmanager, damit beim Abbruch keine Ressourcen verloren gehen.

Echtes Beispiel: Concurrent API Fetcher

Hier ist ein vollständiges, produktionsorientiertes Beispiel: gleichzeitiges Abrufen mehrerer URLs mit einem Verbindungslimit, einem Zeitlimit pro Anfrage und strukturierter Fehlerbehandlung mitaiohttp.

import asyncio
import aiohttp

async def fetch_one(session, url, semaphore):
    async with semaphore:
        try:
            async with asyncio.timeout(5):
                async with session.get(url) as resp:
                    return url, resp.status, await resp.text()
        except (TimeoutError, aiohttp.ClientError) as e:
            return url, None, str(e)

async def fetch_all(urls, max_concurrent=10):
    semaphore = asyncio.Semaphore(max_concurrent)
    async with aiohttp.ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_one(session, u, semaphore)) for u in urls]
        return [t.result() for t in tasks]

urls = ["https://example.com"] * 50
results = asyncio.run(fetch_all(urls))
print(len(results), "requests completed")

DasSemaphorebegrenzt, wie viele Anfragen gleichzeitig ausgeführt werden, was sowohl Ihren Client als auch den Zielserver davor schützt, durch Hunderte gleichzeitiger Verbindungen überlastet zu werden. Dieses Muster – Sitzungswiederverwendung, Semaphor-begrenzte Parallelität, Zeitüberschreitung pro Anfrage, strukturierte Aufgabengruppe – kommt dem nahe, was Sie in einem Produktions-Scraper oder API-Aggregator sehen würden.

Häufige Fehler

• Vergessen, auf eine Coroutine zu warten.Aufruf einesasync defFunktion ohneawaitErstellt lediglich ein Coroutine-Objekt und unternimmt nichts. Python gibt eine Warnung aus: „Coroutine wurde nie erwartet“, aber in verrauschten Protokollen ist dies leicht zu übersehen.
• Aufgaben erstellen, ohne eine Referenz zu behalten. asyncio.create_task()enthält intern nur eine schwache Referenz; Wenn für das Aufgabenobjekt vor dessen Abschluss Müll gesammelt wird, kann es stillschweigend abgebrochen werden. Speichern Sie Aufgaben in einer Liste oder einem Satz, bis sie abgeschlossen sind.
• Mischen von synchronen und asynchronen Datenbanktreibern.Die Verwendung eines blockierenden Treibers (wie einfachpsycopg2) innerhalb von asynchronem Code macht den Zweck zunichte – verwenden Sie einen asynchronen Treiber (asyncpg,aiomysql) oder schließen Sie den Aufruf mitto_thread()ein .
• Keine Zeitüberschreitungen festlegen.Ein einzelner hängender Netzwerkaufruf ohne Zeitüberschreitung kann einen gesamten Anforderungshandler unter Last auf unbestimmte Zeit blockieren.
• Übermäßige Nutzung von Asyncio für CPU-gebundene Arbeit.Wenn Ihr Engpass in der Berechnung und nicht im Warten liegt, erhöht Asyncio die Komplexität, ohne die Geschwindigkeit zu erhöhen.