¿Cómo conseguir que el ThreadPoolExecutor aumente los hilos al máximo antes de hacer cola?

Establezca el tamaño del núcleo y el tamaño máximo en el mismo valor, y permita que los hilos del núcleo se eliminen del grupo con allowCoreThreadTimeOut(true).

Ya tengo otras dos respuestas a esta pregunta, pero sospecho que esta es la mejor.

Se basa en la técnica de la respuesta actualmente aceptada , a saber:

1. Invalida el método offer() de la cola para (a veces) devolver false,
2. que hace que el ThreadPoolExecutor genere un nuevo hilo o rechace la tarea, y
3. establece el RejectedExecutionHandlera en realidad encola la tarea en caso de rechazo.

El problema es cuando offer() debería devuelve false. La respuesta actualmente aceptada devuelve false cuando la cola tiene un par de tareas en ella, pero como he señalado en mi comentario allí, esto causa efectos indeseables. Alternativamente, si siempre devuelves false, seguirás generando nuevos hilos incluso cuando tengas hilos esperando en la cola.

La solución es usar Java 7 LinkedTransferQueue y tienen offer() llamar tryTransfer(). Cuando hay un subproceso de consumidor en espera, la tarea simplemente se pasará a ese subproceso. De lo contrario, offer() devolverá false y el ThreadPoolExecutor generará un nuevo hilo.

    BlockingQueue<Runnable> queue = new LinkedTransferQueue<Runnable>() {
        @Override
        public boolean offer(Runnable e) {
            return tryTransfer(e);
        }
    };
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 50, 60, TimeUnit.SECONDS, queue);
    threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            try {
                executor.getQueue().put(r);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    });

Nota: ahora prefiero y recomiendo mi otra respuesta.

Aquí hay una versión que me parece mucho más sencilla: Aumentar el tamaño de la base de datos (hasta el límite de maximumPoolSize) cada vez que se ejecuta una nueva tarea, luego disminuir el tamaño de la base de datos (hasta el límite del "tamaño del núcleo" especificado por el usuario) cada vez que se completa una tarea.

Para decirlo de otra manera, lleve un registro del número de tareas en ejecución o en cola, y asegúrese de que el corePoolSize sea igual al número de tareas siempre y cuando esté entre el usuario especificado "core pool size" y el maximumPoolSize.

public class GrowBeforeQueueThreadPoolExecutor extends ThreadPoolExecutor {
    private int userSpecifiedCorePoolSize;
    private int taskCount;

    public GrowBeforeQueueThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        userSpecifiedCorePoolSize = corePoolSize;
    }

    @Override
    public void execute(Runnable runnable) {
        synchronized (this) {
            taskCount++;
            setCorePoolSizeToTaskCountWithinBounds();
        }
        super.execute(runnable);
    }

    @Override
    protected void afterExecute(Runnable runnable, Throwable throwable) {
        super.afterExecute(runnable, throwable);
        synchronized (this) {
            taskCount--;
            setCorePoolSizeToTaskCountWithinBounds();
        }
    }

    private void setCorePoolSizeToTaskCountWithinBounds() {
        int threads = taskCount;
        if (threads < userSpecifiedCorePoolSize) threads = userSpecifiedCorePoolSize;
        if (threads > getMaximumPoolSize()) threads = getMaximumPoolSize();
        setCorePoolSize(threads);
    }
}

Tal como está escrito, la clase no admite cambiar el corePoolSize o maximumPoolSize especificado por el usuario después de la construcción, y no admite la manipulación de la cola de trabajo directamente o a través de remove() o purge().

Tenemos una subclase de ThreadPoolExecutor que toma un creationThreshold adicional y anula execute.

public void execute(Runnable command) {
    super.execute(command);
    final int poolSize = getPoolSize();
    if (poolSize < getMaximumPoolSize()) {
        if (getQueue().size() > creationThreshold) {
            synchronized (this) {
                setCorePoolSize(poolSize + 1);
                setCorePoolSize(poolSize);
            }
        }
    }
}

Tal vez eso también ayude, pero el tuyo se ve más artístico, por supuesto {

La respuesta recomendada resuelve solo uno (1) del problema con el grupo de subprocesos JDK:

1. Los grupos de subprocesos JDK están sesgados hacia las colas. Así que en lugar de generar un nuevo hilo, pondrán en cola la tarea. Solo si la cola alcanza su límite, el grupo de subprocesos generará un nuevo subproceso.

2. El retiro del hilo no ocurre cuando la carga se aligera. Por ejemplo, si tenemos una ráfaga de trabajos que golpean la piscina que hace que la piscina vaya a max, seguido de una carga ligera de max 2 tareas a la vez, el grupo utilizará todos los subprocesos para mantener la carga ligera evitando el retiro de subprocesos. (solo se necesitarían 2 hilos...)

Insatisfecho con el comportamiento anterior, seguí adelante e implementé un pool para superar las deficiencias anteriores.

Para resolver 2) El uso de la programación Lifo resuelve el problema. Esta idea fue presentada por Ben Maurer en la conferencia ACM applicative 2015: Systems @ Facebook scale

Así que una nueva implementación fue nacido:

LifoThreadPoolExecutorSQP

Hasta ahora esta implementación mejora el rendimiento de ejecución asincrónica para ZEL.

La implementación es spin capaz de reducir la sobrecarga del cambio de contexto, produciendo un rendimiento superior para ciertos casos de uso.

Espero que ayude...

PD: JDK Fork Join Pool implementa ExecutorService y funciona como un grupo de subprocesos "normal", La implementación es eficiente, utiliza la programación de subprocesos LIFO, sin embargo, no hay control sobre el tamaño de la cola interna, tiempo de espera de retiro...

Nota: ahora prefiero y recomiendo mi otra respuesta.

Tengo otra propuesta, siguiendo la idea original de cambiar la cola para devolver false. En esta, todas las tareas pueden entrar en la cola, pero cada vez que una tarea se encuela después de execute(), la seguimos con una tarea no-op centinela que la cola rechaza, causando que aparezca un nuevo hilo, que ejecutará la no-op inmediatamente seguido por algo de la cola.

Porque los hilos de trabajo pueden estar sondeando el LinkedBlockingQueue para una tarea nueva, es posible que una tarea se ponga en cola incluso cuando hay un hilo disponible. Para evitar generar nuevos subprocesos incluso cuando hay subprocesos disponibles, necesitamos realizar un seguimiento de cuántos subprocesos están esperando nuevas tareas en la cola, y solo generar un nuevo subproceso cuando hay más tareas en la cola que subprocesos en espera.

final Runnable SENTINEL_NO_OP = new Runnable() { public void run() { } };

final AtomicInteger waitingThreads = new AtomicInteger(0);

BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {
    @Override
    public boolean offer(Runnable e) {
        // offer returning false will cause the executor to spawn a new thread
        if (e == SENTINEL_NO_OP) return size() <= waitingThreads.get();
        else return super.offer(e);
    }

    @Override
    public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
        try {
            waitingThreads.incrementAndGet();
            return super.poll(timeout, unit);
        } finally {
            waitingThreads.decrementAndGet();
        }
    }

    @Override
    public Runnable take() throws InterruptedException {
        try {
            waitingThreads.incrementAndGet();
            return super.take();
        } finally {
            waitingThreads.decrementAndGet();
        }
    }
};

ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 50, 60, TimeUnit.SECONDS, queue) {
    @Override
    public void execute(Runnable command) {
        super.execute(command);
        if (getQueue().size() > waitingThreads.get()) super.execute(SENTINEL_NO_OP);
    }
};
threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (r == SENTINEL_NO_OP) return;
        else throw new RejectedExecutionException();            
    }
});

La mejor solución que se me ocurre es extender.

ThreadPoolExecutor ofrece algunos métodos de gancho: beforeExecute y afterExecute. En su extensión puede mantener una cola limitada para alimentar las tareas y una segunda cola ilimitada para manejar el desbordamiento. Cuando alguien llama a submit, puede intentar colocar la solicitud en la cola acotada. Si te encuentras con una excepción, simplemente pegas la tarea en tu cola de desbordamiento. A continuación, podría utilizar el gancho afterExecute para ver si hay algo en el cola de desbordamiento después de terminar una tarea. De esta manera, el ejecutor se encargará de las cosas en su cola limitada primero, y automáticamente extraerá de esta cola ilimitada según lo permita el tiempo.

Parece más trabajo que su solución, pero al menos no implica dar a las colas comportamientos inesperados. También imagino que hay una mejor manera de verificar el estado de la cola y los hilos en lugar de depender de excepciones, que son bastante lentas de lanzar.