asyncio ou threading ? Bien choisir en 3 questions

1. Présentation

Beaucoup de débutants en concurrence Python demandent : « asyncio ou threading, c’est mieux ? » — cette question est mal posée dès le départ, car ces deux outils résolvent deux problèmes différents. Cet article vous pose 3 questions spécifiques pour que vous trouviez la réponse vous-même, avec du code fonctionnel. Après la lecture, vous saurez exactement quand utiliser threading, quand utiliser asyncio, et quand aucun des deux ne servira (il faudra multiprocessing).

2. Pourquoi la question « lequel est plus rapide » est mal posée

Python a une Global Interpreter Lock (GIL) — un seul thread Python exécute du bytecode à la fois, peu importe le nombre de cores du processeur. threading et asyncio ne contournent pas la GIL. Les deux vous aident simplement à éviter que le CPU ne « rouille » quand vous attendez de l’I/O (lire des fichiers, faire des appels réseau, interroger une DB) — ils ne rendent pas les calculs plus rapides.

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"Sequential: {time.perf_counter() - start:.2f}s")

Sequential: 1.85s

Même si vous exécutez deux fois cpu_bound_task() avec threading.Thread ou asyncio.gather, le temps total ne diminue pas, car la GIL ne laisse qu’un thread exécuter du code Python à la fois. C’est pourquoi la première question ci-dessous est la plus importante.

Remarque : Si votre problème est CPU-bound (calculs mathématiques lourds, traitement d’images, chiffrement…), ni threading ni asyncio ne vous aident. Vous avez besoin de multiprocessing ou d’écrire la partie lourde en extension C/Rust (NumPy, Cython).

3. Question 1 : Votre travail est-il I/O-bound ou CPU-bound ?

C’est la question décisive : devrais-je utiliser la concurrence type « attente » (I/O-bound) ou pas.

3.1. I/O-bound : threading et asyncio sont tous deux efficaces

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 lance 5 requêtes en parallèle avec 5 OS threads. Parce que requests.get est un appel bloquant qui attend le réseau, la GIL est libérée quand le thread attend l’I/O au niveau C — donc les 5 requêtes tournent vraiment en parallèle, le temps total est ~1s au lieu de 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]

Le résultat est équivalent au threading, mais le mécanisme est différent : un seul thread, la boucle d’événements await activement et cède le contrôle quand elle attend l’I/O, sans que l’OS crée de nouveaux threads.

3.2. CPU-bound : les deux perdent contre 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 crée des processus séparés, chacun a sa propre GIL, donc on peut vraiment exploiter plusieurs cores du CPU. Si vous hésitez entre « threading ou asyncio » pour un problème CPU-bound, la bonne réponse est aucun des deux — utilisez multiprocessing.

4. Question 2 : Votre écosystème de bibliothèques supporte-t-il async ?

C’est la question la plus pragmatique quand on travaille dans un vrai codebase, où vous n’écrivez pas tout à partir de zéro.

import psycopg2  # driver synchrone, bloquant

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 est une bibliothèque bloquante — il n’y a pas de version await. Pour exécuter plusieurs requêtes en parallèle avec ce driver, threading (ou run_in_executor d’asyncio) est le choix naturel, car vous n’avez rien à réécrire.

import asyncpg  # driver async-natif

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"]}

Si votre écosystème a déjà des drivers async-natifs (asyncpg, aiohttp, motor pour MongoDB…), asyncio montre tout son potentiel : des milliers de connections simultanées sur un seul thread, sans la surcharge de créer un OS thread.

Remarque : Mélanger du code bloquant dans une fonction async def sans bien l’await (par exemple, appeler requests.get directement dans une coroutine) va bloquer toute la boucle d’événements — tous les autres tasks doivent attendre, même ceux qui n’y sont pas liés. C’est l’erreur la plus courante quand on migre de threading à asyncio.

5. Question 3 : Combien de tasks simultanés vous faut-il ?

Le coût en ressources entre un thread et une coroutine change radicalement quand le nombre de tasks augmente.

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 avec 2000 threads: {peak_kb / 1024:.1f} MB")

for t in threads:
    t.join()

Peak RSS avec 2000 threads: 1850.3 MB

Chaque OS thread consomme par défaut quelques centaines de KB à quelques MB de stack, donc 2000 threads engloutissent facilement 2 GB de RAM et créent une forte contention sur le planificateur de l’OS.

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 avec 2000 coroutines: {peak_kb / 1024:.1f} MB")

asyncio.run(main())

Peak RSS avec 2000 coroutines: 95.7 MB

Une coroutine est un objet Python pur, elle ne demande pas un thread OS séparé — créer 2000 tasks ne coûte que quelques dizaines de MB. Si vous devez gérer des milliers de connexions simultanées (serveur web, serveur de chat, crawler à grande échelle), asyncio est quasi obligatoire ; threading épuiserait les ressources avant d’atteindre la limite de bande passante.

6. Tableau de décision rapide

Question	Réponse → Choix
1. I/O-bound ou CPU-bound ?	CPU-bound → multiprocessing. I/O-bound → question 2
2. Vos bibliothèques ont-elles du async-natif ?	Non (vieux drivers bloquants) → threading. Oui (aiohttp, asyncpg…) → asyncio
3. Combien de tasks simultanés ?	Quelques dizaines → threading assez simple. Centaines/milliers → asyncio

En pratique, beaucoup de systèmes utilisent les deux : asyncio pour la couche I/O réseau en gros volume, combiné avec run_in_executor pour exécuter les bouts de code bloquant (vieilles bibliothèques) dans un thread pool séparé pour ne pas bloquer la boucle d’événements.

import asyncio

def legacy_blocking_call(x: int) -> int:
    import time
    time.sleep(1)  # simule une vieille bibliothèque bloquante
    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 pousse la fonction bloquante vers un thread du pool par défaut, la boucle d’événements reste libre de traiter les autres coroutines pendant qu’on attend.

7. Tester le comportement avec pytest

Exemple de test qui confirme que asyncio.gather lance vraiment les tasks en parallèle, pas séquentiellement :

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]
    # En séquentiel, ce serait ~1.5s, en parallèle ~0.5s
    assert elapsed < 0.7

Dépendances nécessaires pour le test :

pytest
pytest-asyncio

Ce test confirme la nature d’asyncio : trois tasks sleep(0.5) tournent ensemble sur un seul thread en seulement ~0.5s total, pas 1.5s — si l’assertion elapsed < 0.7 échoue, c’est que votre code fait des await séquentiellement au lieu de les regrouper.

8. Erreurs courantes

• Utiliser time.sleep() dans une coroutine : il faut utiliser await asyncio.sleep(). time.sleep() est un appel bloquant qui paralyse toute la boucle d’événements.
• Oublier await : appeler fetch(session, url) sans await crée juste un objet coroutine, n’exécute rien — Python lève RuntimeWarning: coroutine was never awaited.
• Partager un état entre threads sans verrous : threading a de vraies race conditions (plusieurs threads écrivent la même variable), besoin de threading.Lock. asyncio sur un seul thread n’a pas ce type de race condition, mais peut quand même avoir des bugs de logique si on await au mauvais endroit dans du code supposément « atomique ».
• Utiliser ThreadPoolExecutor pour du CPU-bound puis être déçu par le manque de vitesse — c’est parce que la GIL bloque toujours, utilisez ProcessPoolExecutor.

Remarque : Si votre équipe maîtrise threading et le codebase est plein de vieilles bibliothèques bloquantes, ne vous précipitez pas à refactoriser vers asyncio juste parce que c’est « plus neuf ». Le coût de réécrire tous les drivers, gérer les exceptions différemment, et déboguer les stack traces compliquées des coroutines est bien plus lourd que les bénéfices, si le nombre de tasks simultanés n’est pas énorme.

9. Conclusion

Choisir entre asyncio et threading n’est pas une question de « lequel est plus rapide » mais de trois questions précises : votre problème est-il I/O-bound ou CPU-bound, votre écosystème a-t-il du async-natif, et quelle scale de concurrence vous faut-il. CPU-bound demande toujours multiprocessing. I/O-bound avec une vieille bibliothèque et peu de tasks, threading est plus simple et facile à intégrer. I/O-bound à grande scale avec du async-natif, asyncio est le choix juste pour la performance et les ressources.