Java并發(fā)線程池實例分析講解
更新時間：2023年02月02日 14:07:35 作者：飛奔的小付
這篇文章主要介紹了Java并發(fā)線程池實例，線程池——控制線程創(chuàng)建、釋放，并通過某種策略嘗試復用線程去執(zhí)行任務的一個管理框架，從而實現(xiàn)線程資源與任務之間一種平衡
一.為什么要用線程池
先來看個簡單的例子
1.直接new Thread的情況：
public static void main(String[] args) throws InterruptedException {
long start = System.currentTimeMillis();
final List<Integer> list = new ArrayList<>();
final Random random = new Random();
for (int i = 0; i < 100000; i++) {
Thread thread = new Thread() {
@Override
public void run() {
list.add(random.nextInt());
}
};
thread.start();
thread.join();
}
System.out.println("執(zhí)行時間:" + (System.currentTimeMillis() - start));
System.out.println("執(zhí)行大小：" + list.size());
}
執(zhí)行時間:6437
執(zhí)行大?。?00000
2.使用線程池時
public static void main(String[] args) throws InterruptedException {
long start = System.currentTimeMillis();
final List<Integer> list = new ArrayList<>();
final Random random = new Random();
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i = 0; i < 100000; i++) {
executorService.execute(()->{
list.add(random.nextInt());
});
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.DAYS);
System.out.println("執(zhí)行時間:" + (System.currentTimeMillis() - start));
System.out.println("執(zhí)行大?。? + list.size());
}
執(zhí)行時間:82
執(zhí)行大小：100000
從執(zhí)行時間可以看出來，使用線程池的效率要遠遠超過直接new Thread。
二.線程池的好處
- 降低資源消耗。通過重復利用已創(chuàng)建的線程降低線程創(chuàng)建和銷毀造成的消耗。
- 提高響應速度。當任務到達時，任務可以不需要的等到線程創(chuàng)建就能立即執(zhí)行。
- 提高線程的可管理性。線程是稀缺資源，如果無限制的創(chuàng)建，不僅會消耗系統(tǒng)資源，還會降低系統(tǒng)的穩(wěn)定性，使用線程池可以進行統(tǒng)一的分配，調(diào)優(yōu)和監(jiān)控。
三.原理解析
四.4種線程池
1.newCachedThreadPool
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
特點：newCachedThreadPool會創(chuàng)建一個可緩存線程池，如果當前線程池的長度超過了處理的需要時，可以靈活的回收空閑的線程，當需要增加時，它可以靈活的添加新的線程，而不會對線程池的長度作任何限制。
因為其最大線程數(shù)是Integer.MAX_VALUE，若新建的線程數(shù)多了，會超過機器的可用內(nèi)存而OOM，但是因為其不是無界隊列，所以在OOM之前一般會CPU 100%。
2.newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
該方法會創(chuàng)建一個固定長度的線程池，控制最大并發(fā)數(shù)，超出的線程會在隊列中等待，因為線程的數(shù)量是固定的，但是阻塞隊列是無界的，如果請求數(shù)較多時，會造成阻塞隊列越來越長，超出可用內(nèi)存 進而OOM，所以要根據(jù)系統(tǒng)資源設置線程池的大小。Runtime.getRuntime().availableProcessors()
3.newSingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
會創(chuàng)建一個單一的線程，前一個任務執(zhí)行完畢才會執(zhí)行下一個線程，F(xiàn)IFO，保證順序執(zhí)行。但是高并發(fā)下不太適用
4.newScheduledThreadPool
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
創(chuàng)建一個固定長度的線程池，而且支持定時的以及周期性的任務執(zhí)行，所有任務都是串行執(zhí)行的，同一時間只能有一個任務在執(zhí)行，前一個任務的延遲或異常都將會影響到之后的任務。
阿里規(guī)范中不推薦使用以上線程池，推薦使用自定義的線程池，當然如果你的項目中的數(shù)量級比較小的話那到?jīng)]什么影響。
自定義線程池：
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 20,
0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(10),new MonkeyRejectedExecutionHandler());
執(zhí)行優(yōu)先級 ： 核心線程>非核心線程>隊列
提交優(yōu)先級 ： 核心線程>隊列>非核心線程
五.線程池處理流程
流程圖：
六.源碼分析
流程圖
ThreadPoolExecutor的execute方法
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//1.判斷線程數(shù)是否小于核心線程數(shù)，如果是則使用入?yún)⑷蝿胀ㄟ^addWorker方法創(chuàng)建一個新的線程，如果能完成新線程創(chuàng)建execute方法結束，成功提交任務
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//2.在第一步?jīng)]有完成任務提交；狀態(tài)為運行并且能成功加入任務到工作隊列后，再進行一次check，如果狀態(tài)在任務加入隊列后變?yōu)榱朔沁\行（有可能是在執(zhí)行到這里線程池shtdown了）,非運行狀態(tài)下當然是需要reject；
// offer和add方法差不多，add方法就是調(diào)用的offer,只不過比offer多拋出一個異常 throw new IllegalStateException("Queue full")
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
//3.判斷當前工作線程池數(shù)是否為0,如果是創(chuàng)建一個null任務，任務在堵塞隊列存在了就會從隊列中取出這樣做的意義是保證線程池在running狀態(tài)必須有一個任務在執(zhí)行
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//4.如果不能加入任務到工作隊列，將嘗試使用任務新增一個線程，如果失敗，則是線程池已經(jīng)shutdown或者線程池已經(jīng)達到飽和狀態(tài)，所以reject.拒絕策略不僅僅是在飽和狀態(tài)下使用，在線程池進入到關閉階段同樣需要使用到；
else if (!addWorker(command, false))
reject(command);
}
}
再進入到addWork方法
private boolean addWorker(Runnable firstTask, boolean core) {
// goto寫法 重試
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
//線程狀態(tài)非運行并且非shutdown狀態(tài)任務為空，隊列非空就不能新增線程了
return false;
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
//當前線程達到了最大閾值 就不再新增線程了
return false;
if (compareAndIncrementWorkerCount(c))
//ctl+1工作線程池數(shù)量+1如果成功 就跳出死循環(huán)
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
//進來的狀態(tài)和此時的狀態(tài)發(fā)生改變重頭開始重試
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);//內(nèi)部類封裝了線程和任務 通過threadfactory創(chuàng)建線程
//毎一個worker就是一個線程數(shù)
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//重新獲取線程狀態(tài)
int rs = runStateOf(ctl.get());
// 狀態(tài)小于shutdown 就是running狀態(tài) 或者 為shutdown并且firstTask為空是從隊列中處理 任務那就可以放到集合中
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// 線程還沒start就是alive就直接異常
if (t.isAlive())
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
// 記錄最大線程數(shù)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
//失敗回退從wokers移除w線程數(shù)減1嘗試結束線程池
addWorkerFailed(w);
}
return workerStarted;
}
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
//正在運行woker線程
final Thread thread;
/** Initial task to run. Possibly null. */
//傳入的任務
Runnable firstTask;
/** Per-thread task counter */
//完成的任務數(shù)監(jiān)控用
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
//禁止線程中斷
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
再來看runworker方法
final void runWorker(Worker w) {
//獲取當前線程
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts 把state從‐1改為0意思是可以允許中斷
boolean completedAbruptly = true;
try {
//task不為空或者阻塞隊列中拿到了任務
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
//如果當前線程池狀態(tài)等于stop就中斷
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
//這設置為空等下次循環(huán)就會從隊列里面獲取
task = null;
//完成任務數(shù)+1
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
獲取任務的方法
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);//獲取線程池運行狀態(tài)
// Check if queue empty only if necessary.
//shutdown或者為空那就工作線程‐1同時返回為null
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
//重新獲取工作線程數(shù)
int wc = workerCountOf(c);
// Are workers subject to culling?
// timed是標志超時銷毀 核心線程池也是可以銷毀的
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
runWorker中的processWorkerExit
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
ThreadPoolExecutor內(nèi)部有實現(xiàn)4個拒絕策略：(1)、
- CallerRunsPolicy，由調(diào)用execute方法提交任務的線程來執(zhí)行這個任務；
- AbortPolicy，拋出異常RejectedExecutionException拒絕提交任務；
- DiscardPolicy，直接拋棄任務，不做任何處理；
- DiscardOldestPolicy，去除任務隊列中的第一個任務（最舊的），重新提交
ScheduledThreadPoolExecutor
- schedule：延遲多長時間之后只執(zhí)行一次；
- scheduledAtFixedRate固定：延遲指定時間后執(zhí)行一次，之后按照固定的時長周期執(zhí)行；
- scheduledWithFixedDelay非固定：延遲指定時間后執(zhí)行一次，之后按照：上一次任務執(zhí)行時長+周期的時長的時間去周期執(zhí)行；
private void delayedExecute(RunnableScheduledFuture<?> task) {
//如果線程池不是RUNNING狀態(tài)，則使用拒絕策略把提交任務拒絕掉
if (isShutdown())
reject(task);
else {
//與ThreadPoolExecutor不同，這里直接把任務加入延遲隊列
super.getQueue().add(task);
//如果當前狀態(tài)無法執(zhí)行任務，則取消
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
else
//和ThreadPoolExecutor不一樣，corePoolSize沒有達到會增加Worker;
//增加Worker，確保提交的任務能夠被執(zhí)行
ensurePrestart();
}
}
add方法里其實是調(diào)用了offer方法
public boolean add(Runnable e) {
return offer(e);
}
public boolean offer(Runnable x) {
if (x == null)
throw new NullPointerException();
RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
final ReentrantLock lock = this.lock;
lock.lock();
try {
int i = size;
if (i >= queue.length)
//容量擴增50%
grow();
size = i + 1;
if (i == 0) {
queue[0] = e;
setIndex(e, 0);
} else {
//插入堆尾
siftUp(i, e);
}
if (queue[0] == e) {
//如果新加入的元素成為了堆頂,則原先的leader就無效了
leader = null;
//由于原先leader已經(jīng)無效被設置為null了,這里隨便喚醒一個線程(未必是原先的leader)來取走堆頂任務
available.signal();
}
} finally {
lock.unlock();
}
return true;
}
siftup方法：主要是對隊列進行排序
private void siftUp(int k, RunnableScheduledFuture<?> key) {
while (k > 0) {
//獲取父節(jié)點
int parent = (k - 1) >>> 1;
RunnableScheduledFuture<?> e = queue[parent];
//如果key節(jié)點的執(zhí)行時間大于父節(jié)點的執(zhí)行時間，不需要再排序了
if (key.compareTo(e) >= 0)
break;
//如果key.compareTo(e)<0，說明key節(jié)點的執(zhí)行時間小于父節(jié)點的執(zhí)行時間，需要把父節(jié)點移到后面
queue[k] = e;
setIndex(e, k);
//設置索引為k
k = parent;
}
//key設置為排序后的位置中
queue[k] = key;
setIndex(key, k);
}
run方法：
public void run() {
//是否周期性，就是判斷period是否為0
boolean periodic = isPeriodic();
//檢查任務是否可以被執(zhí)行
if (!canRunInCurrentRunState(periodic))
cancel(false);
//如果非周期性任務直接調(diào)用run運行即可
else if (!periodic)
ScheduledFutureTask.super.run();
//如果成功runAndRest，則設置下次運行時間并調(diào)用reExecutePeriodic
else if (ScheduledFutureTask.super.runAndReset()) {
setNextRunTime();
//需要重新將任務(outerTask)放到工作隊列中。此方法源碼會在后文介紹ScheduledThreadPoolExecutor本身API時提及
reExecutePeriodic(outerTask);
}
}
private void setNextRunTime() {
long p = period;
//fixed‐rate模式，時間設置為上一次時間+p,這里的時間只是可以被執(zhí)行的最小時間，不代表到點就要執(zhí)行
if (p > 0)
time += p;
else
//fixed‐delay模式，計算下一次任務可以被執(zhí)行的時間, 差不多就是當前時間+delay值
time = triggerTime(-p);
}
long triggerTime(long delay) {
//如果delay<Long.Max_VALUE/2,則下次執(zhí)行時間為當前時間+delay,否則為了避免隊列中出現(xiàn)由于溢出導致的排序紊亂,需要調(diào)用overflowFree來修正一下delay
return now() +
((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}
/**
* 主要就是有這么一種情況：
* 工作隊列中維護任務順序是基于compareTo的，在compareTo中比較兩個任務的順序會用time相減，負數(shù)則說明優(yōu)先級高,那么就有可能出現(xiàn)一個delay為正數(shù),減去另一個為負數(shù)的delay，結果上溢為負數(shù)，則會導致compareTo產(chǎn)生錯誤的結果.
* 為了特殊處理這種情況，首先判斷一下隊首的delay是不是負數(shù)，如果是正數(shù)不用管了,怎么減都不會溢出。
* 否則可以拿當前delay減去隊首的delay來比較看，如果不出現(xiàn)上溢，則整個隊列都ok，排序不會亂。
* 不然就把當前delay值給調(diào)整為Long.MAX_VALUE+隊首delay
/
private long overflowFree(long delay) {
Delayed head = (Delayed) super.getQueue().peek();
if (head != null) {
long headDelay = head.getDelay(NANOSECONDS);
if (headDelay < 0 && (delay - headDelay < 0))
delay = Long.MAX_VALUE + headDelay;
}
return delay;
}
到此這篇關于Java并發(fā)線程池實例分析講解的文章就介紹到這了,更多相關Java并發(fā)線程池內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
mybatis collection關聯(lián)查詢多個參數(shù)方式
在使用MyBatis進行關聯(lián)查詢時,往往需要根據(jù)多個參數(shù)進行查詢,例如,使用evtId和businessType作為查詢條件,同時在resultMap中配置id和businessType1作為結果映射,這種情況下,可以通過<sql>標簽定義參數(shù)模板,或者使用@Param注解指定參數(shù)名稱2024-10-10
Java實現(xiàn)調(diào)用jython執(zhí)行python文件的方法
這篇文章主要介紹了Java實現(xiàn)調(diào)用jython執(zhí)行python文件的方法,結合實例形式分析了Java調(diào)用jython執(zhí)行python文件的常見操作技巧及相關問題解決方法,需要的朋友可以參考下2018-03-03
這是一篇入門級文章，高手請略過。在這篇文章中我們將學習如何用 Java 對圖像進行剪裁并將剪裁出來的部分單獨保存到文件中。2014-09-09
連接池就是用于存儲數(shù)據(jù)庫連接的一個容器,容器其實就是一個集合對象，本文給大家介紹MyBatis中的連接池以及事物控制的相關知識，感興趣的朋友一起看看吧2021-05-05
使用mybatis的interceptor修改執(zhí)行sql以及傳入?yún)?shù)方式
這篇文章主要介紹了使用mybatis的interceptor修改執(zhí)行sql以及傳入?yún)?shù)方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12
Java?IO流之StringWriter和StringReader用法分析
這篇文章主要介紹了Java?IO流之StringWriter和StringReader用法分析，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12
Spring中@PropertySource的使用方法和運行原理詳解
這篇文章主要介紹了Spring中@PropertySource的使用方法和運行原理詳解,PropertySource注解可以方便和靈活的向Spring的環(huán)境容器(org.springframework.core.env.Environment?Environment)中注入一些屬性,這些屬性可以在Bean中使用,需要的朋友可以參考下2023-11-11
這篇文章主要為大家詳細介紹了Java8實現(xiàn)任意參數(shù)的鏈棧，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2020-10-10
詳解java封裝返回結果與RestControllerAdvice注解
這篇文章主要為大家介紹了java封裝返回結果與RestControllerAdvice注解實例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-09-09
Java?Collections.sort()實現(xiàn)List排序的默認方法和自定義方法
這篇文章主要介紹了Java?Collections.sort()實現(xiàn)List排序的默認方法和自定義方法,本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2017-06-06