Класс ThreadPoolExecutor

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

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

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

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 потоков, исполнитель всегда предпочитает добавление нового потока, а не помещение в очередь.
• Если запущено corePoolSize или больше потоков, исполнитель всегда предпочитает помещение запроса в очередь, а не добавление нового потока.
• Если запрос не может быть помещен в очередь, создается новый поток, если это не превысит maximumPoolSize, в противном случае задача будет отклонена.

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

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

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

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

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

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

Методы-обработчики

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

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

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

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

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

Пул, на который больше нет ссылок в программе И у которого нет оставшихся потоков, может быть освобожден (сбор мусора) без явного закрытия. Вы можете настроить пул таким образом, чтобы все неиспользуемые потоки в конечном итоге погибли, установив соответствующие значения keep-alive, используя нижнюю границу нулевых основных потоков и/или установив 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();
     }
   }
 }