線程在執(zhí)行任務(wù)時芯侥，正常的情況是這樣的：
Thread t=new Thread(new Runnable() {
@Override
public void run() {
// TODO Auto-generated method stub
}
});
t.start();
Thread 在初始化的時候傳入一個Runnable,以后就沒有機會再傳入一個Runable了澎埠。那么祈噪，woker作為一個已經(jīng)啟動的線程雕擂。是如何不斷獲取Runnable的呢踊东？
這個時候可以使用一個包裝器贰健，將線程包裝起來胞四，在Run方法內(nèi)部獲取任務(wù)。
public final class Worker implements Runnable {
Thread thread = null;
Runnable task;
private BlockingQueue<Runnable> queues;
public Worker(Runnable task, BlockingQueue<Runnable> 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<Runnable> queues=new ArrayBlockingQueue<Runnable>(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ā)編程中經(jīng)典的Cyclic Work Distribution模式。
??那么脊另，這種實現(xiàn)的線程池性能如何呢导狡？
??由于其任務(wù)隊列使用的是阻塞隊列，在隊列內(nèi)部是自旋的偎痛。Reeteenlok是改進的CLH隊列烘豌。自旋鎖會耗費一定CPU的資源，在擁有大量任務(wù)執(zhí)行下的情況下比較有效看彼。而且廊佩，線程池中的線程并沒有睡眠，而是進入了自旋狀態(tài)靖榕。
CPU的線程與關(guān)系
如果是不支持超線程的CPU标锄，在同一時刻的確只能處理2個線程，但是并不意味著雙核的CPU只能處理兩個線程茁计，它可以通過切換上下文來執(zhí)行多個線程料皇。比如我只有一個大腦谓松，但是我要處理5個人提交的任務(wù)，我可以處理完A的事情后践剂，把事情的中間結(jié)果保存下鬼譬，然后再處理B的，然后再讀取A的中間結(jié)果逊脯，處理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)方法內(nèi)部是這樣的：
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一個合并類型的值军洼。將當(dāng)前線程數(shù)和線程池狀態(tài)通過數(shù)學(xué)運算合并到了一個值巩螃。具體是如何合并的可以參看一下源碼，這里就不敘述了匕争。繼續(xù)向下走：
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
可以看到避乏，如果當(dāng)前線程數(shù)量小于了核心線程數(shù)量corePoolSize，就直接增加線程處理任務(wù)甘桑。與隊列沒有關(guān)系拍皮。但是緊接著又檢查了一遍狀態(tài)，因為在這個過程中跑杭，別的線程也可能在添加任務(wù)春缕。繼續(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)的，就把線程添加到任務(wù)隊列艘蹋。workQueue是構(gòu)造函數(shù)傳遞過來的锄贼，可以是有界隊列，也可以是無界隊列女阀≌纾可以看出來，隊列如果是無界的浸策，直接往隊列里面添加任務(wù)冯键，這個時候，線程池中的線程也不會增加庸汗，一直會等于核心線程數(shù)惫确。
??如果隊列是有界的，就嘗試直接新增線程處理任務(wù)蚯舱，如果添加任務(wù)失敗改化，就調(diào)用reject方法來處理添加失敗的任務(wù)：
else if (!addWorker(command, false))
reject(command);
來看看addWorker是如何實現(xiàn)的，邏輯流程已經(jīng)直接在注釋中說明了枉昏。
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//如果狀態(tài)大于SHUTDOWN陈肛，不再接受新的任務(wù)，直接返回
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
/**根據(jù)core來判斷兄裂，如果當(dāng)前線程數(shù)量大于corePoolSize或者最大線程數(shù)句旱，直接返回阳藻。添加任務(wù)失敗。
**如果隊列是有界的或者任務(wù)添加到隊列失敗(參數(shù)core是false)谈撒，那么就會新開一個線程處理業(yè)務(wù)腥泥，但如果線程已經(jīng)大于了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)建失敗的情況下會調(diào)用addWorkerFailed方法，從而將減少實際線程數(shù)立宜。
addWorker中for循環(huán)的意義
在addWorker中有這么一段代碼冒萄，表示為當(dāng)前線程數(shù)加1：
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
由于多線程可能同時操作臊岸。expect值可能會變化橙数。僅僅一次的操作compareAndIncrementWorkerCount可能一次并不會成功，而且帅戒，一個線程在執(zhí)行addWork的過程中間灯帮，另外一個線程假設(shè)直接shotdown這個線程池。for循環(huán)的存在可以保證狀態(tài)一定是一致的逻住。
任務(wù)的執(zhí)行
在Worker中間實際上是調(diào)用的runWorker方法來執(zhí)行的具體業(yè)務(wù)：
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)不斷的從隊列中取出任務(wù)執(zhí)行钟哥。如果task==null 并且getTask()等于null的話，那么就會跳出循環(huán)瞎访，進入到processWorkerExit腻贰，run方法執(zhí)行完畢以后，這個線程也被銷毀了扒秸。但是為什么在各自的線程執(zhí)行播演，為什么還需要加鎖呢？答案是因為要線程池需要判斷這個線程是否在執(zhí)行任務(wù)伴奥。在interruptIdleWorkers方法中写烤，要中斷那寫目前空閑的線程，通過當(dāng)前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()可以獲取到鎖暂衡，那么就意味著當(dāng)前的 Woker并沒有處理任務(wù)（沒有進入到循環(huán)里面或者被getTask方法所阻塞，無法獲取鎖）崖瞭。
Work之所以繼承AbstractQueuedSynchronizer古徒，而不去使用ReentrantLock。是因為ReentrantLock是可重入鎖读恃，在調(diào)用lock方法獲取鎖之后隧膘，再調(diào)用tryLock()還是會返回true代态。
public static void main(String[] args) {
ReentrantLock lock = new ReentrantLock();
lock.lock();
System.out.println(lock.tryLock());
}
輸出結(jié)果是true，所以使用ReentrantLock則難以判斷當(dāng)前Worker是否在執(zhí)行任務(wù)疹吃。
線程超時allowCoreThreadTimeOut蹦疑、keepAliveTime以及線程死亡
在上面的interruptIdleWorkers方法中，線程被中斷萨驶。普通的線程被中斷會導(dǎo)致線程繼續(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í)行，當(dāng)run方法執(zhí)行完畢谤草，那么線程也就終結(jié)死亡跟束。但是對于ReentrantLock和AbstractQueuedSynchronizer這種自旋+CAS實現(xiàn)的“邏輯鎖”，是不起作用的丑孩。
而且runWork本身也是While循環(huán)冀宴，靠中斷是無法退出循環(huán)的。
但是在ThreadPoolExecutor的構(gòu)造函數(shù)中温学，有一個允許設(shè)置線程超時allowCoreThreadTimeOut參數(shù)的方法略贮。如果允許超時，多于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并且調(diào)用interruptIdleWorkers方法并不能使線程退出。那么線程池又如何殺掉這個線程呢箩帚？
??沒錯真友，就是getTask方法。只有當(dāng)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，從而結(jié)束循環(huán)鹅搪。（線程死亡）
或者狀態(tài)大于SHUTDOWN并且線程大于STOP（STOP一定大于SHUTDOWN站绪，所以可以直接說線程大于STOP）返回null，從而結(jié)束循環(huán)丽柿。（線程死亡）
再往下可以看到如果超過了maximumPoolSize恢准，返回null魂挂，從而結(jié)束循環(huán)。（線程死亡）
超過keepAliveTime時間馁筐，任務(wù)對列沒有數(shù)據(jù)而返回null涂召。從而結(jié)束循環(huán)。（線程死亡）
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;保證了線程池至少留下corePoolSize個線程敏沉。
預(yù)定義的拒接處理協(xié)議
在execute方法中果正，如果線程池拒絕添加任務(wù)，就會有一個鉤子方法來處理被拒絕的任務(wù)盟迟。
可以自己定義秋泳，也可以使用線城池中默認的拒接處理協(xié)議。
AbortPolicy :直接拋出RejectedExecutionException異常攒菠；
CallerRunsPolicy：誰調(diào)用的execute方法迫皱，誰就執(zhí)行這個任務(wù)；
DiscardPolicy:直接丟棄要尔，什么也不做舍杜；
DiscardOldestPolicy：丟棄對列中間最老的任務(wù)新娜，執(zhí)行新任務(wù)赵辕。
有什么問題或者建議，可以加入小密圈和我一起討論概龄，或者在簡書留言还惠，歡迎喜歡和打賞。
本書節(jié)選自：Java并發(fā)編程系統(tǒng)與模型私杜，個人覺得寫得不錯蚕键，比較通俗易懂，非常適合初學(xué)者衰粹，百度閱讀可以下載電子書锣光。
