@(Executors)[newFixedThreadPool]
[TOC]
java線程池
在面向?qū)ο缶幊讨腥战茫瑒?chuàng)建和銷毀對象是很費時間的庭呜，因為創(chuàng)建一個對象要獲取內(nèi)存資源或者其它更多資源。在Java中更是如此钢颂，虛擬機將試圖跟蹤每一個對象钞它，以便能夠在對象銷毀后進行垃圾回收拜银。所以提高服務(wù)程序效率的一個手段就是盡可能減少創(chuàng)建和銷毀對象的次數(shù)殊鞭，特別是一些很耗資源的對象創(chuàng)建和銷毀。如何利用已有對象來服務(wù)就是一個需要解決的關(guān)鍵問題尼桶，其實這就是一些"池化資源"技術(shù)產(chǎn)生的原因操灿。
newFixedPool作用
創(chuàng)建一個固定線程數(shù)的線程池，在任何時候最多只有nThreads個線程被創(chuàng)建泵督。如果在所有線程都處于活動狀態(tài)時趾盐，有其他任務(wù)提交，他們將等待隊列中直到線程可用小腊。如果任何線程由于執(zhí)行過程中的故障而終止救鲤，將會有一個新線程將取代這個線程執(zhí)行后續(xù)任務(wù)。
構(gòu)造方法
newFixedPool擁有兩個構(gòu)造方法：
參數(shù)為nThreads的構(gòu)造方法:
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
構(gòu)造方法中的nThreads代表固定線程池中的線程數(shù)秩冈，當使用此構(gòu)造方法創(chuàng)建線程池后本缠，就會創(chuàng)建nThreads個線程在線程池內(nèi)。
參數(shù)為： nThreads,ThreadFactory的構(gòu)造方法:
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory);
}
ThreadPoolExecutor構(gòu)造方法
newFixedThreadPool(int nThreads) 使用的構(gòu)造方法:
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
此ThreadPoolExecutor構(gòu)造方法的參數(shù)有5個：
corePoolSize：線程池中所保存的線程數(shù)入问，包括空閑線程丹锹，newFixedThreadPool中傳入nThreads
maximumPoolSize：池中允許的最大線程數(shù)，newFixedThreadPool中傳入nThreads芬失，使線程池的最大線程數(shù)與線程池中保存的線程數(shù)一致楣黍，使保證線程池的線程數(shù)是固定的
TimeUnit：參數(shù)的時間單位
keepAliveTime：當線程數(shù)大于corePoolSize時，此為終止前多余的空閑線程等待新任務(wù)的最長時間
LinkedBlockingQueue workQueuqe：當任務(wù)被執(zhí)行前棱烂，放到此阻塞隊列中租漂，任務(wù)將被放在阻塞隊列中直到使用execute方法提交Runnable任務(wù),不同的線程池主要是這個參數(shù)的不通，比如scheduledThreadPoolExecutor這邊使用的就是DelayedWorkQueue這個隊列
**注意：
* The queue used for holding tasks and handing off to worker
* threads. We do not require that workQueue.poll() returning
* null necessarily means that workQueue.isEmpty(), so rely
* solely on isEmpty to see if the queue is empty (which we must
* do for example when deciding whether to transition from
* SHUTDOWN to TIDYING). This accommodates special-purpose
* queues such as DelayQueues for which poll() is allowed to
* return null even if it may later return non-null when delays
* expire.
*/
private final BlockingQueue<Runnable> workQueue;
newFixedThreadPool(in nThreads,ThreadFactory threadFactory)使用的構(gòu)造方法:
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory);
}
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler); //創(chuàng)建默認線程工廠，并設(shè)置駁回的異常處理
}
創(chuàng)建ThreadPoolExecutor時傳入Executors.defaultThreadFactory()這個默認線程工廠實例窜锯。
在代碼中調(diào)用ThreadPoolExecutor中的內(nèi)部類DefaultThreadFactory工廠张肾，在包中的注釋是這么寫的：threadFactory the factory to use when the executor creates a new thread 。這是用來使用executor創(chuàng)建新線程的工廠锚扎。
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-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
調(diào)用DefaultThreadFactory的構(gòu)造方法初始化線程組和生成線程前綴吞瞪。在創(chuàng)建完線程工廠后，需要使用線程工廠來創(chuàng)建線程并啟動驾孔。
使用線程池中的線程執(zhí)行Runnable任務(wù)
當之前的初始化（corePoolSize等于maxiumPoolSize等于nThreads芍秆，并創(chuàng)建defaultThreadFactory）就可以運行定義好的線程了。
執(zhí)行線程的代碼如下：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
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);
}
其中最重要的方法是addWorker翠勉，其方法目的是創(chuàng)建Worker對象妖啥，并將Worker對象加入到HashSet<Worker> workers中。在加入workers之前对碌，先創(chuàng)建ReentrantLock排它鎖荆虱，將worker同步的加入到workers中。當worker成功被加入后朽们，啟動本次放入set中的線程怀读。
private boolean addWorker(Runnable firstTask, boolean core) {
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 {
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;
}
Worker對象的構(gòu)造方法：
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
通過線程工廠 來創(chuàng)建線程。
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
