Procesamiento paralelo y concurrente en Python: threading, multiprocessing y asyncio

Procesamiento paralelo y concurrente en Python: threading, multiprocessing y asyncio

Python ofrece tres enfoques principales para ejecutar código de forma concurrente o paralela. Elegir el correcto depende del tipo de tarea: IO-bound (esperar disco, red, APIs) o CPU-bound (cálculo intensivo).

1. El GIL y cuándo importa

El Global Interpreter Lock (GIL) de CPython permite ejecutar solo un hilo Python a la vez. Esto afecta a:

• CPU-bound: el GIL impide el paralelismo real entre hilos → usa multiprocessing.
• IO-bound: cuando un hilo espera IO, el GIL se libera → threading funciona bien.

Tarea IO-bound (red, disco)   → threading o asyncio
Tarea CPU-bound (cálculo)     → multiprocessing o concurrent.futures
Muchas tareas IO asíncronas   → asyncio (un solo hilo, muy eficiente)

2. threading: hilos para tareas IO-bound

import threading
import time
import requests

def descargar(url, resultados, indice):
    r = requests.get(url, timeout=10)
    resultados[indice] = len(r.content)
    print(f"[{indice}] {url} → {len(r.content)} bytes")

urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
]

resultados = [None] * len(urls)
hilos = []

inicio = time.perf_counter()
for i, url in enumerate(urls):
    h = threading.Thread(target=descargar, args=(url, resultados, i))
    hilos.append(h)
    h.start()

for h in hilos:
    h.join()   # Esperar a que todos terminen

print(f"Total: {time.perf_counter() - inicio:.2f} s")  # ~1 s en vez de ~3 s
print(f"Bytes descargados: {resultados}")

Thread con Lock (datos compartidos)

import threading

contador = 0
lock = threading.Lock()

def incrementar(n):
    global contador
    for _ in range(n):
        with lock:      # Solo un hilo a la vez modifica contador
            contador += 1

hilos = [threading.Thread(target=incrementar, args=(10000,)) for _ in range(5)]
for h in hilos: h.start()
for h in hilos: h.join()
print(f"Contador final: {contador}")  # 50000 (sin lock daría un valor menor)

3. concurrent.futures.ThreadPoolExecutor

La API de alto nivel más sencilla para threading:

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def fetch(url):
    r = requests.get(url, timeout=10)
    return url, r.status_code, len(r.content)

urls = [f"https://httpbin.org/anything/{i}" for i in range(8)]

with ThreadPoolExecutor(max_workers=4) as executor:
    futuros = {executor.submit(fetch, url): url for url in urls}

    for futuro in as_completed(futuros):
        url, status, size = futuro.result()
        print(f"{status} | {size:6d} bytes | {url}")

map() para resultados en orden

from concurrent.futures import ThreadPoolExecutor

def procesar(n):
    return n * n

with ThreadPoolExecutor(max_workers=4) as ex:
    resultados = list(ex.map(procesar, range(10)))

print(resultados)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

4. multiprocessing: procesos para CPU-bound

import multiprocessing
import time

def es_primo(n):
    if n < 2:
        return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0:
            return False
    return True

numeros = list(range(100_000, 100_200))

# Sin paralelismo
inicio = time.perf_counter()
resultados_seq = [es_primo(n) for n in numeros]
print(f"Secuencial: {time.perf_counter() - inicio:.3f} s")

# Con Pool de procesos
inicio = time.perf_counter()
with multiprocessing.Pool() as pool:
    resultados_par = pool.map(es_primo, numeros)
print(f"Paralelo  : {time.perf_counter() - inicio:.3f} s")

primos = [n for n, p in zip(numeros, resultados_par) if p]
print(f"Primos encontrados: {len(primos)}")

Pool.starmap para múltiples argumentos

from multiprocessing import Pool

def potencia(base, exp):
    return base ** exp

pares = [(2, 10), (3, 8), (5, 6), (7, 5)]

with Pool(processes=4) as pool:
    resultados = pool.starmap(potencia, pares)

print(resultados)  # [1024, 6561, 15625, 16807]

5. concurrent.futures.ProcessPoolExecutor

La API de alto nivel para multiprocessing:

from concurrent.futures import ProcessPoolExecutor, as_completed
import math

def calcular_factorial(n):
    return n, math.factorial(n)

numeros = [1000, 2000, 3000, 4000, 5000]

with ProcessPoolExecutor() as executor:
    futuros = {executor.submit(calcular_factorial, n): n for n in numeros}
    for futuro in as_completed(futuros):
        n, resultado = futuro.result()
        print(f"{n}! tiene {len(str(resultado))} dígitos")

6. asyncio: concurrencia asíncrona para IO

asyncio usa un solo hilo y corutinas para manejar miles de operaciones IO concurrentes:

import asyncio
import aiohttp   # pip install aiohttp
import time

async def fetch(session, url):
    async with session.get(url) as r:
        contenido = await r.read()
        return url, r.status, len(contenido)

async def main():
    urls = [f"https://httpbin.org/anything/{i}" for i in range(10)]

    async with aiohttp.ClientSession() as session:
        tareas = [fetch(session, url) for url in urls]
        resultados = await asyncio.gather(*tareas)

    for url, status, size in resultados:
        print(f"{status} | {size:5d} bytes | {url}")

inicio = time.perf_counter()
asyncio.run(main())
print(f"Total: {time.perf_counter() - inicio:.2f} s")

asyncio.gather vs asyncio.wait

import asyncio

async def tarea(nombre, segundos):
    await asyncio.sleep(segundos)
    print(f"[{nombre}] listo tras {segundos}s")
    return nombre

async def demo():
    # gather: espera todas, resultados en orden
    resultados = await asyncio.gather(
        tarea("A", 1),
        tarea("B", 2),
        tarea("C", 1),
    )
    print(f"Resultados: {resultados}")

    # wait: control más fino (FIRST_COMPLETED, etc.)
    tareas = [asyncio.create_task(tarea(f"T{i}", i)) for i in range(1, 4)]
    done, pending = await asyncio.wait(tareas, return_when=asyncio.FIRST_COMPLETED)
    for t in done:
        print(f"Primera en terminar: {t.result()}")
    for t in pending:
        t.cancel()

asyncio.run(demo())

7. asyncio + archivos con aiofiles

import asyncio
import aiofiles   # pip install aiofiles

async def leer_archivo(ruta):
    async with aiofiles.open(ruta, 'r', encoding='utf-8') as f:
        contenido = await f.read()
    return ruta, len(contenido)

async def procesar_varios():
    rutas = ["archivo1.txt", "archivo2.txt", "archivo3.txt"]
    tareas = [leer_archivo(r) for r in rutas]
    resultados = await asyncio.gather(*tareas, return_exceptions=True)
    for res in resultados:
        if isinstance(res, Exception):
            print(f"Error: {res}")
        else:
            ruta, tam = res
            print(f"{ruta}: {tam} bytes")

asyncio.run(procesar_varios())

8. Queue para pipelines productor-consumidor

import asyncio

async def productor(queue: asyncio.Queue, items):
    for item in items:
        await queue.put(item)
        print(f"Producido: {item}")
        await asyncio.sleep(0.1)
    # Señal de fin
    await queue.put(None)

async def consumidor(queue: asyncio.Queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"  Consumido: {item}")
        await asyncio.sleep(0.2)

async def main():
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        productor(queue, range(10)),
        consumidor(queue),
    )

asyncio.run(main())

9. Manejo de errores en tareas paralelas

from concurrent.futures import ThreadPoolExecutor, as_completed

def puede_fallar(n):
    if n == 3:
        raise ValueError(f"Error en {n}")
    return n * 2

with ThreadPoolExecutor(max_workers=4) as ex:
    futuros = {ex.submit(puede_fallar, i): i for i in range(6)}
    for futuro in as_completed(futuros):
        try:
            resultado = futuro.result()
            print(f"OK: {resultado}")
        except Exception as e:
            print(f"ERROR: {e}")

10. ¿Cuál usar? Guía rápida

11. asyncio en un executor para código bloqueante

import asyncio
from concurrent.futures import ProcessPoolExecutor
import time

def tarea_cpu(n):
    """Función bloqueante (CPU-bound)."""
    return sum(i*i for i in range(n))

async def main():
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as executor:
        # Ejecutar CPU-bound sin bloquear el event loop
        resultado = await loop.run_in_executor(executor, tarea_cpu, 1_000_000)
    print(f"Resultado: {resultado}")

asyncio.run(main())

12. Buenas prácticas

• No compartas estado mutable entre procesos (multiprocessing); usa Queue o Manager si es necesario.
• Limita el pool al número de CPUs disponibles (os.cpu_count()); más workers no siempre es mejor.
• asyncio no es multihilo: await no bloquea el event loop, pero el código síncrono dentro de corutinas sí.
• Usa as_completed cuando quieras procesar resultados en cuanto lleguen, no al final.
• Cancela tareas pendientes en asyncio con task.cancel() para evitar resource leaks.
• Perfila primero: antes de paralelizar, mide dónde está el cuello de botella real.

Conversiones relacionadas

Conversiones frecuentes del catálogo:

• JPG a PNG
• MP4 a MP3
• PDF a DOCX
• PNG a JPG
• MP3 a WAV
• DOCX a PDF