Introduction
La concurrence asynchrone est devenue indispensable pour les applications Python modernes traitant des milliers de requêtes simultanées. asyncio offre un socle puissant mais ses performances natives restent limitées par la boucle d'événements par défaut. uvloop, une réimplémentation en C de la boucle d'événements, permet d'atteindre des gains de 2 à 5x sur les workloads I/O intensifs. Ce tutoriel vous guide pas à pas dans l'intégration d'uvloop, la création de patterns avancés et les mesures de performance réelles.
Prérequis
- Python 3.11 ou supérieur
- Connaissances solides en programmation asynchrone
- pip et virtualenv
- Accès à un terminal Linux/macOS pour les benchmarks
Configuration de l'environnement
python -m venv venv
source venv/bin/activate
pip install uvloop aiohttp pytest pytest-asyncioOn crée un environnement isolé et on installe uvloop ainsi que aiohttp pour les exemples HTTP asynchrones et pytest pour les tests.
Intégration d'uvloop dans une application
import asyncio
import uvloop
uvloop.install()
async def main():
print("Boucle uvloop active")
await asyncio.sleep(0.1)
if __name__ == "__main__":
asyncio.run(main())uvloop.install() remplace la boucle par défaut avant toute création d'événement loop. Cela doit être fait une seule fois au démarrage du programme.
Client HTTP haute performance
import asyncio
import aiohttp
import uvloop
uvloop.install()
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
urls = ["https://httpbin.org/delay/1"] * 50
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
print(f"Requêtes terminées: {len([r for r in results if not isinstance(r, Exception)])}/50")
asyncio.run(main())Ce client exécute 50 requêtes concurrentes. gather avec return_exceptions évite qu'une seule erreur n'annule tout le batch.
Pattern avancé : Rate limiting asynchrone
import asyncio
import time
class AsyncRateLimiter:
def __init__(self, max_per_second: int):
self.max_per_second = max_per_second
self.semaphore = asyncio.Semaphore(max_per_second)
self.last_reset = time.monotonic()
async def acquire(self):
await self.semaphore.acquire()
now = time.monotonic()
if now - self.last_reset > 1:
self.last_reset = now
for _ in range(self.max_per_second):
self.semaphore.release()
async def worker(limiter, i):
await limiter.acquire()
print(f"Tâche {i} exécutée")
await asyncio.sleep(0.2)
async def main():
limiter = AsyncRateLimiter(5)
tasks = [worker(limiter, i) for i in range(20)]
await asyncio.gather(*tasks)
asyncio.run(main())Ce rate limiter utilise un sémaphore réinitialisé chaque seconde pour contrôler précisément le débit sans bloquer la boucle asyncio.
Benchmark comparatif
import asyncio
import time
import uvloop
async def cpu_bound(n):
return sum(i * i for i in range(n))
async def main():
start = time.perf_counter()
results = await asyncio.gather(*[cpu_bound(10_000_000) for _ in range(8)])
print(f"Temps total: {time.perf_counter() - start:.2f}s")
print("Avec uvloop:")
uvloop.install()
asyncio.run(main())Ce benchmark mesure l'impact d'uvloop sur des tâches mixtes. Notez que les tâches CPU pures bénéficient moins d'uvloop que les workloads I/O.
Bonnes pratiques
- Toujours appeler uvloop.install() avant toute création de loop
- Utiliser asyncio.TaskGroup (Python 3.11+) pour la gestion d'erreurs
- Mesurer systématiquement avec time.perf_counter et cProfile
- Éviter les appels bloquants dans la boucle principale
- Configurer des timeouts sur toutes les opérations réseau
Erreurs courantes à éviter
- Oublier d'installer uvloop avant asyncio.run
- Utiliser loop.run_until_complete après le démarrage de l'application
- Ignorer les exceptions dans gather sans return_exceptions=True
- Mélanger du code synchrone lourd dans des coroutines
Pour aller plus loin
Approfondissez ces concepts avec nos formations avancées sur la performance Python : https://learni-group.com/formations