¿Bloque ThreadPoolExecutor cuando su cola está llena?

Estoy tratando de ejecutar muchas tareas usando un ThreadPoolExecutor. A continuación se muestra un ejemplo hipotético:

def workQueue = new ArrayBlockingQueue<Runnable>(3, false)
def threadPoolExecutor = new ThreadPoolExecutor(3, 3, 1L, TimeUnit.HOURS, workQueue)
for(int i = 0; i < 100000; i++)
    threadPoolExecutor.execute(runnable)

El problema es que rápidamente obtengo un java.util.concurrent.RejectedExecutionException ya que el número de tareas excede el tamaño de la cola de trabajo. Sin embargo, el comportamiento deseado que busco es tener el hilo principal bloqueado hasta que haya espacio en la cola. Cuál es la mejor manera de lograr esto?

En algunas circunstancias muy limitadas, puede implementar un java.util.concurrent.RejectedExecutionHandler que hace lo que necesita.

RejectedExecutionHandler block = new RejectedExecutionHandler() {
  rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
     executor.getQueue().put( r );
  }
};

ThreadPoolExecutor pool = new ...
pool.setRejectedExecutionHandler(block);

Ahora. Esto es una muy mala idea por las siguientes razones

• Es propenso al interbloqueo porque todos los subprocesos en el grupo pueden morir antes de que se vea lo que puso en la cola. Mitigue esto estableciendo un tiempo de vida razonable.
• La tarea no está envuelta de la forma en que su Ejecutor puede esperar. Muchas implementaciones de ejecutores envuelven sus tareas en algún tipo de objeto de seguimiento antes de la ejecución. Mira la fuente de la tuya.
• La API desaconseja encarecidamente agregar a través de getQueue(), y puede prohibirse en algún momento.

Una estrategia casi siempre mejor es instalar ThreadPoolExecutor.CallerRunsPolicy, que acelerará su aplicación ejecutando la tarea en el subproceso que llama a execute().

Sin embargo, a veces una estrategia de bloqueo, con todos sus riesgos inherentes, es realmente lo que desea. Yo diría que en estas condiciones

• Solo tienes un hilo llamando a ejecutar ()
• Tienes que (o quieres) tener una longitud de cola muy pequeña
• Es absolutamente necesario limitar la cantidad de subprocesos que ejecutan este trabajo (generalmente por razones externas), y una estrategia de ejecución de llamadas rompería eso.
• Sus tareas son de un tamaño impredecible, por lo que las ejecuciones de llamadas podrían causar inanición si el grupo estuviera momentáneamente ocupado con 4 tareas cortas y su ejecución de llamada de un hilo se atascara con una grande.

Entonces, como digo. Rara vez se necesita y puede ser peligroso, pero ahí lo tienes.

Buena suerte.

Lo que debe hacer es envolver su ThreadPoolExecutor en Executor, lo que limita explícitamente la cantidad de operaciones ejecutadas simultáneamente en su interior:

 private static class BlockingExecutor implements Executor {

    final Semaphore semaphore;
    final Executor delegate;

    private BlockingExecutor(final int concurrentTasksLimit, final Executor delegate) {
        semaphore = new Semaphore(concurrentTasksLimit);
        this.delegate = delegate;
    }

    @Override
    public void execute(final Runnable command) {
        try {
            semaphore.acquire();
        } catch (InterruptedException e) {
            e.printStackTrace();
            return;
        }

        final Runnable wrapped = () -> {
            try {
                command.run();
            } finally {
                semaphore.release();
            }
        };

        delegate.execute(wrapped);

    }
}

Puede ajustar concurrentTasksLimit al threadPoolSize + queueSize de su ejecutor delegado y prácticamente resolverá su problema

Podrías usar un semaphore para evitar que los subprocesos entren en el grupo.

ExecutorService service = new ThreadPoolExecutor(
    3, 
    3, 
    1, 
    TimeUnit.HOURS, 
    new ArrayBlockingQueue<>(6, false)
);

Semaphore lock = new Semaphore(6); // equal to queue capacity

for (int i = 0; i < 100000; i++ ) {
    try {
        lock.acquire();
        service.submit(() -> {
            try {
              task.run();
            } finally {
              lock.release();
            }
        });
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

Algunas trampas:

• Solo use este patrón con un grupo de subprocesos fijos. Es poco probable que la cola se llene con frecuencia, por lo que no se crearán nuevos hilos. Consulte los documentos de Java en ThreadPoolExecutor para obtener más detalles: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ThreadPoolExecutor.html Hay una forma de evitar esto, pero está fuera del alcance de esta respuesta.

• El tamaño de la cola debe ser mayor que el número de subprocesos principales. Si tuviéramos que hacer la cola de tamaño 3, lo que terminaría sucediendo es:

  • T0: los tres subprocesos están funcionando, la cola está vacía, no hay permisos disponibles.
  • T1: el subproceso 1 finaliza, libera un permiso.
  • T2: el subproceso 1 sondea la cola en busca de nuevos trabajos, no encuentra ninguno y murga.
  • T3: el subproceso principal envía el trabajo al grupo, el subproceso 1 comienza a funcionar.

  El ejemplo anterior se traduce para enhebrar el hilo principal. bloqueando hilo 1. Puede parecer un período pequeño, pero ahora multiplique la frecuencia por días y meses. De repente, los cortos períodos de tiempo se suman a una gran cantidad de tiempo perdido.

Esto es lo que terminé haciendo:

int NUM_THREADS = 6;
Semaphore lock = new Semaphore(NUM_THREADS);
ExecutorService pool = Executors.newCachedThreadPool();

for (int i = 0; i < 100000; i++) {
    try {
        lock.acquire();
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
    pool.execute(() -> {
        try {
            // Task logic
        } finally {
            lock.release();
        }
    });
}

Una opción bastante sencilla es envolver su BlockingQueue con una implementación que llama put(..) cuando offer(..) se invoca:

public class BlockOnOfferAdapter<T> implements BlockingQueue<T> {

(..)

  public boolean offer(E o) {
        try {
            delegate.put(o);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
        return true;
  }

(.. implement all other methods simply by delegating ..)

}

Esto funciona porque por defecto put(..) espera hasta que haya capacidad en la cola cuando está llena, ver:

    /**
     * Inserts the specified element into this queue, waiting if necessary
     * for space to become available.
     *
     * @param e the element to add
     * @throws InterruptedException if interrupted while waiting
     * @throws ClassCastException if the class of the specified element
     *         prevents it from being added to this queue
     * @throws NullPointerException if the specified element is null
     * @throws IllegalArgumentException if some property of the specified
     *         element prevents it from being added to this queue
     */
    void put(E e) throws InterruptedException;

Sin captura de RejectedExecutionException o bloqueo complicado necesario.

Aquí está mi fragmento de código en este caso:

public void executeBlocking( Runnable command ) {
    if ( threadPool == null ) {
        logger.error( "Thread pool '{}' not initialized.", threadPoolName );
        return;
    }
    ThreadPool threadPoolMonitor = this;
    boolean accepted = false;
    do {
        try {
            threadPool.execute( new Runnable() {
                @Override
                public void run() {
                    try {
                        command.run();
                    }
                    // to make sure that the monitor is freed on exit
                    finally {
                        // Notify all the threads waiting for the resource, if any.
                        synchronized ( threadPoolMonitor ) {
                            threadPoolMonitor.notifyAll();
                        }
                    }
                }
            } );
            accepted = true;
        }
        catch ( RejectedExecutionException e ) {
            // Thread pool is full
            try {
                // Block until one of the threads finishes its job and exits.
                synchronized ( threadPoolMonitor ) {
                    threadPoolMonitor.wait();
                }
            }
            catch ( InterruptedException ignored ) {
                // return immediately
                break;
            }
        }
    } while ( !accepted );
}

threadPool es una instancia local de java.util.concurrent.ExecutorService que ya se ha inicializado.

Resolví este problema usando un RejectedExecutionHandler personalizado, que simplemente bloquea el hilo de llamada por un tiempo y luego intenta enviar la tarea nuevamente:

public class BlockWhenQueueFull implements RejectedExecutionHandler {

    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {

        // The pool is full. Wait, then try again.
        try {
            long waitMs = 250;
            Thread.sleep(waitMs);
        } catch (InterruptedException interruptedException) {}

        executor.execute(r);
    }
}

Esta clase solo se puede usar en el ejecutor del grupo de subprocesos como un RejectedExecutionHandler como cualquier otro. En este ejemplo:

executorPool = new def threadPoolExecutor = new ThreadPoolExecutor(3, 3, 1L, TimeUnit.HOURS, workQueue, new BlockWhenQueueFull())

El único inconveniente que veo es que el subproceso de llamada puede bloquearse un poco más de lo estrictamente necesario (hasta 250 ms). Para muchas tareas de ejecución corta, tal vez reduzca el tiempo de espera a 10 ms aproximadamente. Además, dado que este ejecutor se llama de manera recursiva, las esperas muy largas para que un subproceso esté disponible (horas) pueden provocar un desbordamiento de la pila.

Sin embargo, personalmente me gusta este método. Es compacto, fácil de entender y funciona bien. ¿Me estoy perdiendo algo importante?