Как заставить ThreadPoolExecutor увеличить потоки до максимума перед очередью?

установите размер ядра и максимальный размер в одно и то же значение и разрешите удаление основных потоков из пула с помощью allowCoreThreadTimeOut(true).

у меня уже есть два других ответа на этот вопрос, но я подозреваю, что это один из лучших.

он основан на технике в настоящее время принимаются ответ, а именно:

1. переопределить очередь offer() метод для (иногда) возврата false,
2. что вызывает ThreadPoolExecutor чтобы либо создать новый поток, либо отклонить задачу, и
3. установить RejectedExecutionHandler до на самом деле поставить задачу в очередь отказ.

проблема в том, когда offer() должна возвращать false. В настоящее время принятый ответ возвращает false, когда в очереди есть несколько задач, но, как я указал в своем комментарии, это вызывает нежелательные эффекты. Кроме того, если вы всегда возвращаете false, вы будете продолжать создавать новые потоки, даже если у вас есть потоки, ожидающие очереди.

решение заключается в использовании Java 7 LinkedTransferQueue и offer() вызов tryTransfer(). Когда существует ожидающий поток потребителя, задача будет просто передана этому потоку. В противном случае, offer() вернет false и ThreadPoolExecutor создаст новый поток.

    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();
            }
        }
    });

Примечание: теперь я предпочитаю и рекомендую мой другой ответ.

вот версия, которая кажется мне гораздо более простой: увеличьте corePoolSize (до предела maximumPoolSize) всякий раз, когда выполняется новая задача, а затем уменьшите corePoolSize (до предела указанного пользователем "размера основного пула") всякий раз, когда задача завершается.

другими словами, следите за количеством запущенных или поставленных в очередь задач и убедитесь, что corePoolSize равен равно числу задач, если оно находится между указанным пользователем "размером основного пула" и максимальным размером пула.

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);
    }
}

как написано, класс не поддерживает изменение указанного пользователем corePoolSize или maximumPoolSize после построения и не поддерживает управление рабочей очередью напрямую или через remove() или purge().

у нас есть подкласс ThreadPoolExecutor это требует дополнительного creationThreshold и переопределяет 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);
            }
        }
    }
}

может быть, это тоже помогает, Но ваш выглядит более вычурно, конечно...

рекомендуемый ответ решает только одну (1) проблему с пулом потоков JDK:

1. пулы потоков JDK смещены в сторону очереди. Поэтому вместо того, чтобы порождать новый поток, они будут ставить задачу в очередь. Только если очередь достигнет своего предела, пул потоков создаст новый поток.

2. выход на пенсию потока не происходит, когда нагрузка облегчается. Например, если у нас есть всплеск заданий, попадающих в пул, который заставляет пул перейти к max, а затем при небольшой нагрузке до 2 задач одновременно пул будет использовать все потоки для обслуживания легкой нагрузки, предотвращающей выход потока на пенсию. (только 2 потока будут необходимы...)

недовольный поведением выше, я пошел вперед и реализовал пул для преодоления недостатков выше.

для решения 2) Использование планирования Lifo решает проблему. Эта идея была представлена Беном Маурером на конференции ACM applicative 2015: системы @ Facebook scale

Так родилась новая реализация:

LifoThreadPoolExecutorSQP

до сих пор эта реализация улучшает производительность асинхронного выполнения для зел.

реализация spin способна уменьшить накладные расходы на переключение контекста, что обеспечивает превосходную производительность для определенных случаев использования.

надеюсь, что это помогает...

PS: JDK Fork Join Pool реализует ExecutorService и работает как" обычный " пул потоков, Реализация является производительной, она использует планирование потоков LIFO, однако нет контроля над внутренним размером очереди, таймаутом выхода на пенсию...

Примечание: теперь я предпочитаю и рекомендую мой другой ответ.

у меня есть еще одно предложение, следуя первоначальной идее изменения очереди, чтобы вернуть false. В этом случае все задачи могут войти в очередь, но всякий раз, когда задача ставится в очередь после execute(), мы следуем за ним с помощью задачи sentinel no-op, которую очередь отклоняет, вызывая появление нового потока, который будет выполнять no-op сразу же после чего-то из очереди.

потому что рабочие потоки могут будьте опрашивать LinkedBlockingQueue для новой задачи можно поставить задачу в очередь, даже если есть доступный поток. Чтобы избежать появления новых потоков даже при наличии доступных потоков, мы должны отслеживать, сколько потоков ожидает новых задач в очереди, и только порождать новый поток, когда в очереди больше задач, чем ожидающих потоков.

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();            
    }
});

лучшее решение, которое я могу придумать, это продлить.

ThreadPoolExecutor предлагает несколько методов крюк: beforeExecute и afterExecute. В вашем расширении вы можете использовать ограниченную очередь для подачи задач и вторую неограниченную очередь для обработки переполнения. Когда кто-то звонит submit, вы можете попытаться поместить запрос в ограниченную очередь. Если вы столкнулись с исключением, вы просто вставляете задачу в свою очередь переполнения. Затем вы можете использовать afterExecute крюк, чтобы увидеть, если есть все, что находится в очереди переполнения после завершения задачи. Таким образом, исполнитель сначала позаботится о том, что находится в ограниченной очереди, и автоматически вытащит из этой неограниченной очереди, как только позволит время.

кажется, что это больше работы, чем ваше решение, но по крайней мере это не связано с неожиданным поведением очередей. Я также полагаю, что есть лучший способ проверить состояние очереди и потоков, а не полагаться на исключения, которые довольно медленно выбрасываются.