Класс 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 приоритет и статус не-daemon. Предоставляя другой ThreadFactory, вы можете изменить имя потока, группу потоков, приоритет, статус daemon и т. д. Если 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, обработчик выбрасывает исключение времени выполнения 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();
     }
   }
 }