0Pricing
Python Academy · Lektion

Asynchrone I/O-Muster in der Praxis

Setzen Sie asyncio für HTTP-Clients, Warteschlangen und praxisnahe Pipelines ein.

Asynchrone I/O-Muster in der Praxis ist eine kostenlose Python Academy-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des Python Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der Python Academy-Kurs umfasst insgesamt 4 Lektionen.

Bewährte Vorgehensweisen für Event Loops

Verwenden Sie stets asyncio.run() als einzigen Einstiegspunkt. Vermeiden Sie eine unüberlegte Mischung aus synchronem und asynchronem Code, und blockieren Sie den Event Loop niemals mit lang andauernder CPU-Arbeit.

import asyncio

async def main():
    # Never: time.sleep() — blocks the event loop
    # Always: await asyncio.sleep()
    await asyncio.sleep(1)

asyncio.run(main())

Verbindungspooling

Verwenden Sie Verbindungen wieder, statt für jede Anfrage eine neue zu öffnen. Bibliotheken wie asyncpg und aiohttp stellen integrierte Pools bereit.

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@host/db",
        min_size=5, max_size=20
    )
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT 1")
    await pool.close()

Producer-Consumer mit asyncio.Queue

asyncio.Queue entkoppelt Producer und Consumer, die als nebenläufige Tasks ausgeführt werden.

import asyncio

async def producer(q):
    for i in range(5):
        await q.put(i)
        await asyncio.sleep(0.5)
    await q.put(None)  # sentinel

async def consumer(q):
    while (item := await q.get()) is not None:
        print(f"processing {item}")
        q.task_done()

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

asyncio.run(main())

Ratenbegrenzung mit Semaphore

Begrenzen Sie gleichzeitig ausgeführte ausgehende Anfragen, damit externe APIs nicht überlastet werden und keine Ratenlimits erreicht werden.

import asyncio, aiohttp

SEM = asyncio.Semaphore(10)  # 10 concurrent requests max

async def fetch(session, url):
    async with SEM:
        async with session.get(url) as r:
            return await r.json()

async def main(urls):
    async with aiohttp.ClientSession() as s:
        return await asyncio.gather(*[fetch(s, u) for u in urls])

Wiederholungsversuche mit exponentiellem Back-off

Umschließen Sie wiederholbare Vorgänge mit einer Back-off-Logik, um vorübergehende Fehler kontrolliert zu behandeln.

import asyncio

async def retry(coro_fn, retries=3, base=1):
    for attempt in range(retries):
        try:
            return await coro_fn()
        except Exception as e:
            if attempt == retries - 1: raise
            delay = base * 2 ** attempt
            await asyncio.sleep(delay)

Sauberes Herunterfahren

Fangen Sie asyncio.CancelledError in lang laufenden Tasks ab und verarbeiten Sie Betriebssystemsignale, um sauber herunterzufahren.

import asyncio, signal

async def server():
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, stop.set)
    await stop.wait()
    print("Shutting down gracefully")

Streaming-Antworten

Verwenden Sie asynchrone Generatoren, um große Antworten zu streamen, ohne sie vollständig in den Speicher zu laden.

import asyncio, aiohttp

async def stream(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            async for chunk in r.content.iter_chunked(4096):
                process(chunk)

async def process(chunk):
    await asyncio.sleep(0)
    print(len(chunk), "bytes")

Hintergrund-Tasks in FastAPI

FastAPIs BackgroundTasks reiht Arbeiten zur Ausführung nach dem Senden der Antwort ein, sodass Endpunkte schnell bleiben.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, body: str):
    # heavy synchronous work
    pass

@app.post("/register")
async def register(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email, "Welcome!")
    return {"status": "registered"}

Heartbeat-Muster

Führen Sie mit create_task eine periodische Hintergrundaufgabe neben der Hauptschleife des Servers aus.

import asyncio

async def heartbeat():
    while True:
        print("alive")
        await asyncio.sleep(30)

async def main():
    asyncio.create_task(heartbeat())
    await serve_forever()

Asynchronen Code profilieren

Verwenden Sie yappi oder py-spy für asynchronitätsbewusstes Profiling. Aktivieren Sie den asyncio-Debugmodus, um Warnungen zu langsamen Callbacks zu erhalten.

# Enable debug to catch slow callbacks:
# asyncio.run(main(), debug=True)

# PYTHONASYNCIODEBUG=1 python server.py
# logs: Executing <Task coroutine=...> took 0.150 seconds

Häufige Fallstricke vermeiden

Wichtige Regeln: Rufen Sie niemals blockierenden Code im Event Loop auf; warten Sie stets auf Koroutinen; brechen Sie Tasks beim Herunterfahren ab; verwenden Sie Verbindungspools; begrenzen Sie die Nebenläufigkeit mit Semaphoren.

import asyncio

async def bad():
    import time
    time.sleep(2)        # blocks the entire event loop!

async def good():
    await asyncio.sleep(2)  # yields control

Wissenscheck

Welche asyncio-Komponente entkoppelt Producer und Consumer, die als nebenläufige Tasks ausgeführt werden?

Zusammenfassung

Asynchrone Muster für den Produktiveinsatz: Verwenden Sie Verbindungspools, steuern Sie die Nebenläufigkeit mit Semaphore, implementieren Sie Wiederholungsversuche mit Back-off, streamen Sie große Datenmengen, fahren Sie auf Signale hin sauber herunter und führen Sie periodische Tasks mit create_task aus.

Häufig gestellte Fragen

Ist die Lektion „Asynchrone I/O-Muster in der Praxis“ kostenlos?

Ja — der vollständige Text von „Asynchrone I/O-Muster in der Praxis“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des Python Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der Python Academy-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Asynchrone I/O-Muster in der Praxis“?

Setzen Sie asyncio für HTTP-Clients, Warteschlangen und praxisnahe Pipelines ein. Du übst Python Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um Python Academy zu starten?

Keine Vorkenntnisse erforderlich. Python Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Asynchrone I/O-Muster in der Praxis“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser Python Academy-Lektion Code schreiben und ausführen?

Ja. Jede Python Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Coroutinen und die Ereignisschleife
  2. await, Tasks und Zusammenfassen
  3. Asynchrone Kontextmanager und Iteratoren
  4. Asynchrone I/O-Muster in der Praxis
← Zurück zu Python Academy