El módulo concurrent.futures ofrece una interfaz de alto nivel para ejecutar funciones de manera concurrente. En lugar de crear y coordinar threads o procesos manualmente, una aplicación envía llamadas a un executor y recibe objetos Future que representan resultados que todavía pueden no estar listos. El mismo modelo sirve para executors basados en threads y procesos.
La API simplifica colas, recopilación de resultados, propagación de excepciones, timeouts, cancelación y shutdown. No elimina las dificultades de la concurrencia: todavía debes elegir el executor correcto, limitar trabajo pendiente, evitar deadlocks, proteger estado compartido y decidir cómo una falla afecta al resto de la operación.
Executor y Future
Un Executor acepta trabajo. submit() devuelve un Future. El objeto permite consultar estado, esperar, obtener un valor, observar una excepción o cancelar trabajo que aún no comenzó.
from concurrent.futures import ThreadPoolExecutor
def cuadrado(numero):
return numero * numero
with ThreadPoolExecutor(max_workers=4) as executor:
futuro = executor.submit(cuadrado, 12)
print(futuro.result())
El bloque with llama a shutdown() al terminar y aplica la política normal de espera. Esto evita pools abandonados.
ThreadPoolExecutor
ThreadPoolExecutor ejecuta threads dentro del mismo proceso. Suele ser adecuado para operaciones que pasan gran parte del tiempo esperando I/O: HTTP, archivos, bases de datos, subprocesses, brokers y APIs externas.
Las threads comparten memoria, por lo que pasar objetos es sencillo, pero el estado mutable compartido crea condiciones de carrera. Usa queues, locks, datos inmutables o reglas de ownership.
Trabajo limitado por CPU
Para funciones que ejecutan bytecode Python continuamente, varias threads normalmente no ofrecen paralelismo completo en múltiples núcleos en builds tradicionales del intérprete. Un pool de procesos puede encajar mejor.
Extensiones nativas que liberan el lock del intérprete pueden escalar con threads. Mide el workload real.
ProcessPoolExecutor
ProcessPoolExecutor usa procesos separados y puede distribuir trabajo de CPU entre varios núcleos. Argumentos y resultados deben ser serializables, y las funciones enviadas deben poder importarse en los workers.
from concurrent.futures import ProcessPoolExecutor
def calcular(numero):
return sum(i * i for i in range(numero))
if __name__ == "__main__":
with ProcessPoolExecutor() as executor:
futuros = [executor.submit(calcular, n) for n in (10000, 20000, 30000)]
print([f.result() for f in futuros])
La protección if __name__ == "__main__" es esencial en plataformas que inician workers importando nuevamente el módulo principal.
Elige la cantidad de workers
Más workers no implican automáticamente más rendimiento. Demasiadas threads aumentan context switching, conexiones y presión sobre servicios externos. Demasiados procesos aumentan memoria, serialización, startup y competencia por CPU.
Basa el límite en el recurso más restringido: núcleos, pool de base, rate limits, ancho de banda, memoria, descriptores o capacidad del sistema downstream.
Envía tareas individuales
submit(funcion, *args, **kwargs) ofrece control sobre cada tarea.
with ThreadPoolExecutor(max_workers=8) as executor:
futuros = {
executor.submit(descargar, url, timeout=10): url
for url in urls
}
Relacionar futuros con sus entradas facilita diagnóstico y retries selectivos.
Usa map para transformaciones ordenadas
executor.map() aplica una función a varias entradas y devuelve resultados en el orden de entrada.
with ThreadPoolExecutor(max_workers=4) as executor:
resultados = list(executor.map(procesar, items))
El orden es cómodo, pero una tarea lenta puede retrasar resultados posteriores ya terminados. Usa as_completed() cuando prefieras orden de finalización.
Consume con as_completed
as_completed() produce futuros a medida que terminan.
from concurrent.futures import as_completed
for futuro in as_completed(futuros):
origen = futuros[futuro]
try:
resultado = futuro.result()
except Exception as error:
registrar_fallo(origen, error)
else:
guardar(resultado)
Una excepción del worker vuelve a ser lanzada por result() en la thread o proceso que recopila el resultado.
Espera grupos
wait() puede esperar todas las tareas, la primera finalizada o la primera excepción. Devuelve conjuntos de futuros concluidos y pendientes.
from concurrent.futures import wait, FIRST_EXCEPTION
concluidos, pendientes = wait(futuros, return_when=FIRST_EXCEPTION)
Después de una falla, decide explícitamente si el trabajo pendiente debe continuar, cancelarse o terminar para mantener consistencia.
Timeouts
future.result(timeout=...), wait() y as_completed() limitan cuánto espera el caller. Un timeout de espera no detiene automáticamente la función que ya está en ejecución.
from concurrent.futures import TimeoutError
try:
valor = futuro.result(timeout=5)
except TimeoutError:
registrar_aviso("la tarea superó el límite de espera")
La propia tarea debe aplicar timeouts a sockets, bases, subprocesses y APIs para poder finalizar.
Cancelación
cancel() funciona solo antes de que una tarea comience. cancelled() informa el éxito y running() indica que la ejecución empezó.
Detener trabajo ya iniciado requiere cooperación con un Event, deadline, token o señal compartida.
def procesar(items, detener):
for item in items:
if detener.is_set():
return None
ejecutar_paso(item)
Shutdown
shutdown(wait=True) cierra el executor y normalmente espera tareas enviadas. Cancelar futuros pendientes puede ser útil durante una falla o el cierre de la aplicación.
No dejes indefinida la vida de un executor global. Establece quién lo crea, quién lo cierra y qué ocurre con operaciones incompletas.
Deadlocks en pools de threads
Puede ocurrir un deadlock cuando una tarea espera otra del mismo pool y todos los workers ya están ocupados.
def tarea_a():
return futuro_b.result()
Evita dependencias bloqueantes entre tareas del mismo executor. Coordina secuencias fuera del pool, usa callbacks o combina resultados en la thread que envía.
La trampa de un solo worker
Con un único worker, una tarea que envía otra al mismo executor y llama a result() nunca libera el worker necesario para ejecutar la segunda.
Aumentar la cantidad solo oculta el problema estructural. Elimina el ciclo.
Backpressure
Enviar millones de tareas de una vez consume memoria para argumentos, futuros y queues internas. Produce trabajo en lotes o mantiene una ventana limitada.
def ejecutar_lotes(executor, items, tamano=100):
lote = []
for item in items:
lote.append(executor.submit(procesar, item))
if len(lote) == tamano:
for futuro in lote:
yield futuro.result()
lote.clear()
Una ventana deslizante puede enviar un reemplazo cada vez que termina un futuro.
Inicialización de workers
Los executors pueden configurar cada worker mediante un initializer. Si falla, el pool puede quedar inutilizable y los futuros pendientes reciben un error de executor roto.
Mantén la inicialización corta, determinística y, cuando sea posible, independiente de servicios de red frágiles.
Estado global y procesos
Los procesos no comparten objetos Python normales. Cada worker tiene sus propios módulos y globals. Cambiar una variable global del worker no actualiza automáticamente al padre.
Pasa entradas de forma explícita y devuelve resultados. Para la capa de procesos del sistema, consulta posix en Python y usa abstracciones de multiprocessing.
Serialización
Funciones locales, lambdas, generators abiertos, locks, sockets y conexiones activas normalmente no pueden enviarse a un pool de procesos. Define funciones a nivel de módulo y pasa datos simples.
Los objetos grandes aumentan el coste de copia. Puede ser mejor que cada worker abra su recurso desde un identificador.
Callbacks
add_done_callback() registra una función ejecutada después de la finalización.
def al_terminar(futuro):
try:
futuro.result()
except Exception:
metricas.fallo()
else:
metricas.exito()
futuro.add_done_callback(al_terminar)
Los callbacks deben ser cortos y no bloquear. Publica un evento en una queue para trabajo mayor.
Contexto y logging
Las threads comparten logging, pero el contexto de una request puede no propagarse automáticamente. Los procesos necesitan configuración propia o logging basado en queue.
Incluye un identificador de tarea para conectar entrada, intento, duración, worker y excepción.
Retries
Los executors no repiten llamadas fallidas automáticamente. Reintenta solo fallos probablemente transitorios, con límite, demora, jitter e idempotencia.
Los errores de validación no mejoran al repetirse. Una escritura parcial puede duplicar datos sin clave idempotente.
Seguridad
No envíes funciones o payloads serializados proporcionados por usuarios no confiables a un pool de procesos. La serialización Python no es un formato seguro para entrada hostil.
Limita la concurrencia para no provocar denial of service contra tu propia base, filesystem, red o API externa.
Pruebas
Prueba éxito, excepción, timeout, cancelación antes del inicio, shutdown, pool roto, finalización desordenada, entrada vacía, tareas lentas e interrupción. Usa pocos workers para exponer dependencias ocultas.
Evita assertions basadas en timing exacto. Sincroniza con eventos y estado observable.
Errores comunes
Los fallos frecuentes son usar threads para CPU sin medir, crear procesos sin la protección del módulo principal, llamar result() desde tareas del mismo pool, enviar trabajo ilimitado, pensar que un timeout mata la ejecución, ignorar excepciones, pasar objetos no serializables y olvidar cerrar el executor.
Conclusión
concurrent.futures ofrece una capa uniforme orientada a tareas. Usa threads para I/O con espera, procesos para CPU cuando corresponda, limita workers, aplica backpressure y trata cada futuro como una operación que puede fallar, expirar o cancelarse.
Consulta la documentación oficial de concurrent.futures y signal en Python para shutdown coordinado.
