Introducción a Java ThreadPools

Cómo funciona un Pool de Hilos (Thread Pool)

En lugar de iniciar un nuevo hilo para cada tarea que se ejecutará concurrentemente, la tarea puede pasarse a un pool de hilos. Tan pronto como el pool tenga hilos inactivos, la tarea se asigna a uno de ellos y se ejecuta. Internamente, las tareas se insertan en una Cola Bloqueante de la cual los hilos en el pool están extrayendo elementos. Cuando se inserta una nueva tarea en la cola, uno de los hilos inactivos la extraerá con éxito y la ejecutará. El resto de los hilos inactivos en el pool quedarán bloqueados esperando para extraer tareas.

Casos de uso del Thread Pool

Los pools de hilos se utilizan a menudo en servidores multihilo. Cada conexión que llega al servidor a través de la red se envuelve como una tarea y se pasa a un pool de hilos. Los hilos en el pool procesarán las solicitudes en las conexiones de forma concurrente.

Thread Pool incorporado en Java

Java viene con pools de hilos incorporados en el paquete java.util.concurrent, por lo que no es necesario implementar tu propio pool de hilos. Podemos leer más sobre esto en la documentación de java.util.concurrent.ExecutorService. Aun así, puede ser útil conocer un poco sobre la implementación de un pool de hilos.

Implementación de un Thread Pool en Java

Acá vemos una implementación simple de un pool de hilos. La implementación utiliza la BlockingQueue estándar de Java que viene incluida desde Java 5.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ThreadPool {

    private BlockingQueue<Runnable> colaDetareas = null;
    private List<PoolThreadRunnable> runnables = new ArrayList<>();
    private boolean estaDetenido = false;

    public ThreadPool(int numDeHilos, int maxNumDeTareas){
        colaDetareas = new ArrayBlockingQueue<Runnable>(maxNumDeTareas);

        for(int i=0; i<numDeHilos; i++){
            PoolThreadRunnable poolThreadRunnable =
                    new PoolThreadRunnable(colaDetareas);

            runnables.add(poolThreadRunnable);
        }
        for(PoolThreadRunnable ejecutable : runnables){
            new Thread(ejecutable).start();
        }
    }

    public synchronized void execute(Runnable tarea) throws Exception{
        if(this.estaDetenido) throw
                new IllegalStateException("ThreadPool está detenido");

        this.colaDetareas.offer(tarea);
    }

    public synchronized void stop(){
        this.estaDetenido = true;
        for(PoolThreadRunnable ejecutable : runnables){
            ejecutable.doStop();
        }
    }

    public synchronized void esperarHastaQueTodasLasTareasTerminen() {
        while(this.colaDetareas.size() > 0) {
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

Ahora veamos la clase PoolThreadRunnable que implementa la interfaz Runnable:

import java.util.concurrent.BlockingQueue;

public class PoolThreadRunnable implements Runnable {

    private Thread hilo = null;
    private BlockingQueue<Runnable> colaDetareas = null;
    private boolean estaDetenido = false;

    public PoolThreadRunnable(BlockingQueue<Object> cola){
        colaDetareas = cola;
    }

    public void run(){
        this.hilo = Thread.currentThread();
        while(!isStopped()){
            try{
                Runnable runnable = colaDetareas.take();
                runnable.run();
            } catch(Exception e){
                // Registrar o reportar la excepción,
                // pero mantener el hilo del pool vivo.
            }
        }
    }

    public synchronized void doStop(){
        estaDetenido = true;
        // Sacar al hilo del pool de la llamada a dequeue().
        this.hilo.interrupt();
    }

    public synchronized boolean isStopped(){
        return estaDetenido;
    }
}

Y finalmente, un ejemplo de cómo usar el ThreadPool:

public class App {
    public static void main(String[] args) throws Exception {
        int numTareas = 20;
        int numHilos = 4;
        ThreadPool threadPool = new ThreadPool(numHilos, numTareas);

        for(int i=0; i<numTareas; i++) {

            int numTarea = i;
            threadPool.execute( () -> {
                String mensaje = Thread.currentThread().getName() + ": Tarea " + numTarea;
                System.out.println(mensaje);
            });
        }

        threadPool.esperarHastaQueTodasLasTareasTerminen();
        threadPool.stop();

    }
}

La implementación del pool de hilos consta de dos partes. Una clase ThreadPool que es la interfaz pública del pool de hilos, y una clase PoolThreadRunnable que implementa los hilos que ejecutan las tareas.

Para ejecutar una tarea, se llama al método ThreadPool.execute(Runnable r) con una implementación de Runnable como parámetro. El Runnable se encola en la BlockingQueue internamente, esperando ser desencolado.

El Runnable será desencolado por un PoolThreadRunnable inactivo y ejecutado. Podemos ver esto en el método PoolThreadRunnable.run(). Después de la ejecución, el PoolThreadRunnable hace un ciclo e intenta desencolar otra tarea de nuevo, hasta que se detenga.

Para detener el ThreadPool, se llama al método ThreadPool.stop(). La llamada a stop se anota internamente en el atributo estaDetenido. Luego, cada hilo en el pool se detiene llamando a doStop() en cada hilo. Observemos cómo el método execute() lanzará una IllegalStateException si se llama después de que se haya llamado a stop().

Los hilos se detendrán después de terminar cualquier tarea que estén ejecutando actualmente. Observemos la llamada a this.interrupt() en PoolThreadRunnable.doStop(). Esto asegura que un hilo bloqueado en una llamada wait() dentro de la llamada colaDetareas.take() salga de la llamada de wait() y salga del método take() con una InterruptedException lanzada. Esta excepción se captura en el método PoolThreadRunnable.run(), se reporta, y luego se verifica la variable estaDetenido. Como estaDetenido es ahora verdadero, PoolThreadRunnable.run() saldrá y el hilo morirá.