¿Imposible hacer un grupo de subprocesos en caché con un límite de tamaño?

Parece imposible hacer un grupo de subprocesos en caché con un límite en el número de subprocesos que puede crear.

Así es como se implementa Executors.newCachedThreadPool estático en la biblioteca estándar de Java:

 public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

Entonces, usando esa plantilla para crear un grupo de subprocesos en caché de tamaño fijo:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new SynchronusQueue<Runable>());

Ahora, si usa esto y envía 3 tareas, todo estará bien. El envío de cualquier otra tarea dará como resultado excepciones de ejecución rechazadas.

Intentando esto:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runable>());

El resultado será que todos los hilos se ejecuten secuencialmente. Es decir, el grupo de subprocesos nunca hará más de un subproceso para manejar sus tareas.

Este es un error en el método de ejecución de ThreadPoolExecutor? O tal vez esto es intencional? ¿O hay alguna otra manera?

Editar: quiero algo exactamente como el grupo de subprocesos en caché (crea subprocesos bajo demanda y luego los mata después de un tiempo de espera) pero con un límite en el número de subprocesos que puede crear y la capacidad de continuar haciendo cola tareas adicionales una vez que ha golpear su límite de hilo. Según la respuesta de sjlee, esto es imposible. Mirando el método execute () de ThreadPoolExecutor, es realmente imposible. Necesitaría subclasificar ThreadPoolExecutor y anular execute () de manera similar a SwingWorker, pero lo que SwingWorker hace en execute () es un hack completo.

ThreadPoolExecutor tiene los siguientes comportamientos clave, y estos problemas pueden explicarse por estos comportamientos.

Cuando se envían tareas,

1. Si el grupo de subprocesos no ha alcanzado el tamaño del núcleo, crea nuevos subprocesos.
2. Si se ha alcanzado el tamaño del núcleo y no hay hilos inactivos, pone en cola las tareas.
3. Si se ha alcanzado el tamaño del núcleo, no hay subprocesos inactivos y la cola se llena, crea nuevos subprocesos (hasta que alcanza el tamaño máximo).
4. Si se ha alcanzado el tamaño máximo, no hay hilos inactivos y la cola se llena, la política de rechazo se activa.

En el primer ejemplo, tenga en cuenta que SynchronousQueue tiene esencialmente un tamaño de 0. Por lo tanto, en el momento en que alcanza el tamaño máximo (3), la política de rechazo entra en acción (# 4).

En el segundo ejemplo, la cola de elección es un LinkedBlockingQueue que tiene un tamaño ilimitado. Por lo tanto, te quedas atascado con el comportamiento # 2.

Realmente no puede jugar mucho con el tipo en caché o el tipo fijo, ya que su comportamiento está casi completamente determinado.

Si desea tener un grupo de subprocesos dinámico y acotado, debe usar un tamaño de núcleo positivo y un tamaño máximo combinados con una cola de un tamaño finito. Por ejemplo,

new ThreadPoolExecutor(10, // core size
    50, // max size
    10*60, // idle timeout
    TimeUnit.SECONDS,
    new ArrayBlockingQueue<Runnable>(20)); // queue with a size

Anexo : esta es una respuesta bastante antigua, y parece que JDK cambió su comportamiento cuando se trata del tamaño del núcleo de 0. Desde JDK 1.6, si el tamaño del núcleo es 0 y el grupo no tiene ningún subproceso, el ThreadPoolExecutor agregará un hilo para ejecutar esa tarea. Por lo tanto, el tamaño del núcleo de 0 es una excepción a la regla anterior. Gracias Steve por llamarme la atención.

A menos que me haya perdido algo, la solución a la pregunta original es simple. El siguiente código implementa el comportamiento deseado como se describe en el póster original. Generará hasta 5 hilos para trabajar en una cola ilimitada y los hilos inactivos terminarán después de 60 segundos.

tp = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<Runnable>());
tp.allowCoreThreadTimeOut(true);

Tuve el mismo problema. Como ninguna otra respuesta reúne todos los problemas, agrego la mía:

Ahora está claramente escrito en documentos : si usa una cola que no bloquea ( LinkedBlockingQueue) la configuración de subprocesos máximos no tiene efecto, solo se usan subprocesos principales.

entonces:

public class MyExecutor extends ThreadPoolExecutor {

    public MyExecutor() {
        super(4, 4, 5,TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
        allowCoreThreadTimeOut(true);
    }

    public void setThreads(int n){
        setMaximumPoolSize(Math.max(1, n));
        setCorePoolSize(Math.max(1, n));
    }

}

Este ejecutor tiene:

1. No hay concepto de subprocesos máximos ya que estamos usando una cola ilimitada. Esto es bueno porque dicha cola puede hacer que el ejecutor cree una cantidad masiva de subprocesos extra no centrales si sigue su política habitual.

2. Una cola de tamaño máximo Integer.MAX_VALUE. Submit()lanzará RejectedExecutionExceptionsi el número de tareas pendientes excede Integer.MAX_VALUE. No estoy seguro de que primero se nos acabe la memoria o esto sucederá.

3. Tiene 4 hilos principales posibles. Los subprocesos de núcleo inactivo salen automáticamente si están inactivos durante 5 segundos. Entonces, sí, los subprocesos estrictamente a pedido. El número se puede variar utilizando el setThreads()método.

4. Se asegura de que el número mínimo de subprocesos principales nunca sea inferior a uno, o submit()rechazará todas las tareas. Dado que los subprocesos centrales deben ser> = subprocesos máximos, el método también setThreads()establece subprocesos máximos, aunque la configuración de subprocesos máximos es inútil para una cola ilimitada.

En su primer ejemplo, las tareas posteriores se rechazan porque AbortPolicyes el valor predeterminado RejectedExecutionHandler. ThreadPoolExecutor contiene las siguientes políticas, que puede cambiar mediante el setRejectedExecutionHandlermétodo:

CallerRunsPolicy
AbortPolicy
DiscardPolicy
DiscardOldestPolicy

Parece que quiere un grupo de subprocesos en caché con CallerRunsPolicy.

Ninguna de las respuestas aquí solucionó mi problema, que tenía que ver con la creación de una cantidad limitada de conexiones HTTP utilizando el cliente HTTP de Apache (versión 3.x). Como me llevó algunas horas encontrar una buena configuración, compartiré:

private ExecutorService executor = new ThreadPoolExecutor(5, 10, 60L,
  TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
  Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());

Esto crea un ThreadPoolExecutorque comienza con cinco y contiene un máximo de diez hilos que se ejecutan simultáneamente CallerRunsPolicypara ejecutar.

Según el Javadoc para ThreadPoolExecutor:

Si hay más subprocesos corePoolSize pero inferiores al máximo de PoolsSize en ejecución, se creará un nuevo subproceso solo si la cola está llena . Al configurar corePoolSize y maximumPoolSize de la misma manera, crea un grupo de subprocesos de tamaño fijo.

(El énfasis es mío).

La respuesta de Jitter es lo que quieres, aunque la mía responde a tu otra pregunta. :)

Hay una opción más. En lugar de usar el nuevo SynchronousQueue, también puede usar cualquier otra cola, pero debe asegurarse de que su tamaño sea 1, de modo que obligue al servicio del ejecutor a crear un nuevo hilo.

No parece que ninguna de las respuestas realmente responda la pregunta, de hecho, no puedo ver una forma de hacerlo, incluso si subclasifica de PooledExecutorService ya que muchos de los métodos / propiedades son privados, por ejemplo, haciendo que addIfUnderMaximumPoolSize esté protegido, podría Haz lo siguiente:

class MyThreadPoolService extends ThreadPoolService {
    public void execute(Runnable run) {
        if (poolSize() == 0) {
            if (addIfUnderMaximumPoolSize(run) != null)
                return;
        }
        super.execute(run);
    }
}

Lo más cerca que estuve fue esto, pero incluso eso no es una muy buena solución

new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()) {
    public void execute(Runnable command) {
        if (getPoolSize() == 0 && getActiveCount() < getMaximumPoolSize()) {        
            super.setCorePoolSize(super.getCorePoolSize() + 1);
        }
        super.execute(command);
    }

    protected void afterExecute(Runnable r, Throwable t) {
         // nothing in the queue
         if (getQueue().isEmpty() && getPoolSize() > min) {
             setCorePoolSize(getCorePoolSize() - 1);
         }
    };
 };

ps no probado lo anterior

Aquí hay otra solución. Creo que esta solución se comporta como lo desea (aunque no está orgullosa de esta solución):

final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {
    public boolean offer(Runnable o) {
        if (size() > 1)
            return false;
        return super.offer(o);
    };

    public boolean add(Runnable o) {
        if (super.offer(o))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }
};

RejectedExecutionHandler handler = new RejectedExecutionHandler() {         
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        queue.add(r);
    }
};

dbThreadExecutor =
        new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS, queue, handler);

Esto es lo que quieres (al menos supongo que sí). Para una explicación, verifique la respuesta de Jonathan Feinberg

Executors.newFixedThreadPool(int n)

Crea un grupo de subprocesos que reutiliza un número fijo de subprocesos que operan en una cola compartida no acotada. En cualquier momento, a lo sumo, los hilos nThreads serán tareas de procesamiento activas. Si se envían tareas adicionales cuando todos los hilos están activos, esperarán en la cola hasta que haya un hilo disponible. Si algún subproceso termina debido a un error durante la ejecución antes del cierre, uno nuevo tomará su lugar si es necesario para ejecutar tareas posteriores. Los subprocesos en el grupo existirán hasta que se cierre explícitamente.

1. Puede usar ThreadPoolExecutorcomo lo sugiere @sjlee

   Puede controlar el tamaño del grupo de forma dinámica. Echa un vistazo a esta pregunta para más detalles:

   Grupo de subprocesos dinámicos

   O

2. Puede usar newWorkStealingPool API, que se ha introducido con java 8.

   public static ExecutorService newWorkStealingPool()

   Crea un grupo de subprocesos que roba trabajo utilizando todos los procesadores disponibles como su nivel de paralelismo objetivo.

De forma predeterminada, el nivel de paralelismo se establece en el número de núcleos de CPU en su servidor. Si tiene un servidor de CPU de 4 núcleos, el tamaño del grupo de subprocesos sería 4. Esta API devuelve el ForkJoinPooltipo de ExecutorService y permite robar trabajo de subprocesos inactivos robando tareas de subprocesos ocupados en ForkJoinPool.

El problema se resumió de la siguiente manera:

Quiero algo exactamente como el grupo de subprocesos en caché (crea subprocesos bajo demanda y luego los mata después de un tiempo de espera) pero con un límite en el número de subprocesos que puede crear y la capacidad de continuar haciendo cola en tareas adicionales una vez que ha alcanzado su límite límite de hilo.

Antes de señalar la solución, explicaré por qué las siguientes soluciones no funcionan:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());

Esto no pondrá en cola ninguna tarea cuando se alcance el límite de 3 porque SynchronousQueue, por definición, no puede contener ningún elemento.

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

Esto no creará más de un subproceso porque ThreadPoolExecutor solo crea subprocesos que exceden el corePoolSize si la cola está llena. Pero LinkedBlockingQueue nunca está lleno.

ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 3, 60, TimeUnit.SECONDS,
    new LinkedBlockingQueue<Runnable>());
executor.allowCoreThreadTimeOut(true);

Esto no reutilizará subprocesos hasta que se haya alcanzado corePoolSize porque ThreadPoolExecutor aumenta el número de subprocesos hasta que se alcance corePoolSize incluso si los subprocesos existentes están inactivos. Si puede vivir con esta desventaja, entonces esta es la solución más fácil para el problema. También es la solución descrita en "Concurrencia de Java en la práctica" (nota al pie de página en p172).

La única solución completa al problema descrito parece ser la que implica anular el offermétodo de la cola y escribir un mensajeRejectedExecutionHandler como se explica en las respuestas a esta pregunta: ¿Cómo hacer que ThreadPoolExecutor aumente los hilos al máximo antes de hacer cola?

Esto funciona para Java8 + (y otros, por ahora ..)

     Executor executor = new ThreadPoolExecutor(3, 3, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>()){{allowCoreThreadTimeOut(true);}};

donde 3 es el límite de hilos y 5 es tiempo de espera para hilos inactivos.

Si desea verificar si funciona usted mismo , aquí está el código para hacer el trabajo:

public static void main(String[] args) throws InterruptedException {
    final int DESIRED_NUMBER_OF_THREADS=3; // limit of number of Threads for the task at a time
    final int DESIRED_THREAD_IDLE_DEATH_TIMEOUT=5; //any idle Thread ends if it remains idle for X seconds

    System.out.println( java.lang.Thread.activeCount() + " threads");
    Executor executor = new ThreadPoolExecutor(DESIRED_NUMBER_OF_THREADS, DESIRED_NUMBER_OF_THREADS, DESIRED_THREAD_IDLE_DEATH_TIMEOUT, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>()) {{allowCoreThreadTimeOut(true);}};

    System.out.println(java.lang.Thread.activeCount() + " threads");

    for (int i = 0; i < 5; i++) {
        final int fi = i;
        executor.execute(() -> waitsout("starting hard thread computation " + fi, "hard thread computation done " + fi,2000));
    }
    System.out.println("If this is UP, it works");

    while (true) {
        System.out.println(
                java.lang.Thread.activeCount() + " threads");
        Thread.sleep(700);
    }

}

static void waitsout(String pre, String post, int timeout) {
    try {
        System.out.println(pre);
        Thread.sleep(timeout);
        System.out.println(post);
    } catch (Exception e) {
    }
}

salida del código anterior para mí es

1 threads
1 threads
If this is UP, it works
starting hard thread computation 0
4 threads
starting hard thread computation 2
starting hard thread computation 1
4 threads
4 threads
hard thread computation done 2
hard thread computation done 0
hard thread computation done 1
starting hard thread computation 3
starting hard thread computation 4
4 threads
4 threads
4 threads
hard thread computation done 3
hard thread computation done 4
4 threads
4 threads
4 threads
4 threads
3 threads
3 threads
3 threads
1 threads
1 threads