資訊專欄INFORMATION COLUMN
摘要：而且，線程池中的線程并沒有睡眠，而是進入了自旋狀態(tài)。普通的線程被中斷會導致線程繼續(xù)執(zhí)行，從而方法運行完畢，線程退出。線程死亡超過時間，任務對列沒有數據而返回。線程死亡保證了線程池至少留下個線程。
線程在執(zhí)行任務時，正常的情況是這樣的：
Thread t=new Thread(new Runnable() {
@Override
public void run() {
// TODO Auto-generated method stub
}
});
t.start();
??Thread 在初始化的時候傳入一個Runnable,以后就沒有機會再傳入一個Runable了。那么，woker作為一個已經啟動的線程。是如何不斷獲取Runnable的呢？
這個時候可以使用一個包裝器，將線程包裝起來，在Run方法內部獲取任務。
public final class Worker implements Runnable {
Thread thread = null;
Runnable task;
private BlockingQueue queues;
public Worker(Runnable task, BlockingQueue queues) {
this.thread = new Thread(this);
this.task = task;
this.queues = queues;
}
public void run() {
if (task != null) {
task.run();
}
try {
while (true) {
task = queues.take();
if (task != null) {
task.run();
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void start() {
this.thread.start();
}
}
public class Main {
public static void main(String[] args) {
BlockingQueue queues=new ArrayBlockingQueue(100);
Worker worker=new Worker(new Runnable() {
public void run() {
System.out.println("hello!!! ");
try {
Thread.currentThread().sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, queues);
worker.start();
for(int i=0;i<100;i++){
queues.offer(new Runnable() {
public void run() {
System.out.println("hello!!! ");
try {
Thread.currentThread().sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
}
??這樣我們就簡單地實現(xiàn)了一個“線程池”（可以將這個“線程池”改造成官方的模式，不過可以自己嘗試一下）。ThreadPool的這種實現(xiàn)模式是并發(fā)編程中經典的Cyclic Work Distribution模式。
??那么，這種實現(xiàn)的線程池性能如何呢？
??由于其任務隊列使用的是阻塞隊列，在隊列內部是自旋的。Reeteenlok是改進的CLH隊列。自旋鎖會耗費一定CPU的資源，在擁有大量任務執(zhí)行下的情況下比較有效。而且，線程池中的線程并沒有睡眠，而是進入了自旋狀態(tài)。
??如果是不支持超線程的CPU，在同一時刻的確只能處理2個線程，但是并不意味著雙核的CPU只能處理兩個線程，它可以通過切換上下文來執(zhí)行多個線程。比如我只有一個大腦，但是我要處理5個人提交的任務，我可以處理完A的事情后，把事情的中間結果保存下，然后再處理B的，然后再讀取A的中間結果，處理A的事情。
JDK中的線程池實現(xiàn)分析??Woker自身繼承了Runnable,并對Thread做了一個包裝。Woker代碼如下所示：
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
Runnable firstTask;
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
protected boolean isHeldExclusively() {
return getState() != 0;
}
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
??execute(Runnable command)方法內部是這樣的：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}
??ctl一個合并類型的值。將當前線程數和線程池狀態(tài)通過數學運算合并到了一個值。具體是如何合并的可以參看一下源碼，這里就不敘述了。繼續(xù)向下走：
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
??可以看到，如果當前線程數量小于了核心線程數量corePoolSize，就直接增加線程處理任務。與隊列沒有關系。但是緊接著又檢查了一遍狀態(tài)，因為在這個過程中，別的線程也可能在添加任務。繼續(xù)向下走：
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
??可以看到如果線程池是運行態(tài)的，就把線程添加到任務隊列。workQueue是構造函數傳遞過來的，可以是有界隊列，也可以是無界隊列。可以看出來，隊列如果是無界的，直接往隊列里面添加任務，這個時候，線程池中的線程也不會增加，一直會等于核心線程數。
??如果隊列是有界的，就嘗試直接新增線程處理任務，如果添加任務失敗，就調用reject方法來處理添加失敗的任務：
else if (!addWorker(command, false))
reject(command);
??來看看addWorker是如何實現(xiàn)的，邏輯流程已經直接在注釋中說明了。
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//如果狀態(tài)大于SHUTDOWN，不再接受新的任務，直接返回
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
/**根據core來判斷，如果當前線程數量大于corePoolSize或者最大線程數，直接返回。添加任務失敗。
**如果隊列是有界的或者任務添加到隊列失敗(參數core是false)，那么就會新開一個線程處理業(yè)務，但如果線程已經大于了maximumPoolSize，就會出現(xiàn)添加失敗，返回false。
*/
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
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);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
??如果創(chuàng)建失敗的情況下會調用addWorkerFailed方法，從而將減少實際線程數。
addWorker中for循環(huán)的意義??在addWorker中有這么一段代碼，表示為當前線程數加1：
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
??由于多線程可能同時操作。expect值可能會變化。僅僅一次的操作compareAndIncrementWorkerCount可能一次并不會成功，而且，一個線程在執(zhí)行addWork的過程中間，另外一個線程假設直接shotdown這個線程池。for循環(huán)的存在可以保證狀態(tài)一定是一致的。
任務的執(zhí)行在Worker中間實際上是調用的runWorker方法來執(zhí)行的具體業(yè)務：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
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 {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
??可以看到while循環(huán)不斷的從隊列中取出任務執(zhí)行。如果task==null 并且getTask()等于null的話，那么就會跳出循環(huán)，進入到processWorkerExit，run方法執(zhí)行完畢以后，這個線程也被銷毀了。但是為什么在各自的線程執(zhí)行，為什么還需要加鎖呢？答案是因為要線程池需要判斷這個線程是否在執(zhí)行任務。在interruptIdleWorkers方法中，要中斷那寫目前空閑的線程，通過當前Worker是否獲得了鎖就能判斷這個worker是否是空閑的:
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock(); //中斷不起作用。interrupt()對于自旋鎖是不起作用的。只是邏輯上被阻塞，
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
??可以看到，如果w.tryLock()可以獲取到鎖，那么就意味著當前的 Woker并沒有處理任務（沒有進入到循環(huán)里面或者被getTask方法所阻塞，無法獲取鎖）。
Work之所以繼承AbstractQueuedSynchronizer，而不去使用ReentrantLock。是因為ReentrantLock是可重入鎖，在調用lock方法獲取鎖之后，再調用tryLock()還是會返回true。
public static void main(String[] args) {
ReentrantLock lock = new ReentrantLock();
lock.lock();
System.out.println(lock.tryLock());
}
輸出結果是true，所以使用ReentrantLock則難以判斷當前Worker是否在執(zhí)行任務。
線程超時allowCoreThreadTimeOut、keepAliveTime以及線程死亡??在上面的interruptIdleWorkers方法中，線程被中斷。普通的線程被中斷會導致線程繼續(xù)執(zhí)行，從而run方法運行完畢，線程退出。
對于一個沒有被阻塞的線程，中斷是不起作用的。中斷在如下線程被阻塞的方法中起作用：
the wait(),
wait(long),
wait(long, int)
join(),
join(long),
join(long, int),
sleep(long),
or sleep(long, int)
LockSupport.park(Object object);
LockSupport.park();
??，如果喚醒這些被阻塞的線程，從而能使得run方法繼續(xù)執(zhí)行，當run方法執(zhí)行完畢，那么線程也就終結死亡。但是對于ReentrantLock和AbstractQueuedSynchronizer這種自旋+CAS實現(xiàn)的“邏輯鎖”，是不起作用的。
而且runWork本身也是While循環(huán)，靠中斷是無法退出循環(huán)的。
??但是在ThreadPoolExecutor的構造函數中，有一個允許設置線程超時allowCoreThreadTimeOut參數的方法。如果允許超時，多于corePoolSize的線程將會在處在空閑狀態(tài)之后存活keepAliveTime時長后終止。因此有了一個allowCoreThreadTimeOut方法：
public void allowCoreThreadTimeOut(boolean value) {
if (value && keepAliveTime <= 0)
throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
if (value != allowCoreThreadTimeOut) {
allowCoreThreadTimeOut = value;
if (value)
interruptIdleWorkers();
}
}
?? 正如上面提到的一樣，允許allowCoreThreadTimeOut并且調用interruptIdleWorkers方法并不能使線程退出。那么線程池又如何殺掉這個線程呢？
??沒錯，就是getTask方法。只有當getTask返回null的時候才能跳出While循環(huán)，run方法運行完畢，那么線程自然而然就死亡了。getTask方法如下所示：
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
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;
}
}
}
??可以看到，如果線程池狀態(tài)大于SHUTDOWN并且隊列空，返回null，從而結束循環(huán)。（線程死亡）
??或者狀態(tài)大于SHUTDOWN并且線程大于STOP（STOP一定大于SHUTDOWN，所以可以直接說線程大于STOP）返回null，從而結束循環(huán)。（線程死亡）
再往下可以看到如果超過了maximumPoolSize，返回null，從而結束循環(huán)。（線程死亡）
超過keepAliveTime時間，任務對列沒有數據而返回null。從而結束循環(huán)。（線程死亡）
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;保證了線程池至少留下corePoolSize個線程。
??在execute方法中，如果線程池拒絕添加任務，就會有一個鉤子方法來處理被拒絕的任務。
可以自己定義，也可以使用線城池中默認的拒接處理協(xié)議。
AbortPolicy :直接拋出RejectedExecutionException異常；
CallerRunsPolicy：誰調用的execute方法，誰就執(zhí)行這個任務；
DiscardPolicy:直接丟棄，什么也不做；
DiscardOldestPolicy：丟棄對列中間最老的任務，執(zhí)行新任務。
有什么問題或者建議，可以加入小密圈和我一起討論，或者在簡書留言，歡迎喜歡和打賞。
最后向大家安利一本我寫的關于Java并發(fā)的書籍：Java并發(fā)編程系統(tǒng)與模型，個人覺得寫得不錯，比較通俗易懂，非常適合初學者，百度閱讀可以下載電子書。
文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/67354.html
摘要：理解內存模型對多線程編程無疑是有好處的。干貨高級動畫高級動畫進階，矢量動畫。 這是最好的Android相關原創(chuàng)知識體系（100+篇） 知識體系從2016年開始構建，所有的文章都是圍繞著這個知識體系來寫，目前共收入了100多篇原創(chuàng)文章，其中有一部分未收入的文章在我的新書《Android進階之光》中。最重要的是，這個知識體系仍舊在成長中。 Android 下拉刷新庫，這一個就夠了！ 新鮮出...
摘要：最后，我們會通過對源代碼的剖析深入了解線程池的運行過程和具體設計，真正達到知其然而知其所以然的水平。創(chuàng)建線程池既然線程池是一個類，那么最直接的使用方法一定是一個類的對象，例如。單線程線程池單線程線程 我們一般不會選擇直接使用線程類Thread進行多線程編程，而是使用更方便的線程池來進行任務的調度和管理。線程池就像共享單車，我們只要在我們有需要的時候去獲取就可以了。甚至可以說線程池更棒，...
摘要：具體調用鏈路如圖函數主要是解析啟動參數，并過濾選項傳給引擎。查閱文檔之后發(fā)現(xiàn)，通過指定參數可以設置線程池大小。原來的字節(jié)碼編譯優(yōu)化還有都是通過多線程完成又繼續(xù)深入調查，發(fā)現(xiàn)環(huán)境變量會影響的線程池大小。執(zhí)行過程如下調用執(zhí)行。 作者：正龍 （滬江Web前端開發(fā)工程師）本文原創(chuàng)，轉載請注明作者及出處。 隨著Node.js的普及，越來越多的開發(fā)者使用Node.js來搭建環(huán)境，也有很多公司開始把...
閱讀 3717·2021-10-11 10:59
閱讀 1301·2019-08-30 15:44
閱讀 3479·2019-08-29 16:39
閱讀 2888·2019-08-29 16:29
閱讀 1800·2019-08-29 15:24
閱讀 808·2019-08-29 15:05
閱讀 1264·2019-08-29 12:34
閱讀 2302·2019-08-29 12:19
