Python ofrece múltiples mecanismos para ejecutar tareas de forma concurrente. Aunque ambos —hilos y procesos— permiten superponer operaciones, sus características fundamentales, limitaciones y escenarios ideales difieren sustancialmente. Esta guía explora su implementación realista, con ejemplos reescritos, análisis de rendimiento y recomendaciones basadas en el tipo de carga de trabajo.

Hilos: Concurrencia ligera con restricciones

Los hilos comparten el espacio de memoria del proceso padre, lo que facilita el intercambio de datos pero impone una limitación crítica: el Global Interpreter Lock (GIL) en CPython. Este mecanismo evita la ejecución simultánea de bytecodes de Python, haciendo ineficaz el uso de hilos para cargas intensivas en CPU.

No obstante, son altamente efectivos en tareas bloqueantes como operaciones de E/S (redes, archivos, bases de datos), donde el hilo cede el control mientras espera una respuesta externa.

A continuación, un ejemplo revisado que ilustra la concurrencia con hilos, usando nombres de variables descriptivos y estructura modular:

import threading
import time

def fetch_data_from_source(name: str, delay_ms: float) -> None:
    for idx in range(3):
        print(f"[{name}] → Etapa {idx + 1}")
        time.sleep(delay_ms / 1000)

# Configuración de hilos simulando llamadas asíncronas
source_a = threading.Thread(target=fetch_data_from_source, args=("API-Usuario", 150))
source_b = threading.Thread(target=fetch_data_from_source, args=("BD-Reportes", 200))

source_a.start()
source_b.start()

source_a.join()
source_b.join()
print("✅ Ambas fuentes finalizaron.")

Procesos: Paralelismo real sin GIL

Al contrario de los hilos, cada proceso tiene su propio intérprete de Python y memoria aislada. Esto elimina la barrera del GIL y permite aprovechar todos los núcleos físicos disponibles —ideal para cálculos pesados como procesamiento numérico, cifrado o transformaciones masivas de datos.

La desventaja radica en el mayor costo de creación, comunicación interproceso (requiere serialización mediente Queue, Pipe o memoria compartida) y consumo de recursos.

Ejemplo refactorizado con lógica clara y manejo explícito de resultados:

import multiprocessing as mp
from math import sqrt

def compute_prime_squares(limit: int) -> list[int]:
    primes = []
    for num in range(2, limit + 1):
        if all(num % i != 0 for i in range(2, int(sqrt(num)) + 1)):
            primes.append(num ** 2)
    return primes[:5]  # Solo primeros 5 cuadrados

if __name__ == "__main__":
    # Ejecución paralela de dos cálculos independientes
    task_a = mp.Process(target=compute_prime_squares, args=(100,))
    task_b = mp.Process(target=compute_prime_squares, args=(200,))

    task_a.start()
    task_b.start()

    task_a.join()
    task_b.join()
    print("✅ Procesamiento matemático completado en paralelo.")

Abstracción de alto nivel: concurrent.futures

El módulo concurrent.futures simplifica drásticamente la gestión de concurrencia al ofrecer una interfaz uniforme para hilos y procesos. En lugar de manipular objetos de bajo nivel (Thread, Process), se trabaja con Executors que gestionan automáticamente pools, colas y ciclo de vida.

Este enfoque mejora la legibilidad, reduce errores comunes (como olvidar join()) y facilita el cambio entre estrategias según el perfil de la carga:

import concurrent.futures
import time

def simulate_io_task(task_id: int, duration_sec: float) -> str:
    time.sleep(duration_sec)
    return f"Tarea-{task_id} completada tras {duration_sec}s"

def run_with_executor(executor_class, max_workers: int):
    with executor_class(max_workers=max_workers) as executor:
        # Envío simultáneo de tareas
        futures = [
            executor.submit(simulate_io_task, 1, 0.8),
            executor.submit(simulate_io_task, 2, 1.2),
            executor.submit(simulate_io_task, 3, 0.5),
        ]
        
        # Recolección ordenada de resultados
        results = [f.result() for f in futures]
        return results

# Uso con ThreadPoolExecutor (recomendado para E/S)
io_results = run_with_executor(concurrent.futures.ThreadPoolExecutor, 3)

# Uso con ProcessPoolExecutor (recomendado para CPU)
cpu_results = run_with_executor(concurrent.futures.ProcessPoolExecutor, 2)