Класс ThreadPoolExecutor

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

Для полезного использования в широком диапазоне контекстов этот класс предоставляет множество настраиваемых параметров и расширяемых точек. Однако программистам настоятельно рекомендуется использовать более удобные фабричные методы класса Executors - Executors.newCachedThreadPool() (неограниченный пул потоков с автоматическим освобождением потоков), Executors.newFixedThreadPool(int) (пул потоков фиксированного размера) и Executors.newSingleThreadExecutor() (одиночный фоновый поток), которые предварительно настраивают параметры для наиболее распространенных сценариев использования. В противном случае следуйте руководству при настройке и настройке этого класса:

Основные и максимальные размеры пула

A ThreadPoolExecutor будет автоматически настраивать размер пула (см. getPoolSize()) в соответствии с границами, заданными corePoolSize (см. getCorePoolSize()) и maximumPoolSize (см. getMaximumPoolSize()). При отправке новой задачи в методе execute(Runnable), если запущено меньше corePoolSize потоков, создается новый поток для обработки запроса, даже если другие рабочие потоки простаивают. В противном случае, если запущено меньше maximumPoolSize потоков, новый поток будет создан для обработки запроса только в случае, если очередь заполнена. Установив corePoolSize и maximumPoolSize одинаковыми, вы создадите пул потоков фиксированного размера. Установив maximumPoolSize на существенно неограниченное значение, например, Integer.MAX_VALUE, вы разрешите пулу вместить произвольное количество одновременных задач. Чаще всего, основные и максимальные размеры пула устанавливаются только при создании, но их также можно изменить динамически с помощью setCorePoolSize(int) и setMaximumPoolSize(int).

Создание по требованию

По умолчанию, даже основные потоки изначально создаются и запускаются только при поступлении новых задач, но это можно динамически изменить, используя метод prestartCoreThread() или prestartAllCoreThreads(). Вероятно, вам захочется запустить потоки, если вы создаете пул с непустой очередью.

Создание новых потоков

Новые потоки создаются с использованием ThreadFactory. Если не указано иное, используется Executors.defaultThreadFactory(), который создает потоки, которые все находятся в одной ThreadGroup и имеют одинаковый NORM_PRIORITY приоритет и не являются демонами. Предоставив другой ThreadFactory, вы можете изменить имя потока, группу потоков, приоритет, статус демона и т. д. Если ThreadFactory не удается создать поток, вернув null из newThread, исполняемый модуль продолжит работу, но может не выполнить какие-либо задачи. Потоки должны обладать правом "modifyThread" RuntimePermission. Если рабочие потоки или другие потоки, использующие пул, не обладают этим правом, производительность может быть снижена: изменения конфигурации могут не вступить в силу своевременно, и пул завершения может оставаться в состоянии, в котором завершение возможно, но не выполнено.

Время ожидания

Если в пуле в данный момент больше, чем corePoolSize потоков, избыточные потоки будут завершены, если они простаивали более keepAliveTime (см. getKeepAliveTime(TimeUnit)). Это позволяет уменьшить потребление ресурсов, когда пул не используется активно. Если позже пул станет более активным, будут созданы новые потоки. Этот параметр также можно изменить динамически, используя метод setKeepAliveTime(long, TimeUnit). Использование значения Long.MAX_VALUE TimeUnit.NANOSECONDS фактически отключает возможность завершения простаивающих потоков до завершения работы. По умолчанию политика ожидания применяется только тогда, когда потоков больше, чем corePoolSize, но метод allowCoreThreadTimeOut(boolean) может быть использован для применения этой политики тайм-аута к основным потокам также, при условии, что значение keepAliveTime не равно нулю.

Очередь

Любая BlockingQueue может использоваться для передачи и хранения отправленных задач. Использование этой очереди взаимодействует с размером пула:

• Если запущено меньше corePoolSize потоков, Executor всегда предпочитает добавление нового потока, а не помещение в очередь.
• Если запущено corePoolSize или более потоков, Executor всегда предпочитает помещение запроса в очередь, а не добавление нового потока.
• Если запрос не может быть помещен в очередь, создается новый поток, если это не превысит maximumPoolSize, в противном случае задача будет отклонена.

Существует три общие стратегии для очереди:

1. Прямые переходы. Хорошим выбором для очереди задач является SynchronousQueue, который передает задачи потокам без удержания их в противном случае. Здесь попытка поместить задачу в очередь завершится неудачей, если нет потоков, готовых ее выполнить, поэтому будет создан новый поток. Эта политика позволяет избежать блокировок при обработке наборов запросов, которые могут иметь внутренние зависимости. Прямые переходы обычно требуют неограниченных maximumPoolSizes, чтобы избежать отклонения новых отправленных задач. Это, в свою очередь, допускает возможность неограниченного роста потоков, когда команды в среднем поступают быстрее, чем могут быть обработаны.
2. Неограниченные очереди. Использование неограниченной очереди (например, LinkedBlockingQueue без предварительно определенной емкости) приведет к ожиданию новых задач в очереди, когда все потоки corePoolSize заняты. Таким образом, никогда не будет создано более corePoolSize потоков. (А значение maximumPoolSize, следовательно, не имеет никакого эффекта.) Это может быть уместно, когда каждая задача полностью независима от других, поэтому задачи не могут влиять на выполнение друг друга; например, в веб-сервере. Хотя этот стиль очереди может быть полезен для сглаживания кратковременных всплесков запросов, он допускает возможность неограниченного роста очереди задач, когда команды в среднем поступают быстрее, чем могут быть обработаны.
3. Ограниченные очереди. Ограниченная очередь (например, ArrayBlockingQueue) помогает предотвратить истощение ресурсов при использовании конечных maximumPoolSizes, но может быть сложнее настроить и контролировать. Размеры очереди и максимальные размеры пула могут быть взаимно компенсированы друг другом: использование больших очередей и малых пулов минимизирует использование процессора, ресурсы ОС и накладные расходы переключения контекста, но может привести к искусственно низкой производительности. Если задачи часто блокируются (например, если они связаны с вводом-выводом), система может запланировать время для большего количества потоков, чем вы иначе разрешаете. Использование небольших очередей обычно требует больших размеров пула, что удерживает процессоры более загруженными, но может столкнуться с неприемлемыми накладными расходами планирования, что также снижает пропускную способность.

Отклоненные задачи

Новые задачи, отправленные в методе execute(Runnable), будут отклонены, когда Executor был остановлен, а также когда Executor использует конечные границы как для максимального числа потоков, так и для емкости очереди задач, и насыщен. В любом случае, метод execute вызывает метод RejectedExecutionHandler.rejectedExecution(Runnable, ThreadPoolExecutor) своего RejectedExecutionHandler. Предоставляются четыре предопределенные политики обработчика:

1. В стандартной ThreadPoolExecutor.AbortPolicy, обработчик выбрасывает исключение runtime RejectedExecutionException при отклонении.
2. В ThreadPoolExecutor.CallerRunsPolicy, поток, вызывающий execute сам выполняет задачу. Это обеспечивает простую механику обратной связи, которая замедлит темп отправки новых задач.
3. В ThreadPoolExecutor.DiscardPolicy, задача, которую нельзя выполнить, просто отбрасывается. Эта политика предназначена только для тех редких случаев, когда завершение задачи никогда не используется.
4. В ThreadPoolExecutor.DiscardOldestPolicy, если Executor не остановлен, задача в начале очереди задач отбрасывается, а затем выполняется попытка выполнения (которая может снова завершиться неудачей, вызывая повторение этого процесса). Эта политика редко приемлема. Почти во всех случаях вы также должны отменить задачу, чтобы вызвать исключение в любом компоненте, ожидающем ее завершения, и/или записать ошибку, как показано в документации ThreadPoolExecutor.DiscardOldestPolicy.

Возможно определить и использовать другие типы RejectedExecutionHandler классов. Для этого требуется осторожность, особенно когда политики предназначены для работы только в определенных условиях емкости или политики очереди.

Методы обратного вызова

Этот класс предоставляет protected переопределяемые beforeExecute(Thread, Runnable) и afterExecute(Runnable, Throwable) методы, которые вызываются до и после выполнения каждой задачи. Они могут использоваться для управления средой выполнения; например, для повторной инициализации ThreadLocals, сбора статистики или добавления записей журнала. Кроме того, метод terminated() может быть переопределен для выполнения любой специальной обработки, которая должна быть выполнена после полного завершения Executor.

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

Управление очередью

Метод getQueue() позволяет получить доступ к очереди задач для целей мониторинга и отладки. Использование этого метода для других целей настоятельно не рекомендуется. Два предоставляемых метода, remove(Runnable) и purge(), доступны для помощи в освобождении памяти при отмене большого количества задач в очереди.

Освобождение ресурсов

Пул, на который больше нет ссылок в программе И в котором нет оставшихся потоков, может быть освобожден (утилизирован сборщиком мусора), не закрываясь явно. Вы можете настроить пул, чтобы позволить всем неиспользуемым потокам в конечном итоге умереть, установив соответствующие значения времени ожидания, используя нижнюю границу нулевых основных потоков и/или установив allowCoreThreadTimeOut(boolean).

Пример расширения. Большинство расширений этого класса переопределяют один или несколько защищенных методов обратного вызова. Например, вот подкласс, который добавляет простую функцию паузы/возобновления:

 
 class PausableThreadPoolExecutor extends ThreadPoolExecutor {
   private boolean isPaused;
   private ReentrantLock pauseLock = new ReentrantLock();
   private Condition unpaused = pauseLock.newCondition();

   public PausableThreadPoolExecutor(...) { super(...); }

   protected void beforeExecute(Thread t, Runnable r) {
     super.beforeExecute(t, r);
     pauseLock.lock();
     try {
       while (isPaused) unpaused.await();
     } catch (InterruptedException ie) {
       t.interrupt();
     } finally {
       pauseLock.unlock();
     }
   }

   public void pause() {
     pauseLock.lock();
     try {
       isPaused = true;
     } finally {
       pauseLock.unlock();
     }
   }

   public void resume() {
     pauseLock.lock();
     try {
       isPaused = false;
       unpaused.signalAll();
     } finally {
       pauseLock.unlock();
     }
   }
 }