resueltos - java threadpoolexecutor example

ThreadPoolExecutor con una cola ilimitada que no crea nuevos hilos

Question

Mi ThreadPoolExecutor no puede crear nuevos hilos. De hecho, escribí un LinkedBlockingQueue un tanto LinkedBlockingQueue que aceptará cualquier tarea (es decir, no tiene límites) pero llamé a un controlador adicional, que en mi aplicación muestra un rastro de advertencia de que el grupo está detrás, lo que me da información muy explícita a la que el TPE se está negando. cree nuevos hilos aunque la cola tenga miles de entradas. Mi constructor es el siguiente:

private final ExecutorService s3UploadPool = 
new ThreadPoolExecutor(1, 40, 1, TimeUnit.HOURS, unboundedLoggingQueue);

¿Por qué no está creando nuevos hilos?

Answer 1

Como lo menciona @djechlin, esto es parte del comportamiento definido (sorprendente para muchos) del ThreadPoolExecutor . Creo que he encontrado una solución algo elegante para este comportamiento que muestro en mi respuesta aquí:

¿Cómo hacer que ThreadPoolExecutor aumente el número de subprocesos al máximo antes de poner en cola?

Básicamente, usted extiende LinkedBlockingQueue para que siempre devuelva false para queue.offer(...) lo que agregará subprocesos adicionales a la agrupación, si es necesario. Si el grupo ya está en el número máximo de subprocesos y todos están ocupados, se llamará a RejectedExecutionHandler . Es el manejador el que luego put(...) en la cola.

Ver mi código allí.

Answer 2

Este gotcha está cubierto en esta entrada de blog :

Esta construcción de grupo de hilos simplemente no funcionará como se esperaba. Esto se debe a la lógica dentro del ThreadPoolExecutor donde se agregan nuevos subprocesos si no se puede ofrecer una tarea a la cola. En nuestro caso, utilizamos un LinkedBlockingQueue ilimitado, donde siempre podemos ofrecer una tarea a la cola. De hecho, significa que nunca creceremos por encima del tamaño de la agrupación principal y hasta el tamaño máximo de la agrupación.

Si también necesita desacoplar el mínimo de los tamaños máximos de la piscina, tendrá que hacer un poco de codificación extendida. No tengo conocimiento de una solución que exista en las bibliotecas de Java o en Apache Commons. La solución es crear un BlockingQueue acoplado que tenga conocimiento del TPE, y hará todo lo posible para rechazar una tarea si sabe que el TPE no tiene subprocesos disponibles y luego se vuelve a poner en cola manualmente. Se trata con más detalle en el post vinculado. En última instancia, su construcción se verá como:

public static ExecutorService newScalingThreadPool(int min, int max, long keepAliveTime) {
   ScalingQueue queue = new ScalingQueue();
   ThreadPoolExecutor executor =
      new ScalingThreadPoolExecutor(min, max, keepAliveTime, TimeUnit.MILLISECONDS, queue);
   executor.setRejectedExecutionHandler(new ForceQueuePolicy());
   queue.setThreadPoolExecutor(executor);
   return executor;
}

Sin embargo, más simplemente configure corePoolSize en maxPoolSize y no se preocupe por esta tontería.

Answer 3

Hay una solución a este problema. Considere la siguiente implementación:

int corePoolSize = 40;
int maximumPoolSize = 40;
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 
    60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
threadPoolExecutor.allowCoreThreadTimeOut(true);

Al establecer allowCoreThreadTimeOut() en true , los subprocesos de la agrupación pueden terminar después del tiempo de espera especificado (60 segundos en este ejemplo). Con esta solución, es el argumento del constructor corePoolSize que determina el tamaño máximo de grupo en la práctica, ya que el grupo de subprocesos crecerá hasta el corePoolSize y luego comenzará a agregar trabajos a la cola. Es probable que la agrupación nunca crezca más que eso, ya que la agrupación no generará nuevos subprocesos hasta que la cola esté llena (lo que, dado que LinkedBlockingQueue tiene una capacidad Integer.MAX_VALUE , nunca puede ocurrir). En consecuencia, no tiene mucho sentido establecer la configuración de maximumPoolSize en un valor mayor que corePoolSize .

Consideración: el grupo de subprocesos tiene 0 subprocesos inactivos después de que el tiempo de espera haya expirado, lo que significa que habrá algo de latencia antes de que se creen los subprocesos (normalmente, siempre corePoolSize subprocesos corePoolSize disponibles).

Se pueden encontrar más detalles en el ThreadPoolExecutor de ThreadPoolExecutor .