【Java】線程池梳理
點(diǎn)擊關(guān)注，與你共同成長(zhǎng)！
【Java】線程池梳理
前言
線程池：本質(zhì)上是一種對(duì)象池，用于管理線程資源。在任務(wù)執(zhí)行前，需要從線程池中拿出線程來(lái)執(zhí)行。在任務(wù)執(zhí)行完成之后，需要把線程放回線程池。通過(guò)線程的這種反復(fù)利用機(jī)制，可以有效地避免直接創(chuàng)建線程所帶來(lái)的壞處。
優(yōu)點(diǎn)：1、降低資源的消耗。線程本身是一種資源，創(chuàng)建和銷毀線程會(huì)有CPU開(kāi)銷；創(chuàng)建的線程也會(huì)占用一定的內(nèi)存；2、提高任務(wù)執(zhí)行的響應(yīng)速度。任務(wù)執(zhí)行時(shí)，可以不必等到線程創(chuàng)建完之后再執(zhí)行；3、提高線程的可管理性。線程不能無(wú)限制地創(chuàng)建，需要進(jìn)行統(tǒng)一的分配、調(diào)優(yōu)和監(jiān)控。
缺點(diǎn)：1、頻繁的線程創(chuàng)建和銷毀會(huì)占用更多的CPU和內(nèi)存；2、頻繁的線程創(chuàng)建和銷毀會(huì)對(duì)GC產(chǎn)生比較大的壓力；3、線程太多，線程切換帶來(lái)的開(kāi)銷將不可忽視；4、線程太少，多核CPU得不到充分利用，是一種浪費(fèi)。
流程
判斷核心線程池是否已滿，如果不是，則創(chuàng)建線程執(zhí)行任務(wù)； 如果核心線程池滿了，判斷隊(duì)列是否滿了，如果隊(duì)列沒(méi)滿，將任務(wù)放在隊(duì)列中； 如果隊(duì)列滿了，則判斷線程池是否已滿，如果沒(méi)滿，創(chuàng)建線程執(zhí)行任務(wù)； 如果線程池也滿了，則按照拒絕策略對(duì)任務(wù)進(jìn)行處理。
方式
入門級(jí)例子
package cn.com.codingce.juc;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolTest {
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
executor.submit(() -> {
System.out.println("Thread id is " + Thread.currentThread().getId());
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
}
在這個(gè)例子中，首先創(chuàng)建了一個(gè)固定長(zhǎng)度為5的線程池。然后使用循環(huán)的方式往線程池中提交了10個(gè)任務(wù)，每個(gè)任務(wù)休眠1秒。在任務(wù)休眠之前，將任務(wù)所在的線程id進(jìn)行打印輸出。
Thread id is 11
Thread id is 13
Thread id is 12
Thread id is 15
Thread id is 14
Thread id is 11
Thread id is 13
Thread id is 15
Thread id is 14
Thread id is 12
Executors
Executors是一個(gè)線程池工廠，提供了很多的工廠方法。
// 創(chuàng)建單一線程的線程池
public static ExecutorService newSingleThreadExecutor();
// 創(chuàng)建固定數(shù)量的線程池
public static ExecutorService newFixedThreadPool(int nThreads);
// 創(chuàng)建帶緩存的線程池
public static ExecutorService newCachedThreadPool();
// 創(chuàng)建定時(shí)調(diào)度的線程池
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize);
// 創(chuàng)建流式（fork-join）線程池
public static ExecutorService newWorkStealingPool();
newSingleThreadExecutor
創(chuàng)建一個(gè)單線程的線程池，若多個(gè)任務(wù)被提交到此線程池，那么會(huì)被緩存到隊(duì)列（隊(duì)列長(zhǎng)度為Integer.MAX_VALUE ），可保證所有任務(wù)按照指定順序(FIFO, LIFO, 優(yōu)先級(jí))執(zhí)行。
private static void createSingleThreadPool() {
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
final int index = i;
executorService.execute(() -> {
// 獲取線程名稱,默認(rèn)格式:pool-1-thread-1
System.out.println(new Date() + " " + Thread.currentThread().getName() + " " + index);
// 等待2秒
try {
sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
output
Mon Jan 02 11:49:58 CST 2023 pool-1-thread-1 0
Mon Jan 02 11:50:00 CST 2023 pool-1-thread-1 1
Mon Jan 02 11:50:02 CST 2023 pool-1-thread-1 2
Mon Jan 02 11:50:04 CST 2023 pool-1-thread-1 3
Mon Jan 02 11:50:06 CST 2023 pool-1-thread-1 4
Mon Jan 02 11:50:08 CST 2023 pool-1-thread-1 5
Mon Jan 02 11:50:10 CST 2023 pool-1-thread-1 6
Mon Jan 02 11:50:12 CST 2023 pool-1-thread-1 7
Mon Jan 02 11:50:14 CST 2023 pool-1-thread-1 8
Mon Jan 02 11:50:16 CST 2023 pool-1-thread-1 9
因?yàn)橹挥幸粋€(gè)線程，所以線程名均相同，且是每隔2秒按順序輸出的。
newFixedThreadPool
創(chuàng)建一個(gè)固定大小的線程池，可控制并發(fā)的線程數(shù)，超出的線程會(huì)在隊(duì)列中等待。和創(chuàng)建單一線程的線程池類似，只是可以并行處理任務(wù)的線程數(shù)更多一些。若多個(gè)任務(wù)被提交到此線程池，會(huì)有下面的處理過(guò)程。
如果線程的數(shù)量未達(dá)到指定數(shù)量，則創(chuàng)建線程來(lái)執(zhí)行任務(wù)； 如果線程池的數(shù)量達(dá)到了指定數(shù)量，并且有線程是空閑的，則取出空閑線程執(zhí)行任務(wù)； 如果沒(méi)有線程是空閑的，則將任務(wù)緩存到隊(duì)列（隊(duì)列長(zhǎng)度為 Integer.MAX_VALUE）。當(dāng)線程空閑的時(shí)候，按照FIFO的方式進(jìn)行處理
private static void createFixedThreadPool() {
ExecutorService executorService = Executors.newFixedThreadPool(3);
for (int i = 0; i < 10; i++) {
final int index = i;
executorService.execute(() -> {
// 獲取線程名稱,默認(rèn)格式:pool-1-thread-1
System.out.println(new Date() + " " + Thread.currentThread().getName() + " " + index);
// 等待2秒
try {
sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
output
Mon Jan 02 11:49:10 CST 2023 pool-1-thread-2 1
Mon Jan 02 11:49:10 CST 2023 pool-1-thread-1 0
Mon Jan 02 11:49:10 CST 2023 pool-1-thread-3 2
Mon Jan 02 11:49:12 CST 2023 pool-1-thread-2 3
Mon Jan 02 11:49:12 CST 2023 pool-1-thread-3 5
Mon Jan 02 11:49:12 CST 2023 pool-1-thread-1 4
Mon Jan 02 11:49:14 CST 2023 pool-1-thread-1 6
Mon Jan 02 11:49:14 CST 2023 pool-1-thread-2 7
Mon Jan 02 11:49:14 CST 2023 pool-1-thread-3 8
Mon Jan 02 11:49:16 CST 2023 pool-1-thread-2 9
因?yàn)榫€程池大小是固定的，這里設(shè)置的是3個(gè)線程，所以線程名只有3個(gè)。因?yàn)榫€程不足會(huì)進(jìn)入隊(duì)列等待線程空閑，所以日志間隔2秒輸出。
newCachedThreadPool
創(chuàng)建一個(gè)可緩存的線程池，若線程數(shù)超過(guò)處理所需，緩存一段時(shí)間后會(huì)回收，若線程數(shù)不夠，則新建線程。這種方式創(chuàng)建的線程池，核心線程池的長(zhǎng)度為0，線程池最大長(zhǎng)度為Integer.MAX_VALUE。由于本身使用SynchronousQueue作為等待隊(duì)列的緣故，導(dǎo)致往隊(duì)列里面每插入一個(gè)元素，必須等待另一個(gè)線程從這個(gè)隊(duì)列刪除一個(gè)元素。
private static void createCachedThreadPool() {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int index = i;
executorService.execute(() -> {
// 獲取線程名稱,默認(rèn)格式:pool-1-thread-1
System.out.println(new Date() + " " + Thread.currentThread().getName() + " " + index);
// 等待2秒
try {
sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
output
Mon Jan 02 11:56:03 CST 2023 pool-1-thread-8 7
Mon Jan 02 11:56:03 CST 2023 pool-1-thread-3 2
Mon Jan 02 11:56:03 CST 2023 pool-1-thread-1 0
Mon Jan 02 11:56:03 CST 2023 pool-1-thread-5 4
Mon Jan 02 11:56:03 CST 2023 pool-1-thread-9 8
Mon Jan 02 11:56:03 CST 2023 pool-1-thread-6 5
Mon Jan 02 11:56:03 CST 2023 pool-1-thread-2 1
Mon Jan 02 11:56:03 CST 2023 pool-1-thread-4 3
Mon Jan 02 11:56:03 CST 2023 pool-1-thread-7 6
Mon Jan 02 11:56:03 CST 2023 pool-1-thread-10 9
因?yàn)槌跏季€程池沒(méi)有線程，而線程不足會(huì)不斷新建線程，所以線程名都是不一樣的。
newScheduledThreadPool
創(chuàng)建一個(gè)周期性的線程池，支持定時(shí)及周期性執(zhí)行任務(wù)。
private static void createScheduledThreadPool() {
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);
System.out.println(new Date() + " 提交任務(wù)");
for (int i = 0; i < 10; i++) {
final int index = i;
executorService.schedule(() -> {
// 獲取線程名稱,默認(rèn)格式:pool-1-thread-1
System.out.println(new Date() + " " + Thread.currentThread().getName() + " " + index);
// 等待2秒
try {
sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}, 3, TimeUnit.SECONDS);
}
}
Output
Mon Jan 02 11:59:19 CST 2023 提交任務(wù)
Mon Jan 02 11:59:22 CST 2023 pool-1-thread-1 0
Mon Jan 02 11:59:22 CST 2023 pool-1-thread-2 1
Mon Jan 02 11:59:22 CST 2023 pool-1-thread-3 2
Mon Jan 02 11:59:24 CST 2023 pool-1-thread-2 4
Mon Jan 02 11:59:24 CST 2023 pool-1-thread-1 5
Mon Jan 02 11:59:24 CST 2023 pool-1-thread-3 3
Mon Jan 02 11:59:26 CST 2023 pool-1-thread-2 6
Mon Jan 02 11:59:26 CST 2023 pool-1-thread-1 8
Mon Jan 02 11:59:26 CST 2023 pool-1-thread-3 7
Mon Jan 02 11:59:28 CST 2023 pool-1-thread-3 9
因?yàn)樵O(shè)置了延遲3秒，所以提交后3秒才開(kāi)始執(zhí)行任務(wù)。因?yàn)檫@里設(shè)置核心線程數(shù)為3個(gè)，而線程不足會(huì)進(jìn)入隊(duì)列等待線程空閑，所以日志間隔2秒輸出。
newWorkStealingPool(jdk1.8新增)
創(chuàng)建一個(gè)含有足夠多線程的線程池，來(lái)維持相應(yīng)的并行級(jí)別，它會(huì)通過(guò)工作竊取的方式，使得多核的 CPU 不會(huì)閑置，總會(huì)有活著的線程讓 CPU 去運(yùn)行。
工作竊取概念（Work stealing）：工作竊取不是什么 Java 獨(dú)有的東西，.NET 的 TPL 庫(kù)早就存在好幾年了。所謂工作竊取，指的是閑置的線程去處理本不屬于它的任務(wù)。每個(gè)處理器核，都有一個(gè)隊(duì)列存儲(chǔ)著需要完成的任務(wù)。對(duì)于多核的機(jī)器來(lái)說(shuō)，當(dāng)一個(gè)核對(duì)應(yīng)的任務(wù)處理完畢后，就可以去幫助其他的核處理任務(wù)。
private static void createNewWorkStealingPool() {
ExecutorService forkJoin = Executors.newWorkStealingPool();
forkJoin.execute(() -> {
System.out.println("i====>" + 1 + " " + Thread.currentThread().getId());
});
forkJoin.execute(() -> {
System.out.println("i====>" + 2 + " " + Thread.currentThread().getId());
});
forkJoin.execute(() -> {
System.out.println("i====>" + 3 + " " + Thread.currentThread().getId());
});
forkJoin.execute(() -> {
System.out.println("i====>" + 4 + " " + Thread.currentThread().getId());
});
forkJoin.execute(() -> {
System.out.println("i====>" + 5 + " " + Thread.currentThread().getId());
});
}
output
i====>1 11
i====>2 11
i====>3 12
i====>4 12
i====>5 12
ThreadPoolExecutor
理論上，可以通過(guò)Executors來(lái)創(chuàng)建線程池，這種方式非常簡(jiǎn)單。但正是因?yàn)楹?jiǎn)單，所以限制了線程池的功能。比如：無(wú)長(zhǎng)度限制的隊(duì)列，可能因?yàn)槿蝿?wù)堆積導(dǎo)致OOM，這是非常嚴(yán)重的bug，應(yīng)盡可能地避免。怎么避免？歸根結(jié)底，還是需要通過(guò)更底層的方式來(lái)創(chuàng)建線程池。
ThreadPoolExecutor提供了好幾個(gè)構(gòu)造方法，但是最底層的構(gòu)造方法卻只有一個(gè)。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {}
這個(gè)構(gòu)造方法有7個(gè)參數(shù)，逐一來(lái)進(jìn)行分析。
corePoolSize，線程池中的核心線程數(shù)；maximumPoolSize，線程池中的最大線程數(shù)；keepAliveTime，空閑時(shí)間，當(dāng)線程池?cái)?shù)量超過(guò)核心線程數(shù)時(shí)，多余的空閑線程存活的時(shí)間，即：這些線程多久被銷毀；unit，空閑時(shí)間的單位，可以是毫秒、秒、分鐘、小時(shí)和天，等等；workQueue，等待隊(duì)列，線程池中的線程數(shù)超過(guò)核心線程數(shù)時(shí)，任務(wù)將放在等待隊(duì)列，它是一個(gè)BlockingQueue類型的對(duì)象；ArrayBlockingQueue，隊(duì)列是有界的，基于數(shù)組實(shí)現(xiàn)的阻塞隊(duì)列；LinkedBlockingQueue，隊(duì)列可以有界，也可以無(wú)界。基于鏈表實(shí)現(xiàn)的阻塞隊(duì)列；SynchronousQueue，不存儲(chǔ)元素的阻塞隊(duì)列，每個(gè)插入操作必須等到另一個(gè)線程調(diào)用移除操作，否則插入操作將一直處于阻塞狀態(tài)。該隊(duì)列也是Executors.newCachedThreadPool()的默認(rèn)隊(duì)列；PriorityBlockingQueue，帶優(yōu)先級(jí)的無(wú)界阻塞隊(duì)列。threadFactory，線程工廠，可以使用它來(lái)創(chuàng)建一個(gè)線程；Executors的實(shí)現(xiàn)使用了默認(rèn)的線程工廠-DefaultThreadFactory。它的實(shí)現(xiàn)主要用于創(chuàng)建一個(gè)線程，線程的名字為pool-{poolNum}-thread-{threadNum}。handler，拒絕策略，當(dāng)線程池和等待隊(duì)列都滿了之后，需要通過(guò)該對(duì)象的回調(diào)函數(shù)進(jìn)行回調(diào)處理。AbortPolicy：丟棄任務(wù)，拋運(yùn)行時(shí)RejectedExecutionException異常； CallerRunsPolicy：在調(diào)用者線程執(zhí)行任務(wù)； DiscardPolicy：忽視，任務(wù)直接丟棄，什么都不會(huì)發(fā)生； DiscardOldestPolicy：從隊(duì)列中踢出最先進(jìn)入隊(duì)列（最后一個(gè)執(zhí)行）的任務(wù)（最舊的那個(gè)任務(wù)），再嘗試執(zhí)行當(dāng)前任務(wù)。
線程池的執(zhí)行規(guī)則如下：1、當(dāng)線程數(shù)小于核心線程數(shù)時(shí)，創(chuàng)建線程；2、當(dāng)線程數(shù)大于等于核心線程數(shù)，且任務(wù)隊(duì)列未滿時(shí)，將任務(wù)放入任務(wù)隊(duì)列；3、當(dāng)線程數(shù)大于等于核心線程數(shù)，且任務(wù)隊(duì)列已滿。若線程數(shù)小于最大線程數(shù)，創(chuàng)建線程。若線程數(shù)等于最大線程數(shù)，拋出異常，拒絕任務(wù)。
private static void createThreadPool() {
ExecutorService executorService = new ThreadPoolExecutor(2, 10, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<>(5, true), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
for (int i = 0; i < 10; i++) {
final int index = i;
executorService.execute(() -> {
// 獲取線程名稱,默認(rèn)格式:pool-1-thread-1
System.out.println(new Date() + " " + Thread.currentThread().getName() + " " + index);
// 等待2秒
try {
sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
executorService.shutdown();
}
output
Mon Jan 02 12:34:11 CST 2023 pool-1-thread-1 0
Mon Jan 02 12:34:11 CST 2023 pool-1-thread-2 1
Mon Jan 02 12:34:11 CST 2023 pool-1-thread-3 7
Mon Jan 02 12:34:11 CST 2023 pool-1-thread-5 9
Mon Jan 02 12:34:11 CST 2023 pool-1-thread-4 8
Mon Jan 02 12:34:13 CST 2023 pool-1-thread-4 2
Mon Jan 02 12:34:13 CST 2023 pool-1-thread-1 3
Mon Jan 02 12:34:13 CST 2023 pool-1-thread-2 4
Mon Jan 02 12:34:13 CST 2023 pool-1-thread-5 5
Mon Jan 02 12:34:13 CST 2023 pool-1-thread-3 6
因?yàn)楹诵木€程數(shù)為2，隊(duì)列大小為5，存活時(shí)間1分鐘，所以流程是第0-1號(hào)任務(wù)來(lái)時(shí)，陸續(xù)創(chuàng)建2個(gè)線程，然后第2-6號(hào)任務(wù)來(lái)時(shí)，因?yàn)闊o(wú)線程可用，均進(jìn)入了隊(duì)列等待，第7-9號(hào)任務(wù)來(lái)時(shí)，沒(méi)有空閑線程，隊(duì)列也滿了，所以陸續(xù)又創(chuàng)建了3個(gè)線程。所以你會(huì)發(fā)現(xiàn)7-9號(hào)任務(wù)反而是先執(zhí)行的。又因?yàn)楦魅蝿?wù)只需要2秒，而線程存活時(shí)間有1分鐘，所以線程進(jìn)行了復(fù)用，所以總共只創(chuàng)建了5個(gè)線程。
如何正確配置線程池的參數(shù)：CPU密集型：corePoolSize = CPU核數(shù) + 1；IO密集型：corePoolSize = CPU核數(shù) * 2。
提交任務(wù)的幾種方式：往線程池中提交任務(wù)，主要有兩種方法，execute()和submit()。submit()用于提交一個(gè)需要返回果的任務(wù)。該方法返回一個(gè)Future對(duì)象，通過(guò)調(diào)用這個(gè)對(duì)象的get()方法，就能獲得返回結(jié)果。get()方法會(huì)一直阻塞，直到返回結(jié)果返回。另外，也可以使用它的重載方法get(long timeout, TimeUnit unit)，這個(gè)方法也會(huì)阻塞，但是在超時(shí)時(shí)間內(nèi)仍然沒(méi)有返回結(jié)果時(shí)，將拋出異常TimeoutException。
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(2);
Future<Long> future = executor.submit(() -> {
System.out.println("task is executed");
return System.currentTimeMillis();
});
System.out.println("task execute time is: " + future.get());
}
output
task is executed
task execute time is: 1672634764296
線程池監(jiān)控：
ThreadPoolExecutor自帶： long getTaskCount()：獲取已經(jīng)執(zhí)行或正在執(zhí)行的任務(wù)數(shù)；long getCompletedTaskCount()：獲取已經(jīng)執(zhí)行的任務(wù)數(shù)；int getLargestPoolSize()：獲取線程池曾經(jīng)創(chuàng)建過(guò)的最大線程數(shù)，根據(jù)這個(gè)參數(shù)，可以知道線程池是否滿過(guò)；int getPoolSize()：獲取線程池線程數(shù)；int getActiveCount()：獲取活躍線程數(shù)（正在執(zhí)行任務(wù)的線程數(shù)）。ThreadPoolExecutor自定義處理： protected void beforeExecute(Thread t, Runnable r)：任務(wù)執(zhí)行前被調(diào)用；protected void afterExecute(Runnable r, Throwable t)：任務(wù)執(zhí)行后被調(diào)用；protected void terminated()：線程池結(jié)束后被調(diào)用。
關(guān)閉線程池：1、shutdown()會(huì)將線程池狀態(tài)置為SHUTDOWN，不再接受新的任務(wù)，同時(shí)會(huì)等待線程池中已有的任務(wù)執(zhí)行完成再結(jié)束；2、shutdownNow()會(huì)將線程池狀態(tài)置為SHUTDOWN，對(duì)所有線程執(zhí)行interrupt()操作，清空隊(duì)列，并將隊(duì)列中的任務(wù)返回回來(lái)。關(guān)閉線程池涉及到兩個(gè)返回boolean的方法，isShutdown()和isTerminated，分別表示是否關(guān)閉和是否終止。
注意
盡量使用手動(dòng)的方式創(chuàng)建線程池，避免使用 Executors工廠類；根據(jù)場(chǎng)景，合理設(shè)置線程池的各個(gè)參數(shù)，包括線程池?cái)?shù)量、隊(duì)列、線程工廠和拒絕策略； 在調(diào)線程池 submit()方法的時(shí)候，一定要盡量避免任務(wù)執(zhí)行異常被吞掉的問(wèn)題。
以上，便是今天的分享，希望大家喜歡，覺(jué)得內(nèi)容不錯(cuò)的，歡迎「分享」「贊」或者點(diǎn)擊「在看」支持，謝謝各位。