Класс 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 не удаётся создать поток по запросу и метод newThread возвращает null, исполнитель продолжит работу, но может оказаться неспособным выполнять какие-либо задачи.

Время ожидания перед завершением

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

Постановка в очередь

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

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

Существует три основные стратегии постановки в очередь:

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

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

Новые задачи, отправленные методом 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), которые вызываются до и после выполнения каждой задачи. Их можно использовать для изменения среды выполнения, например для повторной инициализации ThreadLocal, сбора статистики или добавления записей в журнал. Кроме того, метод terminated() можно переопределить для выполнения специальных действий после полной остановки исполнителя.

Если методы-перехватчики, методы обратного вызова или методы 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();
    }
  }
}