El event loop de asyncio no es magia universal — es una herramienta específica para un problema específico. Antes de reescribir tu aplicación con async/await, vale la pena entender exactamente qué problema resuelve y cuál es el coste de adoptarlo.
La pregunta clave: ¿dónde pasa el tiempo tu código?
Python tiene un GIL (Global Interpreter Lock) que impide que dos hilos ejecuten bytecode Python simultáneamente. Eso hace que el threading sea ineficaz para trabajo CPU-intensivo. Pero el GIL se libera cuando un hilo está bloqueado esperando I/O del sistema operativo: lectura de red, disco, sockets.
asyncio aprovecha exactamente esa ventana. En lugar de tener N hilos bloqueados esperando respuestas de red, tienes un solo hilo que registra callbacks y deja que el event loop lo coordine todo. El event loop es un bucle select/epoll/kqueue bajo el capó: el SO notifica cuando un descriptor de fichero está listo, y asyncio retoma la corrutina correspondiente.
Esto funciona de maravilla cuando el cuello de botella es I/O-bound: peticiones HTTP, consultas a bases de datos, mensajes WebSocket. Si el tiempo lo consume la CPU calculando algo, asyncio no ayuda en absoluto — una corrutina que hace cálculos pesados bloquea el event loop entero igual que una función normal.
La diferencia crucial entre asyncio y threading no es rendimiento bruto, sino coste de escalado. Un hilo del SO cuesta ~MB de stack y tiene overhead de cambio de contexto. Una corrutina de Python cuesta ~KB y el cambio de contexto es cooperativo, sin llamada al núcleo. Por eso asyncio puede gestionar decenas de miles de conexiones donde el threading empieza a ceder.
El coste real: async infecta hacia arriba
Hay un problema de diseño que debes interiorizar antes de escribir una sola línea async: una función async def solo puede ser esperada desde otra función async def. Esto se llama el problema de “async all the way down”. En cuanto un módulo interno necesita ser async, la superficie async sube por toda la cadena de llamadas hasta el punto de entrada. No puedes simplemente hacer async una función en el fondo de tu stack sin que el efecto se propague.
Si esa propagación afecta código que usa librerías síncronas sin soporte async — drivers de bases de datos sin versión async, librerías de procesamiento que hacen I/O interno — estarás bloqueando el event loop desde dentro, lo que anula todas las ventajas.
Antes de adoptar asyncio, hazte esta pregunta honesta: ¿el problema es realmente I/O-bound, o solo parece lento?
import asyncio
import time
import httpx # cliente HTTP con soporte async nativo
# ── Caso 1: I/O-bound — asyncio brilla ──────────────────────────────────
URLS = [
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/1",
]
async def fetch(client: httpx.AsyncClient, url: str) -> dict:
response = await client.get(url, timeout=10)
return {"url": url, "status": response.status_code}
async def fetch_all_async() -> None:
async with httpx.AsyncClient() as client:
# gather dispara todas las corrutinas concurrentemente
# el event loop las gestiona mientras esperan respuesta de red
tasks = [fetch(client, url) for url in URLS]
results = await asyncio.gather(*tasks)
for r in results:
print(r)
# ── Caso 2: CPU-bound — asyncio NO ayuda, bloquea el event loop ─────────
def cpu_heavy(n: int) -> int:
"""Simula trabajo CPU: calcular números primos."""
return sum(1 for i in range(2, n) if all(i % d != 0 for d in range(2, i)))
async def wrong_approach() -> None:
# ❌ Esta llamada bloquea el event loop entero durante su ejecución.
# Ninguna otra corrutina puede avanzar mientras cpu_heavy trabaja.
result = cpu_heavy(50_000)
print(f"Primos encontrados: {result}")
async def correct_approach() -> None:
loop = asyncio.get_running_loop()
# run_in_executor delega el trabajo CPU a un thread pool del SO,
# liberando el event loop para gestionar otras corrutinas.
result = await loop.run_in_executor(None, cpu_heavy, 50_000)
print(f"Primos encontrados: {result}")
# ── Caso 3: script simple — el overhead no vale ──────────────────────────
def just_use_requests() -> None:
"""
Si solo necesitas hacer UNA petición HTTP y terminar,
asyncio añade complejidad sin beneficio alguno.
"""
import urllib.request
with urllib.request.urlopen("https://httpbin.org/get") as resp:
print(resp.status)
# ── Benchmark ilustrativo ────────────────────────────────────────────────
def benchmark() -> None:
print("=== Benchmark I/O-bound ===")
# Versión síncrona: ~5 segundos (espera secuencial)
start = time.perf_counter()
import urllib.request
for url in URLS:
with urllib.request.urlopen(url) as r:
pass
sync_time = time.perf_counter() - start
print(f"Síncrono: {sync_time:.2f}s")
# Versión async: ~1 segundo (espera concurrente)
start = time.perf_counter()
asyncio.run(fetch_all_async())
async_time = time.perf_counter() - start
print(f"Asyncio: {async_time:.2f}s")
print(f"Speedup: {sync_time / async_time:.1f}x")
if __name__ == "__main__":
benchmark()
print("\n=== CPU-bound (correcto) ===")
asyncio.run(correct_approach())
Qué significa cada decisión
fetch_all_async usa asyncio.gather para despachar las cinco peticiones al mismo tiempo. El event loop registra los cinco sockets y, mientras el servidor remoto procesa cada petición, avanza en las demás. El tiempo total es aproximadamente el de la petición más lenta, no la suma de todas. Eso es exactamente la ventaja de I/O concurrente sin threading.
wrong_approach ilustra el error conceptual más común: meter trabajo CPU dentro de una corrutina asumiendo que asyncio lo va a paralelizar. No lo hace. cpu_heavy bloquea el hilo del event loop de principio a fin, igual que si no existiera el async def a su alrededor. La await solo cede control cuando llama a una función que internamente usa primitivas async del sistema operativo.
correct_approach muestra la solución real: run_in_executor envía la función CPU-bound a un ThreadPoolExecutor (el argumento None usa el executor por defecto del loop). El event loop puede seguir gestionando otras corrutinas mientras el thread del executor trabaja en paralelo, porque el GIL se libera para código C pesado o simplemente porque el trabajo está en otro thread.
just_use_requests no es código irreal — es un recordatorio de que la complejidad tiene que justificarse. Si tu script hace una llamada y termina, el overhead de asyncio (instanciar el event loop, gestionar corrutinas, manejar excepciones async) es puro ruido. Las librerías síncronas estándar son más simples, más debuggeables y suficientemente rápidas para ese caso.
Errores que debes conocer
Error: Mezclar una librería síncrona bloqueante dentro de una corrutina, pensando que await la hará no-bloqueante.
import asyncio
import time
# ❌ Wrong
async def fetch_blocking():
time.sleep(2) # bloquea el event loop; ninguna otra corrutina avanza
return "done"
# ✅ Right
async def fetch_non_blocking():
await asyncio.sleep(2) # cede control al event loop durante la espera
return "done"
time.sleep es una llamada síncrona al SO que detiene el hilo; asyncio.sleep es una primitiva del event loop que suspende solo la corrutina actual.
Error: Usar asyncio.run() dentro de una función que ya está siendo ejecutada en un event loop activo.
import asyncio
async def inner():
return 42
# ❌ Wrong — lanza RuntimeError: "This event loop is already running"
async def outer():
result = asyncio.run(inner()) # no puedes arrancar un loop dentro de otro
return result
# ✅ Right
async def outer_correct():
result = await inner() # simplemente espera la corrutina directamente
return result
asyncio.run crea y cierra un event loop completo; si ya estás dentro de uno (por ejemplo, en un notebook de Jupyter o en otro async def), el anidamiento no está permitido — usa directamente await.
Error: Adoptar asyncio por un único cuello de botella sin evaluar si el resto del stack tiene soporte async.
# ❌ Wrong — el driver síncrono bloquea el event loop en cada query
import asyncio
import psycopg2 # driver síncrono
async def get_users():
conn = psycopg2.connect("dbname=mydb user=postgres")
cur = conn.cursor()
cur.execute("SELECT * FROM users") # bloquea el event loop
return cur.fetchall()
# ✅ Right — usa un driver con soporte async nativo
import asyncpg
async def get_users_correct():
conn = await asyncpg.connect("postgresql://postgres@localhost/mydb")
rows = await conn.fetch("SELECT * FROM users")
await conn.close()
return rows
El problema no es solo rendimiento — es que con psycopg2 estás pagando toda la complejidad de asyncio sin obtener ninguno de sus beneficios de concurrencia.
N° 153