Gestionar Trabajo Concurrente con Thread Pools y Executors

Visión general

Crear un nuevo thread para cada tarea concurrente es costoso. Cada thread consume memoria para su stack (típicamente 1MB), requiere scheduling del SO, y agrega overhead de context-switching. En alta concurrencia, la creación de threads se convierte en cuello de botella — el sistema pasa más tiempo gestionando threads que haciendo trabajo útil. Los thread pools resuelven esto manteniendo un conjunto fijo de worker threads reutilizables. Las tareas se envían a una cola; workers inactivos las toman. Cuando todos los workers están ocupados, las tareas esperan en la cola en lugar de spawnear nuevos threads.

El desafío es dimensionar el pool correctamente y manejar sobrecarga. Una tarea CPU-bound en una máquina de 8 cores se beneficia de 8 threads — más threads solo compiten por cores. Una tarea I/O-bound se beneficia de más threads que cores porque los threads pasan la mayor parte del tiempo esperando disco o red. Cuando la cola se llena, el pool debe decidir si rechazar tareas, bloquear al submitter, o ejecutarlas en el thread del llamador. Aqui se explica como dimensionamiento de pools, patrones de executors y estrategias de rechazo en Java, Python y C#.

Cuándo usarlo

Usa esta receta cuando:

• Procesando un alto volumen de tareas independientes concurrentemente
• Ejecutando computaciones CPU-bound (procesamiento de imágenes, transformación de datos, inferencia ML)
• Ejecutando operaciones I/O-bound donde los threads pasan tiempo esperando (llamadas API, lecturas de archivo). Consulta Async Patterns para alternativas no bloqueantes.
• Limitando uso de recursos para prevenir agotamiento de threads o presión de memoria. Consulta Locks y Mutexes para coordinar acceso compartido.
• Construyendo colas de trabajo donde las tareas deben ejecutarse asíncronamente del submitter

Solución

Java Executors (Fixed Thread Pool)

import java.util.concurrent.*;

public class ImageProcessor {
    private final ExecutorService executor;

    public ImageProcessor(int poolSize) {
        this.executor = Executors.newFixedThreadPool(poolSize);
    }

    public CompletableFuture<String> processAsync(String imageId) {
        return CompletableFuture.supplyAsync(() -> {
            return processImage(imageId);
        }, executor);
    }

    public void shutdown() {
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }
    }
}

ExecutorService executor = new ThreadPoolExecutor(
    4, 8, 30L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(100),
    new ThreadFactory() {
        private final AtomicInteger counter = new AtomicInteger(0);
        public Thread newThread(Runnable r) {
            return new Thread(r, "worker-" + counter.incrementAndGet());
        }
    },
    new ThreadPoolExecutor.CallerRunsPolicy()
);

Python concurrent.futures

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def fetch_url(url):
    response = requests.get(url, timeout=10)
    return response.json()

urls = [
    "https://api.example.com/users/1",
    "https://api.example.com/users/2",
    "https://api.example.com/users/3",
]

with ThreadPoolExecutor(max_workers=20) as executor:
    futures = {executor.submit(fetch_url, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            data = future.result()
            print(f"Fetched {url}: {data}")
        except Exception as e:
            print(f"Failed {url}: {e}")

from concurrent.futures import ProcessPoolExecutor

def process_data(chunk):
    return sum(x ** 2 for x in chunk)

data = [range(0, 1000000), range(1000000, 2000000)]
with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(process_data, data))

C# Task Parallel Library

public class WorkerPool
{
    public async Task ProcessBatchAsync(IEnumerable<string> items)
    {
        var semaphore = new SemaphoreSlim(10);

        var tasks = items.Select(async item =>
        {
            await semaphore.WaitAsync();
            try { return await ProcessItemAsync(item); }
            finally { semaphore.Release(); }
        });

        var results = await Task.WhenAll(tasks);
    }

    private async Task<string> ProcessItemAsync(string item)
    {
        await Task.Delay(100);
        return $"Processed: {item}";
    }
}

ThreadPool.SetMinThreads(4, 4);
ThreadPool.SetMaxThreads(8, 8);

Go Worker Pool (Goroutines + Channels)

package main

import (
    "fmt"
    "sync"
)

func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {
        results <- j * j
    }
}

func main() {
    numWorkers := 4
    jobs := make(chan int, 100)
    results := make(chan int, 100)
    var wg sync.WaitGroup

    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

    for j := 1; j <= 20; j++ {
        jobs <- j
    }
    close(jobs)

    go func() {
        wg.Wait()
        close(results)
    }()

    for r := range results {
        fmt.Println("result:", r)
    }
}

Las goroutines de Go son ligeras (2KB de stack vs 1MB para OS threads), por lo que puedes spawnear miles. Para trabajo CPU-bound, usa runtime.GOMAXPROCS(runtime.NumCPU()) para limitar el paralelismo. Para trabajo I/O-bound, un mayor numero de workers esta bien ya que las goroutines bloqueadas ceden a otras.

Explicación

• Core vs máximo pool size: el core size es el número de threads mantenidos vivos incluso cuando están inactivos. El máximo es el límite superior. Cuando las tareas exceden el core size, nuevos threads se crean hasta el máximo. Threads por encima del core se terminan después del timeout de keep-alive si están inactivos. Esto permite al pool escalar entre una línea base y un pico.
• Cola de trabajo: las tareas enviadas cuando todos los threads están ocupados esperan en una cola. Una cola ilimitada (LinkedBlockingQueue) acepta tareas infinitas pero riesga OutOfMemoryError. Una cola acotada limita memoria pero requiere una política de rechazo cuando se llena.
• Políticas de rechazo: cuando el pool y la cola están saturados, Java ofrece cuatro políticas. AbortPolicy (default) lanza excepción. CallerRunsPolicy ejecuta la tarea en el thread del llamador, ralentizando submission. DiscardPolicy descarta la tarea silenciosamente. DiscardOldestPolicy descarta la tarea más antigua en cola.
• Thread-per-task vs pools: crear un thread por tarea funciona para algunas pocas docenas de operaciones concurrentes. A cientos o miles, el overhead de creación de threads domina. Los pools amortizan el costo de creación a través de la vida de la aplicación y proveen uso acotado de recursos.

Variantes

Tipo de pool	Core threads	Max threads	Cola	Mejor para
Fixed	N	N	Ilimitada	Trabajo CPU steady-state
Cached	0	Ilimitado	Síncrona	Burst I/O, tareas de corta vida
Single	1	1	Ilimitada	Ejecución ordenada
Scheduled	N	N	Cola delayed	Tareas temporizadas/recurrentes
Work stealing	CPU count	CPU count	Deque por thread	Paralelismo fork-join

Lo que funciona

• Dimensiona pools CPU al número de cores: getRuntime(). availableProcessors()oos. cpu_count()`. Threads adicionales solo compiten por cores, causando context switches sin ganancias de throughput. Consulta Load Balancing para distribuir trabajo entre cores.
• Dimensiona pools I/O más alto que core count: para trabajo I/O-bound, los threads se bloquean en red/disco. Usa 2x-4x core count para pools I/O, dependiendo de latencia.
• Siempre shutdown gracefulmente: un executor no terminado filtra threads y previene salida del proceso JVM/Python. Llama shutdown(), espera terminación, luego shutdownNow() si es necesario.
• Usa colas acotadas con políticas de rechazo: las colas ilimitadas ocultan backpressure. Un sistema que acepta tareas infinitas eventualmente se cae. Consulta Rate Limiting para gestionar sobrecarga.
• Nombra tus threads: debuggear un thread dump de 50 threads sin nombre es imposible. Esto hace profiling, logging y debugging triviales.
• Monitorea metricas del pool: rastrea threads activos, tamano de cola, tareas completadas y conteo de rechazos. ThreadPoolExecutor de Java expone estos via getters. En Python, envuelve el executor para rastrear submissions y completions. Alerta cuando la profundidad de cola excede un umbral.
• Usa shutdownNow() con cuidado: shutdownNow() interrumpe threads en ejecucion. Si tus tareas no chequean Thread. interrupted() o manejan InterruptedException, continuaran ejecutandose. Disena tareas cooperativas y responsivas a interrupcion.

Errores comunes

• Bloquear al llamador con Future.get() sin timeout: future. get() espera indefinidamente. Si el worker thread se cuelga (loop infinito, deadlock), el llamador se cuelga para siempre. get(timeout, TimeUnit. SECONDS)`.
• Usar threads para trabajo CPU-bound en Python: el GIL de Python previene paralelismo real de threads para trabajo CPU. Un ThreadPoolExecutor con 8 threads en una máquina de 8 cores ejecuta tareas secuencialmente, no en paralelo.
• Ignorar excepciones en tareas fire-and-forget: enviar una tarea e ignorar el future traga excepciones. La tarea falla silenciosamente.
• Crear un nuevo pool por request: un web handler que crea un nuevo ExecutorService para cada request entrante derrota el propósito. Crea un pool en el startup de la aplicación y reutilízalo. Pásalo como dependencia a los handlers.
• Compartir un solo pool entre workloads no relacionados: las tareas CPU-bound e I/O-bound tienen diferentes tamaños óptimos de pool. Si comparten un pool, un workload priva al otro.
• No manejar RejectedExecutionException: cuando usas AbortPolicy, el pool lanza RejectedExecutionException bajo sobrecarga. Si no lo capturas, la excepción propaga y puede crashear al llamador. Captúralo y degrada gracefulmente.

Lectura Adicional

• Documentación oficial: consulta la referencia actualizada del framework o herramienta utilizada.
• Guías relacionadas: explora las guías de concurrency y async para profundizar.
• Patrones complementarios: revisa los patrones de diseño aplicables a tu stack tecnológico.
• Postmortems públicos: estudia incidentes reales de equipos que enfrentaron problemas similares en producción.

Notas de Producción

• Despliega gradualmente usando canary o blue-green para detectar regresiones temprano.
• Configura alertas para errores, latencia p99 y tasa de fallos antes de habilitar en producción.
• Documenta el rollback en el runbook; prueba el procedimiento en staging al menos una vez por trimestre.
• Revisa logs estructurados con correlation IDs para trazar requests end-to-end en incidentes.

Puntos Clave

• Aplica gestionar trabajo concurrente con thread pools y executors cuando necesites una solución práctica para tu caso de uso.
• Monitorea el rendimiento después de implementar; mide latencia, errores y uso de recursos antes y después.
• Revisa la sección de Troubleshooting ante errores comunes; la mayoría tienen causa raíz documentada con solución.
• Mantén dependencias actualizadas y ejecuta tests en CI para prevenir regresiones en producción.

Troubleshooting

• Race conditions appear under load: protect shared state with locks, atomics, or message passing. Reproduce with targeted stress tests.
• Deadlock between workers: establish a consistent lock acquisition order and keep critical sections short.
• Thread pool saturation: Increase pool size only if CPU and memory allow.
• Actor mailbox grows unbounded: apply backpressure, bounded queues, and load shedding.
• Async task never completes: check for unhandled promise rejections, forgotten awaits, and infinite loops in cooperative scheduling.

Errores Comunes en Producción

• Copiar el ejemplo sin adaptarlo a volúmenes y modos de fallo reales.
• Saltar tests de carga e inyección de errores antes del primer despliegue productivo.
• Codificar valores fijos que deberían ser configurables por entorno.
• Olvidar agregar logging y monitoreo en cada paso.
• Desplegar sin plan de rollback ni estrategia de backup probada.
• Asumir que el ejemplo mínimo escalará sin agregar caché o procesamiento por lotes.
• No documentar la versión y configuración usadas en producción.
• Dejar la receta sin cambios cuando evolucionan las dependencias o la escala.