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

Время ожидания бездействия

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

Очереди

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

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

Существует три общие стратегии организации очередей:

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

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

Новые задачи, отправленные методом execute(Runnable), будут отклонены, если Executor остановлен, а также если Executor использует конечные ограничения как на число потоков, так и на ёмкость рабочей очереди и достиг этих ограничений. В обоих случаях метод 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() можно переопределить для выполнения специальных действий после полной остановки 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();
    }
  }
}