Java多線程系列--“JUC線程池”03之 線程池原理(二)
線程池示例
在分析線程池之前,先看一個簡單的線程池示例。
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
public class ThreadPoolDemo1 {
public static void main(String[] args) {
// 創建一個可重用固定線程數的線程池
ExecutorService pool = Executors.newFixedThreadPool(2);
// 創建實現了Runnable接口對象,Thread對象當然也實現了Runnable接口
Thread ta = new MyThread();
Thread tb = new MyThread();
Thread tc = new MyThread();
Thread td = new MyThread();
Thread te = new MyThread();
// 將線程放入池中進行執行
pool.execute(ta);
pool.execute(tb);
pool.execute(tc);
pool.execute(td);
pool.execute(te);
// 關閉線程池
pool.shutdown();
}
}
class MyThread extends Thread {
@Override
public void run() {
System.out.println(Thread.currentThread().getName()+ " is running.");
}
}
運行結果:
pool-1-thread-1 is running.
pool-1-thread-1 is running.
pool-1-thread-1 is running.
pool-1-thread-1 is running.
pool -1-thread-2 is running.
示例中,包括了線程池的創建,將任務添加到線程池中,關閉線程池這3個主要的步驟。稍後,我們會從這3個方面來分析ThreadPoolExecutor。
線程池源碼分析
(一) 創建“線程池”
下面以newFixedThreadPool()介紹線程池的創建過程。
1. newFixedThreadPool()
newFixedThreadPool()在Executors.java中定義,源碼如下:
/**
* Creates a thread pool that reuses a fixed number of threads
* operating off a shared unbounded queue. At any point, at most
* {@code nThreads} threads will be active processing tasks.
* If additional tasks are submitted when all threads are active,
* they will wait in the queue until a thread is available.
* If any thread terminates due to a failure during execution
* prior to shutdown, a new one will take its place if needed to
* execute subsequent tasks. The threads in the pool will exist
* until it is explicitly {@link ExecutorService#shutdown shutdown}.
*
* @param nThreads the number of threads in the pool
* @return the newly created thread pool
* @throws IllegalArgumentException if {@code nThreads <= 0}
*/
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
說明:newFixedThreadPool(int nThreads)的作用是創建一個線程池,線程池的容量是nThreads。
newFixedThreadPool()在調用ThreadPoolExecutor()時,會傳遞一個LinkedBlockingQueue()對象,而LinkedBlockingQueue是單向鏈表實現的阻塞隊列。在線程池中,就是通過該阻塞隊列來實現"當線程池中任務數量超過允許的任務數量時,部分任務會阻塞等待"。
2. ThreadPoolExecutor()
ThreadPoolExecutor()在ThreadPoolExecutor.java中定義,源碼如下:
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
說明:該函數實際上是調用ThreadPoolExecutor的另外一個構造函數。該函數的源碼如下:
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.corePoolSize = corePoolSize;
// 最大池大小
this.maximumPoolSize = maximumPoolSize;
// 線程池的等待隊列
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
// 線程工廠對象
this.threadFactory = threadFactory;
// 拒絕策略的句柄
this.handler = handler;
}
說明:在ThreadPoolExecutor()的構造函數中,進行的是初始化工作。
corePoolSize, maximumPoolSize, unit, keepAliveTime和workQueue這些變量的值是已知的,它們都是通過newFixedThreadPool()傳遞而來。下面看看threadFactory和handler對象。
2.1 ThreadFactory
線程池中的ThreadFactory是一個線程工廠,線程池創建線程都是通過線程工廠對象(threadFactory)來完成的。
上面所說的threadFactory對象,是通過 Executors.defaultThreadFactory()返回的。Executors.java中的defaultThreadFactory()源碼如下:
public static ThreadFactory defaultThreadFactory() {
return new DefaultThreadFactory();
}
defaultThreadFactory()返回DefaultThreadFactory對象。Executors.java中的DefaultThreadFactory()源碼如下:
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
// 提供創建線程的API。
public Thread newThread(Runnable r) {
// 線程對應的任務是Runnable對象r
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
// 設為“非守護線程”
if (t.isDaemon())
t.setDaemon(false);
// 將優先級設為“Thread.NORM_PRIORITY”
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
說明:ThreadFactory的作用就是提供創建線程的功能的線程工廠。
它是通過newThread()提供創建線程功能的,下面簡單說說newThread()。newThread()創建的線程對應的任務是Runnable對象,它創建的線程都是“非守護線程”而且“線程優先級都是Thread.NORM_PRIORITY”。
2.2 RejectedExecutionHandler
handler是ThreadPoolExecutor中拒絕策略的處理句柄。所謂拒絕策略,是指將任務添加到線程池中時,線程池拒絕該任務所采取的相應策略。
線程池默認會采用的是defaultHandler策略,即AbortPolicy策略。在AbortPolicy策略中,線程池拒絕任務時會拋出異常!
defaultHandler的定義如下:
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
AbortPolicy的源碼如下:
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
// 拋出異常
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
(二) 添加任務到“線程池”
1. execute()
execute()定義在ThreadPoolExecutor.java中,源碼如下:
public void execute(Runnable command) {
// 如果任務為null,則拋出異常。
if (command == null)
throw new NullPointerException();
// 獲取ctl對應的int值。該int值保存了"線程池中任務的數量"和"線程池狀態"信息
int c = ctl.get();
// 當線程池中的任務數量 < "核心池大小"時,即線程池中少於corePoolSize個任務。
// 則通過addWorker(command, true)新建一個線程,並將任務(command)添加到該線程中;然後,啟動該線程從而執行任務。
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
// 當線程池中的任務數量 >= "核心池大小"時,
// 而且,"線程池處於允許狀態"時,則嘗試將任務添加到阻塞隊列中。
if (isRunning(c) && workQueue.offer(command)) {
// 再次確認“線程池狀態”,若線程池異常終止了,則刪除任務;然後通過reject()執行相應的拒絕策略的內容。
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
// 否則,如果"線程池中任務數量"為0,則通過addWorker(null, false)嘗試新建一個線程,新建線程對應的任務為null。
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 通過addWorker(command, false)新建一個線程,並將任務(command)添加到該線程中;然後,啟動該線程從而執行任務。
// 如果addWorker(command, false)執行失敗,則通過reject()執行相應的拒絕策略的內容。
else if (!addWorker(command, false))
reject(command);
}
說明:execute()的作用是將任務添加到線程池中執行。它會分為3種情況進行處理:
情況1 -- 如果"線程池中任務數量" < "核心池大小"時,即線程池中少於corePoolSize個任務;此時就新建一個線程,並將該任務添加到線程中進行執行。
情況2 -- 如果"線程池中任務數量" >= "核心池大小",並且"線程池是允許狀態";此時,則將任務添加到阻塞隊列中阻塞等待。在該情況下,會再次確認"線程池的狀態",如果"第2次讀到的線程池狀態"和"第1次讀到的線程池狀態"不同,則從阻塞隊列中刪除該任務。
情況3 -- 非以上兩種情況。在這種情況下,嘗試新建一個線程,並將該任務添加到線程中進行執行。如果執行失敗,則通過reject()拒絕該任務。
2. addWorker()
addWorker()的源碼如下:
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
// 更新"線程池狀態和計數"標記,即更新ctl。
for (;;) {
// 獲取ctl對應的int值。該int值保存了"線程池中任務的數量"和"線程池狀態"信息
int c = ctl.get();
// 獲取線程池狀態。
int rs = runStateOf(c);
// 有效性檢查
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
// 獲取線程池中任務的數量。
int wc = workerCountOf(c);
// 如果"線程池中任務的數量"超過限制,則返回false。
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 通過CAS函數將c的值+1。操作失敗的話,則退出循環。
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
// 檢查"線程池狀態",如果與之前的狀態不同,則從retry重新開始。
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 {
final ReentrantLock mainLock = this.mainLock;
// 新建Worker,並且指定firstTask為Worker的第一個任務。
w = new Worker(firstTask);
// 獲取Worker對應的線程。
final Thread t = w.thread;
if (t != null) {
// 獲取鎖
mainLock.lock();
try {
int c = ctl.get();
int rs = runStateOf(c);
// 再次確認"線程池狀態"
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// 將Worker對象(w)添加到"線程池的Worker集合(workers)"中
workers.add(w);
// 更新largestPoolSize
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;
}
說明:
addWorker(Runnable firstTask, boolean core) 的作用是將任務(firstTask)添加到線程池中,並啟動該任務。
core為true的話,則以corePoolSize為界限,若"線程池中已有任務數量>=corePoolSize",則返回false;core為false的話,則以maximumPoolSize為界限,若"線程池中已有任務數量>=maximumPoolSize",則返回false。
addWorker()會先通過for循環不斷嘗試更新ctl狀態,ctl記錄了"線程池中任務數量和線程池狀態"。
更新成功之後,再通過try模塊來將任務添加到線程池中,並啟動任務所在的線程。
從addWorker()中,我們能清晰的發現:線程池在添加任務時,會創建任務對應的Worker對象;而一個Workder對象包含一個Thread對象。(01) 通過將Worker對象添加到"線程的workers集合"中,從而實現將任務添加到線程池中。 (02) 通過啟動Worker對應的Thread線程,則執行該任務。
3. submit()
補充說明一點,submit()實際上也是通過調用execute()實現的,源碼如下:
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
(三) 關閉“線程池”
shutdown()的源碼如下:
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
// 獲取鎖
mainLock.lock();
try {
// 檢查終止線程池的“線程”是否有權限。
checkShutdownAccess();
// 設置線程池的狀態為關閉狀態。
advanceRunState(SHUTDOWN);
// 中斷線程池中空閑的線程。
interruptIdleWorkers();
// 鉤子函數,在ThreadPoolExecutor中沒有任何動作。
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
// 釋放鎖
mainLock.unlock();
}
// 嘗試終止線程池
tryTerminate();
}
說明:shutdown()的作用是關閉線程池。
參考:
http://www.cnblogs.com/skywang12345/p/3509954.html
Java多線程系列--“JUC線程池”03之 線程池原理(二)