Размер основного пула против максимального размера пула в ThreadPoolExecutor

В чем именно разница между размером основного пула и максимальным размером пула, когда мы говорим о них ThreadPoolExecutor?
Можно ли это объяснить на примере?

Из этого сообщения в блоге :

Возьмите этот пример. Размер начального пула потоков - 1, размер основного пула - 5, максимальный размер пула - 10, а очередь - 100.

По мере поступления запросов будет создано до 5 потоков, а затем задачи будут добавляться в очередь до тех пор, пока она не достигнет 100. Когда очередь заполнится, новые потоки будут созданы до maxPoolSize. Как только все потоки будут использованы и очередь заполнена, задачи будут отклонены. По мере уменьшения очереди уменьшается и количество активных потоков.

ЕСЛИ запущенные потоки> corePoolSize & <maxPoolSize , то создайте новый поток , если общая очередь задач заполнена и прибывает новый.

Форма документа: (Если количество запущенных потоков превышает corePoolSize, но меньше maximumPoolSize , новый поток будет создан только в том случае, если очередь заполнена.)

Теперь возьмем простой пример,

ThreadPoolExecutor executorPool = new ThreadPoolExecutor(5, 10, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(50));

Здесь 5 - это corePoolSize - означает, что Jvm создаст новый поток для новой задачи для первых 5 задач. и другие задачи будут добавляться в очередь, пока она не заполнится (50 задач).

10 - это maxPoolSize - JVM может создавать максимум 10 потоков. Означает, что если уже запущено 5 задач / потоков и очередь заполнена 50 ожидающими задачами, и если в очередь поступает еще один новый запрос / задача, тогда JVM создаст новый поток до 10 (общее количество потоков = предыдущие 5 + новые 5) ;

new ArrayBlockingQueue (50) = - это общий размер очереди - в ней можно поставить в очередь 50 задач.

после того, как все 10 потоков будут запущены, и если появится новая задача, эта новая задача будет отклонена.

Правила создания потоков внутри SUN:

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

2. Если количество потоков равно (или больше) corePoolSize, поместите задачу в очередь.

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

4. Если очередь заполнена, а количество потоков больше или равно maxPoolSize, отклоните задачу.

Надеюсь, это HelpFul .. и, пожалуйста, поправьте меня, если я ошибаюсь ...

Из документа :

Когда новая задача отправляется в методе execute (java.lang.Runnable) и выполняется меньше потоков corePoolSize, создается новый поток для обработки запроса, даже если другие рабочие потоки простаивают. Если количество запущенных потоков больше corePoolSize, но меньше maximumPoolSize, новый поток будет создан только в том случае, если очередь заполнена.

Более того:

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

Если вы решите создать класс ThreadPoolExecutorвручную вместо использования Executorsфабричного класса, вам нужно будет создать и настроить его с помощью одного из его конструкторов. Самый обширный конструктор этого класса:

public ThreadPoolExecutor(
    int corePoolSize,
    int maxPoolSize,
    long keepAlive,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    RejectedExecutionHandler handler
);

Как видите, вы можете настроить:

• Размер основного пула (размер пула потоков, который пытается придерживаться).
• Максимальный размер пула.
• Время поддержания активности, то есть время, по истечении которого неактивный поток может быть отключен.
• Очередь работ для хранения задач, ожидающих выполнения.
• Политика, применяемая при отклонении отправки задачи.

Ограничение количества задач в очереди

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

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

Решение этой проблемы: Предоставление очереди блокировки исполнителю для хранения ожидающих задач. В случае заполнения очереди отправленное задание будет «отклонено». RejectedExecutionHandlerВызывается , когда представление задачи отвергается, и именно поэтому глагол отвергается был цитируемых в предыдущем пункте. Вы можете реализовать свою собственную политику отказа или использовать одну из встроенных политик, предоставляемых платформой.

В политиках отклонения по умолчанию исполнитель создает файл RejectedExecutionException. Однако другие встроенные политики позволяют:

• Молча откажитесь от работы.
• Отбросьте самую старую работу и попробуйте повторно отправить последнюю.
• Выполнить отклоненную задачу в потоке вызывающего.

Источник

Правила размера пула ThreadPoolExecutor

Правила для размера ThreadPoolExecutor'sпула обычно неправильно понимаются, потому что он работает не так, как вы думаете, или так, как вы этого хотите.

Возьмите этот пример. Размер начального пула потоков - 1, размер основного пула - 5, максимальный размер пула - 10, а очередь - 100.

Способ Sun: по мере поступления запросов в потоках будет создано до 5, затем задачи будут добавляться в очередь до тех пор, пока она не достигнет 100. Когда очередь заполнится, новые потоки будут созданы до maxPoolSize. Как только все потоки будут использованы и очередь заполнена, задачи будут отклонены. По мере уменьшения очереди уменьшается и количество активных потоков.

Ожидаемый пользователем способ: по мере поступления запросов в потоках будет создано до 10, затем задачи будут добавлены в очередь, пока не достигнет 100, после чего они будут отклонены. Количество потоков будет переименовано в максимальное, пока очередь не станет пустой. Когда очередь пуста, потоки отмирают, пока не corePoolSizeостанутся.

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

Вот простые правила Sun для создания потоков:

1. Если количество потоков меньше corePoolSize, создайте новый поток для запуска новой задачи.
2. Если количество потоков равно (или больше) corePoolSize, поместите задачу в очередь.
3. Если очередь заполнена, а количество потоков меньше maxPoolSize, создайте новый поток для выполнения задач.
4. Если очередь заполнена, а количество потоков больше или равно maxPoolSize, отклоните задачу. Суть в том, что новые потоки создаются только при заполнении очереди, поэтому, если вы используете неограниченную очередь, количество потоков не будет превышать corePoolSize.

Более полное объяснение можно получить из уст лошадей: ThreadPoolExecutorдокументация по API.

Есть действительно хорошее сообщение на форуме, в котором рассказывается, как ThreadPoolExecutorработает эта работа с примерами кода: http://forums.sun.com/thread.jspa?threadID=5401400&tstart=0

Подробнее: http://forums.sun.com/thread.jspa?threadID=5224557&tstart=450

Вы можете найти определение терминов corepoolsize и maxpoolsize в javadoc. http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/ThreadPoolExecutor.html

По ссылке выше есть ответ на ваш вопрос. Однако просто чтобы было понятно. Приложение будет продолжать создавать потоки, пока не достигнет corePoolSize. Я думаю, что идея здесь в том, что этого количества потоков должно хватить для обработки потока задач. Если новая задача приходит после создания потоков corePoolSize, задачи будут поставлены в очередь. Как только очередь будет заполнена, исполнитель начнет создавать новые потоки. Это своего рода балансировка. По сути, это означает, что поток задач превышает вычислительную мощность. Итак, Executor снова начнет создавать новые потоки, пока не достигнет максимального количества потоков. Опять же, новые потоки будут созданы тогда и только тогда, когда очередь заполнена.

Хорошее объяснение в этом блоге:

Иллюстрация

public class ThreadPoolExecutorExample {

    public static void main (String[] args) {
        createAndRunPoolForQueue(new ArrayBlockingQueue<Runnable>(3), "Bounded");
        createAndRunPoolForQueue(new LinkedBlockingDeque<>(), "Unbounded");
        createAndRunPoolForQueue(new SynchronousQueue<Runnable>(), "Direct hand-off");
    }

    private static void createAndRunPoolForQueue (BlockingQueue<Runnable> queue,
                                                                      String msg) {
        System.out.println("---- " + msg + " queue instance = " +
                                                  queue.getClass()+ " -------------");

        ThreadPoolExecutor e = new ThreadPoolExecutor(2, 5, Long.MAX_VALUE,
                                 TimeUnit.NANOSECONDS, queue);

        for (int i = 0; i < 10; i++) {
            try {
                e.execute(new Task());
            } catch (RejectedExecutionException ex) {
                System.out.println("Task rejected = " + (i + 1));
            }
            printStatus(i + 1, e);
        }

        e.shutdownNow();

        System.out.println("--------------------\n");
    }

    private static void printStatus (int taskSubmitted, ThreadPoolExecutor e) {
        StringBuilder s = new StringBuilder();
        s.append("poolSize = ")
         .append(e.getPoolSize())
         .append(", corePoolSize = ")
         .append(e.getCorePoolSize())
         .append(", queueSize = ")
         .append(e.getQueue()
                  .size())
         .append(", queueRemainingCapacity = ")
         .append(e.getQueue()
                  .remainingCapacity())
         .append(", maximumPoolSize = ")
         .append(e.getMaximumPoolSize())
         .append(", totalTasksSubmitted = ")
         .append(taskSubmitted);

        System.out.println(s.toString());
    }

    private static class Task implements Runnable {

        @Override
        public void run () {
            while (true) {
                try {
                    Thread.sleep(1000000);
                } catch (InterruptedException e) {
                    break;
                }
            }
        }
    }
}

Вывод :

---- Bounded queue instance = class java.util.concurrent.ArrayBlockingQueue -------------
poolSize = 1, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 3, maximumPoolSize = 5, totalTasksSubmitted = 1
poolSize = 2, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 3, maximumPoolSize = 5, totalTasksSubmitted = 2
poolSize = 2, corePoolSize = 2, queueSize = 1, queueRemainingCapacity = 2, maximumPoolSize = 5, totalTasksSubmitted = 3
poolSize = 2, corePoolSize = 2, queueSize = 2, queueCapacity = 1, maximumPoolSize = 5, totalTasksSubmitted = 4
poolSize = 2, corePoolSize = 2, queueSize = 3, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 5
poolSize = 3, corePoolSize = 2, queueSize = 3, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 6
poolSize = 4, corePoolSize = 2, queueSize = 3, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 7
poolSize = 5, corePoolSize = 2, queueSize = 3, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 8
Task rejected = 9
poolSize = 5, corePoolSize = 2, queueSize = 3, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 9
Task rejected = 10
poolSize = 5, corePoolSize = 2, queueSize = 3, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 10
--------------------

---- Unbounded queue instance = class java.util.concurrent.LinkedBlockingDeque -------------
poolSize = 1, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 2147483647, maximumPoolSize = 5, totalTasksSubmitted = 1
poolSize = 2, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 2147483647, maximumPoolSize = 5, totalTasksSubmitted = 2
poolSize = 2, corePoolSize = 2, queueSize = 1, queueRemainingCapacity = 2147483646, maximumPoolSize = 5, totalTasksSubmitted = 3
poolSize = 2, corePoolSize = 2, queueSize = 2, queueRemainingCapacity = 2147483645, maximumPoolSize = 5, totalTasksSubmitted = 4
poolSize = 2, corePoolSize = 2, queueSize = 3, queueRemainingCapacity = 2147483644, maximumPoolSize = 5, totalTasksSubmitted = 5
poolSize = 2, corePoolSize = 2, queueSize = 4, queueRemainingCapacity = 2147483643, maximumPoolSize = 5, totalTasksSubmitted = 6
poolSize = 2, corePoolSize = 2, queueSize = 5, queueRemainingCapacity = 2147483642, maximumPoolSize = 5, totalTasksSubmitted = 7
poolSize = 2, corePoolSize = 2, queueSize = 6, queueRemainingCapacity = 2147483641, maximumPoolSize = 5, totalTasksSubmitted = 8
poolSize = 2, corePoolSize = 2, queueSize = 7, queueRemainingCapacity = 2147483640, maximumPoolSize = 5, totalTasksSubmitted = 9
poolSize = 2, corePoolSize = 2, queueSize = 8, queueRemainingCapacity = 2147483639, maximumPoolSize = 5, totalTasksSubmitted = 10
--------------------

---- Direct hand-off queue instance = class java.util.concurrent.SynchronousQueue -------------
poolSize = 1, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 1
poolSize = 2, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 2
poolSize = 3, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 3
poolSize = 4, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 4
poolSize = 5, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 5
Task rejected = 6
poolSize = 5, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 6
Task rejected = 7
poolSize = 5, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 7
Task rejected = 8
poolSize = 5, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 8
Task rejected = 9
poolSize = 5, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 9
Task rejected = 10
poolSize = 5, corePoolSize = 2, queueSize = 0, queueRemainingCapacity = 0, maximumPoolSize = 5, totalTasksSubmitted = 10
--------------------


Process finished with exit code 0

Из книги Основы согласования Java :

CorePoolSize : ThreadPoolExecutor имеет атрибут corePoolSize, который определяет, сколько потоков он будет запускать до тех пор, пока новые потоки не будут запущены только при заполнении очереди

MaximumPoolSize : этот атрибут определяет максимальное количество запущенных потоков. Вы можете установить это значение в Integer. MAX_VALUE, чтобы не было верхней границы

java.util.concurrent.ThreadPoolExecutor

  public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

Понимание внутреннего поведения ThreadPoolExecutorпри отправке новой задачи помогло мне понять, чем corePoolSizeи чем maximumPoolSizeотличаются.

Позволять:

• Nбыть количество потоков в пуле, getPoolSize(). Активные потоки + свободные потоки.
• T быть количеством задач, переданных исполнителю / пулу.
• Cбыть основной размер пула, getCorePoolSize(). Сколько потоков может быть создано не более чем в пуле для входящих задач, прежде чем новые задачи попадут в очередь .
• Mбыть максимальный размер пула, getMaximumPoolSize(). Максимальное количество потоков, которое может выделить пул.

Поведение ThreadPoolExecutorв Java при отправке новой задачи:

• Ибо N <= Cнезанятым потокам не назначается новая входящая задача, вместо этого создается новый поток.
• Ибо N > Cи если есть незанятые потоки, то там назначается новая задача.
• Для N > Cа если нет НИКАКИХ праздные темы, новые задачи ставятся в очередь. НЕТ НОВОЙ НИТИ ЗДЕСЬ НЕ СОЗДАНО.
• Когда очередь заполнена , мы создаем новые потоки до M. Если Mдостигается, мы отклоняем задачи. Здесь важно не создавать новые потоки, пока очередь не заполнится!

Источники:

• docs.oracle.com
• отличная статья по теме.

Примеры

Пример с corePoolSize = 0и maximumPoolSize = 10с емкостью очереди 50.

В результате в пуле будет один активный поток, пока в очереди не будет 50 элементов.

executor.execute(task #1):

before task #1 submitted to executor: java.util.concurrent.ThreadPoolExecutor@c52dafe[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]

after task #1 submitted to executor: java.util.concurrent.ThreadPoolExecutor@c52dafe[Running, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 0]

[task #1 immediately queued and kicked in b/c the very first thread is created when `workerCountOf(recheck) == 0`]

execute(task #2):

before task #2 submitted to executor: java.util.concurrent.ThreadPoolExecutor@c52dafe[Running, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]

after task #2 submitted to executor: java.util.concurrent.ThreadPoolExecutor@c52dafe[Running, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 0]

[task #2 not starting before #1 is done]

... executed a few tasks...

execute(task #19)

before task #19 submitted to executor: java.util.concurrent.ThreadPoolExecutor@735afe38[Running, pool size = 1, active threads = 1, queued tasks = 17, completed tasks = 0]

after task #19 submitted to executor: java.util.concurrent.ThreadPoolExecutor@735afe38[Running, pool size = 1, active threads = 1, queued tasks = 18, completed tasks = 0]

...

execute(task #51)

before task submitted to executor: java.util.concurrent.ThreadPoolExecutor@735afe38[Running, pool size = 1, active threads = 1, queued tasks = 50, completed tasks = 0]

after task submitted to executor: java.util.concurrent.ThreadPoolExecutor@735afe38[Running, pool size = 2, active threads = 2, queued tasks = 50, completed tasks = 0]

Queue is full.
A new thread was created as the queue was full.

Пример с corePoolSize = 10и maximumPoolSize = 10с емкостью очереди 50.

В результате в пуле будет 10 активных потоков. Когда в очереди будет 50 элементов, задачи будут отклонены.

execute(task #1)

before task #1 submitted to executor: java.util.concurrent.ThreadPoolExecutor@32d9e072[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]

after task #1 submitted to executor: java.util.concurrent.ThreadPoolExecutor@32d9e072[Running, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]

execute(task #2)

before task #2 submitted to executor: java.util.concurrent.ThreadPoolExecutor@32d9e072[Running, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]

after task #2 submitted to executor: java.util.concurrent.ThreadPoolExecutor@32d9e072[Running, pool size = 2, active threads = 2, queued tasks = 0, completed tasks = 0]

execute(task #3)

before task #3 submitted to executor: java.util.concurrent.ThreadPoolExecutor@32d9e072[Running, pool size = 2, active threads = 2, queued tasks = 0, completed tasks = 0]

after task #3 submitted to executor: java.util.concurrent.ThreadPoolExecutor@32d9e072[Running, pool size = 3, active threads = 3, queued tasks = 0, completed tasks = 0]

... executed a few tasks...

execute(task #11)

before task #11 submitted to executor: java.util.concurrent.ThreadPoolExecutor@32d9e072[Running, pool size = 10, active threads = 10, queued tasks = 0, completed tasks = 0]

after task #11 submitted to executor: java.util.concurrent.ThreadPoolExecutor@32d9e072[Running, pool size = 10, active threads = 10, queued tasks = 1, completed tasks = 0]

... executed a few tasks...

execute(task #51)
before task #51 submitted to executor: java.util.concurrent.ThreadPoolExecutor@32d9e072[Running, pool size = 10, active threads = 10, queued tasks = 50, completed tasks = 0]

Task was rejected as we have reached `maximumPoolSize`.