Создание динамического (растущего/сокращающегося) пула потоков

Мне нужно реализовать пул потоков в Java (java.util.concurrent), число потоков которого при некотором минимальном значении равно нулю, растет до верхней границы (но не более), когда задания отправляются в него быстрее, чем они заканчивают выполнение и сжимаются до нижней границы, когда выполняются все задания, и больше не представлено заданий.

Как бы вы реализовали что-то подобное? Я предполагаю, что это будет довольно распространенный сценарий использования, но, по-видимому, методы java.util.concurrent.Executors factory могут создавать только пулы и пулы с фиксированным размером, которые неограниченно растут при подаче многих заданий. Класс ThreadPoolExecutor предоставляет параметры corePoolSize и maximumPoolSize, но его документация, по-видимому, подразумевает, что единственный способ иметь больше потоков corePoolSize в то же время - использовать ограниченную очередь заданий, в этом случае, если вы достигли потоков maximumPoolSize, вы получите отказ от работы, с которым вам приходится иметь дело? Я придумал это:

//pool creation
ExecutorService pool = new ThreadPoolExecutor(minSize, maxSize, 500, TimeUnit.MILLISECONDS,
    new ArrayBlockingQueue<Runnable>(minSize));
...

//submitting jobs
for (Runnable job : ...) {
    while (true) {
        try {
            pool.submit(job);
            System.out.println("Job " + job + ": submitted");
            break;
        } catch (RejectedExecutionException e) {
            // maxSize jobs executing concurrently atm.; re-submit new job after short wait
            System.out.println("Job " + job + ": rejected...");
            try {
                Thread.sleep(300);
            } catch (InterruptedException e1) {
            }
        }
    }
}

Я что-то пропускаю? Есть лучший способ сделать это? Кроме того, в зависимости от одного из требований может быть проблематично, что приведенный выше код не будет завершен до тех пор, пока, по крайней мере, (я думаю) не завершает работу (total number of jobs) - maxSize. Поэтому, если вы хотите отправить произвольное количество заданий в пул и продолжить немедленно, не дожидаясь окончания какого-либо из них, я не вижу, как вы могли бы это сделать, не имея выделенного потока работы, который управляет требуемую неограниченную очередь для хранения всех представленных заданий. AFAICS, если вы используете неограниченную очередь для самого ThreadPoolExecutor, его количество потоков никогда не будет расти за пределы corePoolSize.

Ответ 1

Один трюк, который может вам помочь, - назначить RejectedExecutionHandler, который использует тот же поток, чтобы отправить задание в очередь блокировки. Это блокирует текущий поток и устраняет необходимость в каком-то цикле.

Смотрите мой ответ здесь:

Как заставить команду ThreadPoolExecutor ждать, нужно ли работать с большим количеством данных?

Здесь обработчик отклонения копируется из этого ответа.

final BlockingQueue queue = new ArrayBlockingQueue<Runnable>(200);
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(nThreads, nThreads,
       0L, TimeUnit.MILLISECONDS, queue);
// by default (unfortunately) the ThreadPoolExecutor will call the rejected
// handler when you submit the 201st job, to have it block you do:
threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
   public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
      // this will block if the queue is full
      executor.getQueue().put(r);
   }
});

Затем вы сможете использовать количество потоков core/max, пока вы понимаете, что ограниченная очередь блокировки, которую вы используете, сначала заполняется до того, как потоки будут созданы над основными потоками. Итак, если у вас есть 10 основных потоков, и вы хотите, чтобы 11-я работа начала 11-й поток, вам понадобится очередь блокировки с размером 0 (возможно, SynchronousQueue). Я чувствую, что это реальное ограничение в классных классах ExecutorService в противном случае.

Ответ 2

Когда нарастание и сокращение происходит вместе с потоком, мне приходит в голову только одно имя: CachedThreadPool из пакета java.util.concurrent.

ExecutorService executor = Executors.newCachedThreadPool();

CachedThreadPool() может повторно использовать поток, а также создавать новые потоки при необходимости. И да, если поток простаивает в течение 60 секунд, CachedThreadPool убьет его.. Это довольно легкий - растет и сокращается в ваших словах!

Ответ 3

Установите maximumPoolSize в Integer.MAX_VALUE. Если у вас когда-либо было более 2 миллиардов потоков... ну, удачи с этим.

Во всяком случае, Javadoc ThreadPoolExecutor гласит:

Установив maximumPoolSize на существенно неограниченное значение, например Integer.MAX_VALUE, вы позволяете пулу вмещать произвольное количество одновременных задач. Как правило, размеры ядра и максимального пула устанавливаются только при построении, но их также можно динамически изменять с помощью setCorePoolSize (int) и setMaximumPoolSize (int).

При такой же неограниченной очереди задач, как a LinkedBlockingQueue, она должна иметь произвольно большую емкость.