Это 8-й день моего участия в Gengwen Challenge, смотрите подробности мероприятия:Обновить вызов
1. Концепция пула потоков:
Пул потоков — это пул для управления потоками.Сначала создаются некоторые потоки, а их коллекция называется пулом потоков. Использование пула потоков может повысить производительность. Пул потоков создает большое количество простаивающих потоков при запуске системы. Программа передает задачу в пул потоков, и пул потоков запускает поток для выполнения задачи. После завершения выполнения , поток It не умрет, а вернется в пул потоков, чтобы снова стать бездействующим, ожидая выполнения следующей задачи.
Преимущества пулов потоков Уменьшите накладные расходы, вызванные созданием и уничтожением потоков, путем повторного использования существующих потоков, Улучшить скорость отклика системы.При поступлении задачи поток напрямую берется из пула потоков.Повторно используя существующий поток, его можно выполнить немедленно, не дожидаясь; Улучшить управляемость потоками. Потому что если потоки создаются без ограничений, это будет не только потреблять системные ресурсы, но и снижать стабильность системы. (OOM генерируется из-за чрезмерного использования памяти и вызывает чрезмерное переключение ЦП)
2. Создание пула потоков Java и основные параметры
Независимо от того, какой тип пула потоков создается (FixedThreadPool, CachedThreadPool...), будет вызываться конструктор ThreadPoolExecutor. newFixedThreadPool (пул потоков с фиксированным количеством потоков) newCachedThreadPool (пул кэшируемых потоков) newSingleThreadExecutor (однопоточный пул потоков) newScheduledThreadPool (пул потоков с временным и периодическим выполнением)
public ThreadPoolExecutor(
int corePoolSize, // 线程池长期维持的线程数，即使线程处于Idle状态，也不会回收。
int maximumPoolSize, // 线程数的上限
long keepAliveTime, TimeUnit unit, // 超过corePoolSize的线程的idle时长，
// 超过这个时间，多余的线程会被回收。
BlockingQueue<Runnable> workQueue, // 任务的排队队列
ThreadFactory threadFactory, // 新线程的产生方式
RejectedExecutionHandler handler) // 拒绝策略
corePoolSize: количество потоков, которые должны храниться в пуле, даже если они простаивают, если не задано {@code allowCoreThreadTimeOut}, если не задано значение allowCoreThreadTimeOut.)
maxPoolSize: максимальное количество потоков, разрешенных в пуле. (Максимальное количество потоков: в пуле потоков может быть создано максимальное количество потоков MaximumPoolSize.)
keepAliveTime: когда количество потоков больше, чем количество ядер, это максимальное время, в течение которого избыточные простаивающие потоки будут ждать новых задач перед завершением. (Время жизни: если потоки, превышающие количество основных потоков, не получили новых задач после времени keepAliveTime, они будут переработаны.)
unit: единица времени для аргумента {@code keepAliveTime} (единица времени для keepAliveTime.)
workQueue: очередь, используемая для хранения задач до их выполнения. Эта очередь будет содержать только задачи {@code Runnable}, отправленные методом {@code execute}. (Очередь для хранения задач для выполнения: когда количество отправленных задач превышает количество основных потоков, отправленные задачи сохраняются здесь. Она используется только для хранения задач Runnable, отправленных методом execute. Поэтому не переводите ее в очередь на работу. А теперь, ладно? Не копай себе яму.)
threadFactory: фабрика, используемая, когда исполнитель создает новый поток. (Технология потоков: используется для создания фабрики потоков. Например, здесь вы можете настроить имя потока. При анализе стека виртуальной машины вы можете узнать, откуда берется поток, посмотрев на имя, и вас не запутают. )
handler : обработчик, который следует использовать, когда выполнение заблокировано из-за достижения границ потока и емкости очереди. (Политика отклонения: когда очередь заполнена задачами и работают потоки с максимальным числом потоков, пул потоков задач, которые продолжают отправляться, не может быть обработан. Какую политику отказа следует реализовать.)
3. Процесс выполнения задачи пула потоков:
Когда задача отправлена и количество основных потоков, оставшихся в пуле потоков, меньше, чем количество потоков corePoolSize, пул потоков создаст основной поток для обработки отправленной задачи.
Если количество основных потоков в пуле потоков заполнено, то есть число потоков равно corePoolSize, вновь отправленная задача будет помещена в очередь задач workQueue для постановки в очередь на выполнение.
Когда количество выживших потоков в пуле потоков равно corePoolSize, а очередь задач workQueue заполнена, определить, достигает ли количество потоков максимальное значениеPoolSize, то есть заполнено ли максимальное количество потоков, и если нет, создать неосновной поток для выполнения отправленной задачи.
Если текущее количество потоков достигает максимального размера пула и появляются новые задачи, политика отклонения будет использоваться напрямую.
4. Политика отклонения пула потоков
Существует несколько типов политик отклонения пула потоков: AbortPolicy: напрямую вызывает RejectedExecutionException, политику по умолчанию.
DiscardPolicy: ничего не делать, просто отменить задачу.
DiscardOldestPolicy: Отменить выполнение следующей задачи (отбросить самую старую задачу в очереди выполнения, пытаясь освободить место для текущей задачи).
CallerRunsPolicy: задача выполняется в основном потоке (задача выполняется отправителем).
5. Очередь работ
Несколько типичных рабочих очередей ArrayBlockingQueue: Ограниченная очередь блокировки, реализованная с использованием массива, с первым в первой функции.
LinkedBlockingQueue: LinkedBlockingQueue (можно установить очередь емкости) — это очередь блокировки, основанная на структуре связанного списка. Задачи сортируются по принципу FIFO. Емкость может быть задана дополнительно. Если она не задана, это будет неограниченная очередь блокировки с максимальным длина Integer.MAX_VALUE.Пропускная способность обычно выше, чем у ArrayBlockingQuene, пул потоков newFixedThreadPool использует эту очередь;
PriorityBlockingQueue: неограниченная очередь блокировки с приоритетом, реализованным с использованием кучи сбалансированного двоичного дерева.
DelayQueue: DelayQueue (очередь с задержкой) — это очередь для отложенного выполнения задачи на определенный период времени. Сортировать по указанному времени выполнения от меньшего к большему, иначе сортировать по порядку постановки в очередь; Пул потоков newScheduledThreadPool использует эту очередь;
SynchronousQueue: блокирующая очередь, в которой не хранятся элементы. Каждая операция вставки должна ждать, пока другой поток не вызовет операцию удаления, иначе операция вставки всегда будет заблокирована. Пропускная способность обычно выше, чем у LinkedBlockingQuene, пул потоков newCachedThreadPool использует эту очередь;
LinkedTransferQueue: неограниченная очередь блокировки, состоящая из структуры связанного списка. По сравнению с другими блокирующими очередями, LinkedTransferQueue имеет больше методов tryTransfer и передачи.
LinkedBlockingDeque: Двунаправленная очередь блокировки, состоящая из структуры связанного списка. Это двунаправленная очередь блокировки, состоящая из структуры связанного списка.
6. Общие пулы потоков
Однопотоковый исполнитель: общедоступный статический ExecutorService newSingleThreadExecutor() { вернуть новый FinalizableDelegatedExecutorService (новый ThreadPoolExecutor(1, 1, 0L, Единица времени.МИЛЛИСЕКУНДЫ, новая LinkedBlockingQueue())); }
Создайте один поток. Он подходит для сценариев приложений, в которых задачи должны выполняться последовательно и в любой момент времени не должно быть активных нескольких потоков.
Для corePoolSize и maxPoolSize SingleThreadExecutor установлено значение 1, и одновременно может существовать только один поток. Используйте неограниченную очередь LinkedBlockingQueue (принцип «первым пришел — первым обслужен», поэтому задачи гарантированно выполняются одна за другой по порядку) в качестве рабочей очереди пула потоков.
Когда в пуле потоков нет потоков, для выполнения задачи создается новый поток.
После того, как в текущем пуле потоков появится поток, добавьте новую задачу в LinkedBlockingQueue.
После того, как поток выполнит первую задачу, он будет повторно получать задачи из LinkedBlockingQueue для выполнения в бесконечном цикле. KeepAliveTime равно 0
Сценарий использования: Подходит для сценария последовательного выполнения задач, одна задача выполняется по одной задаче за раз;
Фиксированный пул потоков:
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
Функции: Количество основных потоков совпадает с максимальным количеством потоков. keepAliveTime равен 0L, что означает, что избыточные потоки немедленно завершаются. Очередь блокировки — это неограниченная очередь LinkedBlockingQueue
FixedThreadPool — это пул потоков с фиксированным числом потоков. Когда потоки простаивают, они не будут перезапущены, пока пул потоков не будет закрыт. Когда все потоки активны, новые задачи будут ждать, пока поток не освободится.
Примечание. Использование пула потоков с неограниченной очередью приведет к увеличению памяти, поскольку newFixedThreadPool использует неограниченную блокирующую очередь LinkedBlockingQueue.Если поток получает задачу, время выполнения задачи будет относительно большим, что приведет к тому, что очередь задач накапливаются все больше и больше, что приводит к машинной памяти. Использование продолжает расти, что в конечном итоге приводит к OOM;
Сценарий использования: Подходит для обработки задач, интенсивно использующих ЦП, обеспечивая, чтобы при длительном использовании ЦП рабочими потоками выделялось как можно меньше потоков, то есть подходит для выполнения долгосрочных задач.
Кэшедтредпул:
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
Функции:
Количество основных потоков равно 0 Все потоки являются неосновными, максимальное количество потоков — Integer.MAX_VALUE. Очередь блокировки — SynchronousQueue (может быть не более одного элемента, и есть новая задача, которая блокирует ожидание) Время простоя неосновных потоков составляет 60 секунд.
Вариант использования: выполнение большого количества задач с коротким жизненным циклом. Поскольку максимальный размер пула не ограничен, скорость отправки задач > скорость обработки задач потока в пуле потоков будет продолжать создавать новые потоки; каждый раз, когда задача отправляется, поток будет обрабатываться немедленно, поэтому CachedThreadPool подходит для обработки большое количество задач с меньшими затратами времени Task.
Запланированный пул потоков:
Порог общего количества потоков — Integer.MAX_VALUE, рабочая очередь использует DelayedWorkQueue,
Неосновные потоки имеют время жизни 0, поэтому пул потоков содержит только фиксированное количество основных потоков.
Отправить задание двумя способами:
scheduleAtFixedRate: выполнять периодически с фиксированной скоростью
scheduleWithFixedDelay: последняя задача выполняется после фиксированной временной задержки.
Сценарии использования: сценарии, в которых задачи выполняются периодически и количество потоков необходимо ограничить.