線程池源碼解析
線程池的概念
系統啟動一個新線程的成本是比較高的，因為它涉及與操作系統的交互。使用線程池可以很好的提高性能，尤其是程序中需要創建大量生存期很短暫的線程。
使用線程池的優勢
- 降低資源消耗。通過重復利用已創建的線程降低線程創建和銷毀的開銷。
- 提高響應速度。當任務到達時，任務可以不需要等到線程創建就能立即執行。
- 提高線程的可管理性。線程是稀缺資源，如果無限制地創建，不僅會消耗系統資源，還會降低系統的穩定性，使用線程池可以進行統一分配、調優和監控。
線程池的流程
- 如果當前woker數量小於corePoolSize，則新建一個woker並把當前任務分配給該woker線程，成功則返回。
- 如果第一步失敗，則嘗試把任務放入阻塞隊列，如果成功則返回。
- 如果第二步失敗，則判斷如果當前woker數量小於maximumPoolSize，則新建一個woker並把當前任務分配給該woker線程，成功則返回。
- 如果第三步失敗，則調用拒絕策略處理該任務。
核心源碼實現
線程池的本質就是使用了一個線程安全的工作隊列連接工作者線程和客戶端線程，客戶端線程將任務放入工作隊列后便返回，而工作者線程則不斷地從工作隊列上取出工作並執行。
當工作隊列為空時，所有的工作者線程均等待在工作隊列上，當有客戶端提交了一個任務之后會通知任意一個工作者線程，隨着大量的任務被提交，更多的工作者線程會被喚醒。
注意的是，核心線程在完成任務后不會被銷毀，而是在循環getTask()時被阻塞隊列阻塞住。只有當線程數大於了核心線程數的那些普通線程會被銷毀。
構造器參數：
corePoolSize：線程池中的核心線程數，當提交一個任務時，線程池創建一個新線程執行任務，直到當前線程數等於corePoolSize,即使有其他空閑線程能夠執行新來的任務,也會繼續創建線程；如果當前線程數為corePoolSize，繼續提交的任務被保存到阻塞隊列中，等待被執行；
maximumPoolSize：線程池中允許的最大線程數。如果當前阻塞隊列滿了，且繼續提交任務，則創建新的線程執行任務，前提是當前線程數小於maximumPoolSize；當阻塞隊列是無界隊列,則maximumPoolSize則不起作用,因為無法提交至核心線程池的線程會一直持續地放入workQueue.
keepAliveTime：線程存活時間(當線程池允許線程超時且運行中的線程數量超過corePoolSize時，會按照此變量設置時間關閉線程)
TimeUnit：keepAliveTime的單位
BlockingQueue<Runnable> workQueue：緩沖隊列，來不及執行的任務存放的阻塞隊列
RejectedExecutionHandler handler：拒絕處理任務類（默認：AbortPolicy 會拋異常）
AbortPolicy：直接拋出異常，默認策略；
CallerRunsPolicy：用調用者所在的線程來執行任務；
DiscardOldestPolicy：丟棄阻塞隊列中靠最前的任務，並執行當前任務；
DiscardPolicy：直接丟棄任務；
當然也可以根據應用場景實現RejectedExecutionHandler接口，自定義飽和策略，如記錄日志或持久化存儲不能處理的任務。
threadFactory：創建線程的工廠，通過自定義的線程工廠可以給每個新建的線程設置一個具有識別度的線程名。默認為DefaultThreadFactory
———————————————————————————————————————————————————————————————————————————————
//構造器
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
ThreadPoolExecutor.java
private final BlockingQueue<Runnable> workQueue;//緩沖隊列
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//原子類用來計數
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//1 當前運行的線程數量小於核心線程數量，直接將任務加入worker啟動運行。
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
//如果失敗，則獲取最新的線程池數據
c = ctl.get();
}
/*2 運行線程數量大於核心線程數量時，上面的if分支針對大於corePoolSize，並且緩存隊列加入任務操作成功的情況。
運行中並且將任務加入緩沖隊列成功，正常來說這樣已經完成了處理邏輯。
但是為了保險起見，增加了狀態出現異常的確認判斷，如果狀態出現異常會繼續remove操作，如果執行true，則按照拒絕處理策略駁回任務；*/
//運行線程數量大於核心線程數量時，如果線程池仍在運行，則把任務放到阻塞隊列中等待執行。
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//當任務成功放入隊列時，如果recheck發現線程池已經不再運行了則從隊列中把任務刪除
if (! isRunning(recheck) && remove(command))
//刪除成功以后，會調用構造參數傳入的拒絕策略。
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
/*3 這里針對運行線程數量超過了corePoolSize，並且緩存隊列也已經放滿的情況。
注意第二個參數是false，可以在下面addWorker方法看到，就是針對線程池最大線程數量maximumPoolSize的判斷。*/
else if (!addWorker(command, false))
//如果基於maximumPoolSize新建woker失敗，此時是線程池中線程數已達到上限，隊列已滿，則調用構造參數中傳入的拒絕策略
reject(command);
}
addWorker方法
private boolean addWorker(Runnable firstTask, boolean core) {
// CAS+死循環實現的關於線程池狀態，線程數量的校驗與更新邏輯
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()))
return 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 {
//把指定任務作為參數新建一個worker線程
w = new Worker(firstTask);
//變量t就是代表woker線程
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();
}
// 線程啟動，執行任務（Worker.thread(firstTask).start()）
// 找到Worker的實現的run方法
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
//如果woker啟動失敗，則進行一些善后工作，比如說修改當前woker數量等等
addWorkerFailed(w);
}
return workerStarted;
}
Worker類
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
//Worker類run方法中調用了runWorker方法
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) {
}
}
}
}
runWorker方法
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
//task就是Woker構造函數入參指定的任務，即用戶提交的任務
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 先執行firstTask，再從workerQueue中取task（getTask()),一直循環執行
//我們都知道構造參數設置的時間代表了線程池中的線程，即woker線程的存活時間，如果到期則回收woker線程
//這個邏輯的實現就在getTask中。
//來不及執行的任務，線程池會放入一個阻塞隊列，getTask方法就是去阻塞隊列中取任務，用戶設置的存活時間，就是
//從這個阻塞隊列中取任務等待的最大時間，如果getTask返回null，意思就是woker等待了指定時間仍然沒有
//取到任務，此時就會跳過循環體，進入woker線程的銷毀邏輯。
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(); //運行傳入的線程的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 {
//當指定任務執行完成，阻塞隊列中也取不到可執行任務時，會進入這里，做一些善后工作
//比如在corePoolSize跟maximumPoolSize之間的woker會進行回收
processWorkerExit(w, completedAbruptly);
}
}
getTask()方法
從阻塞任務隊列中取任務，如果設置了allowCoreThreadTimeOut(true) 或者當前運行的任務數大於設置的核心線程數，那么timed =true 。此時將使用workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)從任務隊列中取任務，而如果沒有設置，那么使用workQueue.take() 取任務，對於阻塞隊列，poll(long timeout, TimeUnit unit) 將會在規定的時間內去任務，如果沒取到就返回null。take()會一直阻塞，等待任務的添加。
到此相信我們都能夠理解為什么我們的線程池能夠一直等待任務的執行而不被銷毀了，其實也就是進入了阻塞狀態而已。
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?注意，此處決定是否銷毀線程
//條件是開啟了 allowCoreThreadTimeOut，或者總線程數大於了核心線程數
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;
}
}
}
線程池的使用
Java通過Executors提供四種線程池，分別為：
- newCachedThreadPool創建一個可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閑線程，若無可回收，則新建線程。
- newFixedThreadPool 創建一個定長線程池，可控制線程最大並發數，超出的線程會在隊列中等待。
- newScheduledThreadPool 創建一個定長線程池，支持定時及周期性任務執行。
- newSingleThreadExecutor 創建一個單線程化的線程池，它只會用唯一的工作線程來執行任務，保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行。
這四種線程池底層都是依賴ThreadPoolExecutor的構造器生成的。
//定長線程池
//corePoolSize跟maximumPoolSize值一樣，同時傳入一個無界阻塞隊列
//該線程池的線程會維持在指定線程數，不會進行回收
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
//緩存線程池
//這個線程池corePoolSize為0，maximumPoolSize為Integer.MAX_VALUE
//意思也就是說來一個任務就創建一個woker，回收時間是60s
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
//調度線程池
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
//單線程池
//線程池中只有一個線程進行任務執行，其他的都放入阻塞隊列
//外面包裝的FinalizableDelegatedExecutorService類實現了finalize方法，在JVM垃圾回收的時候會關閉線程池
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
使用例子：
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class Test {
public static void main(String[] args) {
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();//創建一個可緩存線程池
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2);//創建一個定長線程池，可控制線程最大並發數，超出的線程會在隊列中等待
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);//創建一個定長線程池，支持定時及周期性任務執行
ExecutorService single = Executors.newSingleThreadExecutor();//創建一個單線程化的線程池，它只會用唯一的工作線程來執行任務，保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行
for(int i=0;i<10;i++){
final int index = i;
try {
Thread.sleep(index*500);
} catch (InterruptedException e) {
e.printStackTrace();
}
cachedThreadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(index);
}
});
fixedThreadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(index);
}
});
//定期3s執行
scheduledExecutorService.schedule(new Runnable() {
@Override
public void run() {
System.out.println("delay 3s");
}
}, 3, TimeUnit.SECONDS);
single.execute(new Runnable() {
@Override
public void run() {
System.out.println(index);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
}
可以使用兩個方法向線程池提交任務，分別為execute()和submit()方法。
submit()方法用於提交需要返回值的任務。線程池會返回一個future類型的對象，通過這個
future對象可以判斷任務是否執行成功，並且可以通過future的get()方法來獲取返回值，get()方
法會阻塞當前線程直到任務完成，而使用get（long timeout，TimeUnit unit）方法則會阻塞當前線
程一段時間后立即返回，這時候有可能任務沒有執行完。
Future<Object> future = executor.submit(harReturnValuetask);
try {
Object s = future.get();
} catch (InterruptedException e) {
// 處理中斷異常
} catch (ExecutionException e) {
// 處理無法執行任務異常
} finally {
// 關閉線程池
executor.shutdown();
}
線程池設計大小
任務一般分為：CPU密集型、IO密集型
線程等待時間所占比例越高，需要越多線程。線程CPU時間所占比例越高，需要越少線程。
估算公式：
- 最佳線程數目 = （（線程等待時間+線程CPU時間）/線程CPU時間 ）* CPU數目
- 最佳線程數目 = （線程等待時間與線程CPU時間之比 + 1）* CPU數目
一般經驗上設置大小
1、CPU密集型
盡量使用較小的線程池，一般Cpu核心數+1
因為CPU密集型任務CPU的使用率很高，若開過多的線程，只能增加線程上下文的切換次數，帶來額外的開銷
2、IO密集型
方法一：可以使用較大的線程池，一般CPU核心數 * 2
IO密集型CPU使用率不高，可以讓CPU等待IO的時候處理別的任務，充分利用cpu時間