Bloque ThreadPoolExecutor cuando la cola está llena?

Question

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 consigo un java.util.concurrent.RejectedExecutionException ya que el número de tareas supera el tamaño de la cola de trabajo. Sin embargo, el comportamiento deseado que estoy buscando es tener el bloque de hilos principal hasta que haya espacio en la cola. Cuál es la mejor manera de lograr esto?

Answer 1

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. Esta es una muy mala idea por las siguientes razones

• Es propenso a un punto muerto, porque todos los hilos de la piscina pueden morir antes de lo que usted pone en la cola es visible. Mitigue esto estableciendo un tiempo razonable de mantener vivo.
• La tarea no está envuelta de la manera 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 los tuyos.
• Agregar mediante getQueue() es desaconsejado por la API, y puede estar prohibido en algún momento.

Una estrategia casi siempre mejor es instalar ThreadPoolExecutor.CallerRunsPolicy que estrangulará tu aplicación ejecutando la tarea en el hilo 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

• Sólo tiene un hilo de llamar a execute()
• Tienes que (o quiere) tienen una muy pequeña longitud de la cola
• Es absolutamente necesario limitar el número de los hilos que ejecutan este trabajo (generalmente por razones externas), y una estrategia de ejecución de la llamada rompería eso.
• Sus tareas son de un tamaño impredecible, por lo que las llamadas pueden causar inanición si el grupo estuvo momentáneamente ocupado con 4 tareas cortas y su ejecución de llamada de un hilo quedó atascada con una gran.

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

Buena suerte.

Answer 2

Aquí es 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 se ha inicializado ya.

Answer 3

He resuelto este problema utilizando una costumbre RejectedExecutionHandler, que simplemente bloquea el subproceso de llamada por un tiempo y luego intenta enviar de nuevo la tarea:

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 sólo se puede utilizar en el hilo de la piscina ejecutor como 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 la llamada puede ser que consiga bloqueado un poco más largo de lo estrictamente necesario (hasta 250 ms). Para muchas tareas de ejecución breve, quizás disminuya el tiempo de espera a 10 ms o menos. Además, dado que este ejecutor se está llamando efectivamente de forma recursiva, esperar mucho tiempo para que un hilo esté disponible (horas) puede 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?