Introduction
Asynchronous concurrency has become essential for modern Python applications handling thousands of simultaneous requests. asyncio provides a powerful foundation, but its native performance remains limited by the default event loop. uvloop, a C implementation of the event loop, delivers 2-5x gains on I/O-intensive workloads. This tutorial guides you through uvloop integration, advanced patterns, and real performance measurements.
Prerequisites
- Python 3.11 or higher
- Solid knowledge of asynchronous programming
- pip and virtualenv
- Access to a Linux/macOS terminal for benchmarks
Environment Setup
python -m venv venv
source venv/bin/activate
pip install uvloop aiohttp pytest pytest-asyncioCreate an isolated environment and install uvloop along with aiohttp for async HTTP examples and pytest for testing.
Integrating uvloop into an 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() replaces the default loop before any event loop is created. This must be done once at application startup.
High-Performance HTTP Client
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())This client runs 50 concurrent requests. Using gather with return_exceptions prevents a single error from canceling the entire batch.
Advanced Pattern: Asynchronous Rate Limiting
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())This rate limiter uses a semaphore reset every second to precisely control throughput without blocking the asyncio loop.
Comparative Benchmark
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())This benchmark measures uvloop's impact on mixed tasks. Note that pure CPU tasks benefit less from uvloop than I/O workloads.
Best Practices
- Always call uvloop.install() before creating any loop
- Use asyncio.TaskGroup (Python 3.11+) for error handling
- Measure systematically with time.perf_counter and cProfile
- Avoid blocking calls in the main loop
- Set timeouts on all network operations
Common Mistakes to Avoid
- Forgetting to install uvloop before asyncio.run
- Using loop.run_until_complete after the application has started
- Ignoring exceptions in gather without return_exceptions=True
- Mixing heavy synchronous code inside coroutines
Further Reading
Explore these concepts further with our advanced Python performance courses: https://learni-group.com/formations