asyncio oder threading? In 3 Fragen richtig wählen

1. Übersicht

Viele Anfänger, die Python Concurrency lernen, fragen sich oft: „Ist asyncio oder threading besser?” – diese Frage ist bereits von Grund auf falsch gestellt, denn die beiden Tools lösen völlig unterschiedliche Probleme. Dieser Artikel stellt dir 3 konkrete Fragen, mit denen du selbst antworten kannst, zusammen mit ausführbarem Code. Nachdem du diesen Artikel gelesen hast, wirst du genau wissen, wann du threading einsetzt, wann asyncio, und wann beide sinnlos sind (dann brauchst du multiprocessing).

2. Warum die Frage „was ist schneller” falsch ist

Python hat einen Global Interpreter Lock (GIL) – nur ein Python-Thread kann zu jedem Zeitpunkt Python-Bytecode ausführen, egal wie viele CPU-Kerne die Maschine hat. threading und asyncio können den GIL nicht knacken. Beide helfen nur dabei, dass die CPU nicht stillsteht, wenn sie auf I/O wartet (Datei lesen, Netzwerk-Anfragen, DB-Queries) – sie helfen nicht, Berechnungen schneller zu machen.

import time

def cpu_bound_task():
    total = 0
    for i in range(10_000_000):
        total += i * i
    return total

start = time.perf_counter()
cpu_bound_task()
cpu_bound_task()
print(f"Sequenziell: {time.perf_counter() - start:.2f}s")

Sequenziell: 1.85s

Egal ob du diese beiden cpu_bound_task()-Aufrufe mit threading.Thread oder asyncio.gather ausführst – die Gesamtzeit wird nicht kürzer, weil der GIL nur einen Thread zur gleichen Zeit Python-Code ausführen lässt. Das ist der Grund, warum die erste Frage unten so wichtig ist.

Achtung: Wenn deine Aufgabe CPU-intensiv ist (schwere arithmetische Berechnungen, Bildverarbeitung, Verschlüsselung…), helfen weder threading noch asyncio. Du brauchst multiprocessing oder schreibst die rechenintensive Komponente als C/Rust-Extension (NumPy, Cython).

3. Frage 1: Ist deine Arbeit I/O-gebunden oder CPU-gebunden?

Das ist die entscheidende Frage, ob du überhaupt „Warte”-Concurrency (I/O-gebunden) verwenden solltest.

3.1. I/O-gebunden: threading und asyncio sind beide effektiv

import time
import requests
from concurrent.futures import ThreadPoolExecutor

URLS = ["https://httpbin.org/delay/1"] * 5

def fetch(url: str) -> int:
    return requests.get(url).status_code

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as pool:
    results = list(pool.map(fetch, URLS))
print(f"Threading: {time.perf_counter() - start:.2f}s, results={results}")

Threading: 1.12s, results=[200, 200, 200, 200, 200]

ThreadPoolExecutor führt 5 Anfragen gleichzeitig mit 5 OS-Threads aus. Da requests.get ein blockierender Aufruf ist, der auf das Netzwerk wartet, wird der GIL freigegeben, wenn der Thread auf I/O auf der C-Ebene wartet – also laufen die 5 Anfragen wirklich parallel, die Gesamtzeit beträgt ~1s statt 5s.

import asyncio
import aiohttp
import time

URLS = ["https://httpbin.org/delay/1"] * 5

async def fetch(session: aiohttp.ClientSession, url: str) -> int:
    async with session.get(url) as resp:
        return resp.status

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in URLS]
        return await asyncio.gather(*tasks)

start = time.perf_counter()
results = asyncio.run(main())
print(f"Asyncio: {time.perf_counter() - start:.2f}s, results={results}")

Asyncio: 1.08s, results=[200, 200, 200, 200, 200]

Das Ergebnis ist ähnlich wie bei Threading, aber der Mechanismus ist anders: nur ein einzelner Thread, die Event Loop awaitet aktiv und gibt die Kontrolle ab, wenn auf I/O gewartet wird, ohne dass OS-Threads erstellt werden.

3.2. CPU-gebunden: beide verlieren gegen multiprocessing

from concurrent.futures import ProcessPoolExecutor

def cpu_bound_task():
    return sum(i * i for i in range(10_000_000))

start = time.perf_counter()
with ProcessPoolExecutor(max_workers=2) as pool:
    results = list(pool.map(cpu_bound_task, [None, None]))
print(f"Multiprocessing: {time.perf_counter() - start:.2f}s")

Multiprocessing: 1.02s

ProcessPoolExecutor erstellt separate Prozesse, jeder Prozess hat seinen eigenen GIL, daher kannst du mehrere echte CPU-Kerne nutzen. Wenn du dich bei „threading oder asyncio” für eine CPU-gebundene Aufgabe zweifelst, ist die richtige Antwort beide sind falsch – nutze multiprocessing.

4. Frage 2: Unterstützt dein Ökosystem async-native Treiber?

Das ist die praktischste Frage, wenn du mit echtem Codebase arbeitest, wo du nicht alles von Grund auf schreibst.

import psycopg2  # synchroner Treiber, blockierend

def get_user(user_id: int) -> dict:
    conn = psycopg2.connect("dbname=app")
    cur = conn.cursor()
    cur.execute("SELECT id, name FROM users WHERE id = %s", (user_id,))
    row = cur.fetchone()
    conn.close()
    return {"id": row[0], "name": row[1]}

psycopg2 ist ein synchroner Treiber – es hat keine await-Version. Um mehrere Queries parallel mit diesem Treiber auszuführen, sind threading (oder run_in_executor von asyncio) die natürliche Wahl, da du nichts umschreiben musst.

import asyncpg  # async-nativer Treiber

async def get_user(conn: asyncpg.Connection, user_id: int) -> dict:
    row = await conn.fetchrow(
        "SELECT id, name FROM users WHERE id = $1", user_id
    )
    return {"id": row["id"], "name": row["name"]}

Wenn dein Ökosystem bereits async-native Treiber hat (asyncpg, aiohttp, motor für MongoDB…), zeigt asyncio seinen maximalen Vorteil: Tausende gleichzeitige Verbindungen auf einem einzigen Thread, ohne die Kosten der OS-Thread-Erstellung.

Achtung: Synchronen Code in async def zu mischen, ohne ihn richtig zu awaiten (z.B. requests.get direkt in einer Coroutine aufrufen) wird die ganze Event Loop blockieren – alle anderen Tasks müssen warten, auch die, die nichts damit zu tun haben. Das ist der häufigste Fehler beim Umsteigen von Threading auf Asyncio.

5. Frage 3: Wie viele gleichzeitige Tasks brauchst du?

Der Ressourcen-Unterschied zwischen Thread und Coroutine ist sehr groß, wenn die Anzahl der Tasks wächst.

import threading
import resource

def noop():
    threading.Event().wait(2)

threads = [threading.Thread(target=noop) for _ in range(2000)]
for t in threads:
    t.start()

peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
print(f"Peak RSS mit 2000 Threads: {peak_kb / 1024:.1f} MB")

for t in threads:
    t.join()

Peak RSS mit 2000 Threads: 1850.3 MB

Jeder OS-Thread verbraucht standardmäßig mehrere hundert KB bis mehrere MB Stack-Speicher, also kosten 2000 Threads leicht fast 2 GB RAM und erzeugen großen Druck auf den OS Scheduler durch Context Switching.

import asyncio
import resource

async def noop():
    await asyncio.sleep(2)

async def main():
    tasks = [asyncio.create_task(noop()) for _ in range(2000)]
    await asyncio.gather(*tasks)
    peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    print(f"Peak RSS mit 2000 Coroutines: {peak_kb / 1024:.1f} MB")

asyncio.run(main())

Peak RSS mit 2000 Coroutines: 95.7 MB

Coroutines sind reine Python-Objekte, sie brauchen keinen eigenen OS-Thread – das Erstellen von 2000 Tasks kostet nur mehrere MB. Wenn du Tausende gleichzeitige Verbindungen verarbeiten musst (Web Server, Chat Server, großer Crawler), ist asyncio eine beinahe zwingende Wahl; threading verbraucht Ressourcen, bevor dir die Bandbreite ausgeht.

6. Schnelle Entscheidungstabelle

Frage	Antwort → Wahl
1. I/O-gebunden oder CPU-gebunden?	CPU-gebunden → multiprocessing. I/O-gebunden → zu Frage 2
2. Hat dein Stack async-native Treiber?	Nein (alte synchrone Treiber) → threading. Ja (aiohttp, asyncpg…) → asyncio
3. Wie viele gleichzeitige Tasks brauchst du?	Einige Dutzend → threading einfach genug. Hunderte/Tausende → asyncio

In der Praxis verwenden viele Systeme beides: asyncio für I/O-gebundene Netzwerk-Layer in großer Menge, kombiniert mit run_in_executor, um synchronen Code auszuführen (z.B. alte Bibliotheken) in einem separaten Thread Pool, um die Event Loop nicht zu blockieren.

import asyncio

def legacy_blocking_call(x: int) -> int:
    import time
    time.sleep(1)  # simuliert alte blockierende Bibliothek
    return x * 2

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, legacy_blocking_call, 21)
    print(result)

asyncio.run(main())

42

run_in_executor schiebt die synchrone Funktion in einen Thread im Standard-Thread Pool, während die Event Loop frei andere Coroutines verarbeiten kann, während sie wartet.

7. Verhalten mit pytest testen

Beispiel eines Tests, der bestätigt, dass asyncio.gather Tasks wirklich parallel ausführt, nicht nacheinander:

import asyncio
import time
import pytest

async def sleepy_task(seconds: float) -> float:
    await asyncio.sleep(seconds)
    return seconds

@pytest.mark.asyncio
async def test_gather_runs_concurrently():
    start = time.perf_counter()
    results = await asyncio.gather(
        sleepy_task(0.5), sleepy_task(0.5), sleepy_task(0.5)
    )
    elapsed = time.perf_counter() - start

    assert results == [0.5, 0.5, 0.5]
    # Wenn sequenziell würde es ~1.5s dauern, parallel nur ~0.5s
    assert elapsed < 0.7

Abhängigkeiten zum Testen:

pytest
pytest-asyncio

Dieser Test bestätigt das wahre Wesen von asyncio: Drei sleep(0.5)-Tasks laufen gleichzeitig auf einem Thread und dauern nur ~0.5s, nicht 1.5s – wenn die Assertion elapsed < 0.7 fehlschlägt, bedeutet das, dass dein Code nacheinander awaitet statt gathered.

8. Häufige Fehler

• time.sleep() in Coroutine verwenden: muss await asyncio.sleep() sein. time.sleep() ist ein blockierender Aufruf, der die ganze Event Loop blockiert.
• await vergessen: Aufrufen von fetch(session, url) ohne await erstellt nur ein Coroutine-Objekt, führt nichts aus – Python warnt RuntimeWarning: coroutine was never awaited.
• State zwischen Threads ohne Lock teilen: threading hat echte Race Conditions (mehrere Threads schreiben in eine Variable), braucht threading.Lock. asyncio auf einem einzigen Thread hat diesen Race-Condition-Typ nicht, kann aber Logic-Bugs haben, wenn await in der Mitte eines Code-Segments, das „atomic” sein sollte, steht.
• ThreadPoolExecutor für CPU-gebundene Aufgaben verwenden und dann enttäuscht sein, dass es nicht schneller ist – weil der GIL noch blockiert, nutze ProcessPoolExecutor.

Achtung: Wenn dein Team threading gewohnt ist und der Codebase nur aus synchronen Bibliotheken besteht, refaktorisiere nicht einfach zu asyncio, weil es „neuer” ist. Die Kosten für das Umschreiben aller Treiber, unterschiedliche Exception-Verarbeitung und das Debuggen komplexer Coroutine Stack Traces sind viel höher als der Nutzen, wenn die Anzahl der gleichzeitigen Tasks nicht sehr groß ist.

9. Fazit

Die Wahl zwischen asyncio und threading ist nicht „was ist schneller”, sondern drei konkrete Fragen: Ist das Problem I/O-gebunden oder CPU-gebunden, hat dein Bibliotheks-Ökosystem async-native Unterstützung, und wie groß ist die Concurrency-Skalierung, die du brauchst. CPU-gebunden braucht immer multiprocessing. I/O-gebunden mit alten Bibliotheken und kleine Task-Menge – dann ist threading einfach und leicht zu integrieren. I/O-gebunden in großem Maßstab mit async-native Bibliotheken – dann ist asyncio die richtige Wahl für Performance und Ressourcen-Effizienz.