在什么情況下使用線程池？

1.單個任務處理的時間比較短

2.將需處理的任務的數量大

使用線程池的好處:

1. 降低資源消耗：　　　　　　通過重復利用已創建的線程降低線程創建和銷毀造成的消耗。

2. 提高響應速度：　　　　　　當任務到達時，任務可以不需要等到線程創建就能立即執行。

3. 提高線程的可管理性：　　　線程是稀缺資源，如果無限制的創建。不僅僅會降低系統的穩定性，使用線程池可以統一分配，調優和監控。但是要做到合理的利用線程池。必須對于其實現原理了如指掌。

一個線程池包括以下四個基本組成部分：

1、線程池管理器(ThreadPool)：用于創建并管理線程池，包括 創建線程池，銷毀線程池，添加新任務；

2、工作線程(PoolWorker)：線程池中線程，在沒有任務時處于等待狀態，可以循環的執行任務；

3、任務接口(Task)：每個任務必須實現的接口，以供工作線程調度任務的執行，它主要規定了任務的入口，任務執行完后的收尾工作，任務的執行狀態等；

4、任務隊列(taskQueue)：用于存放沒有處理的任務。提供一種緩沖機制。

在JDK1.6中研究ThreadPoolExecutor類:

volatile intrunState;static final int RUNNING = 0;static final int SHUTDOWN = 1;static final int STOP = 2;static final int TERMINATED = 3;

runState表示當前線程池的狀態，它是一個volatile變量用來保證線程之間的可見性；

當創建線程池后，初始時，線程池處于RUNNING狀態；

如果調用了shutdown()方法，則線程池處于SHUTDOWN狀態，此時線程池不能夠接受新的任務，它會等待所有任務執行完畢；

如果調用了shutdownNow()方法，則線程池處于STOP狀態，此時線程池不能接受新的任務，并且會去嘗試終止正在執行的任務；

當線程池處于SHUTDOWN或STOP狀態，并且所有工作線程已經銷毀，任務緩存隊列已經清空或執行結束后，線程池被設置為TERMINATED狀態。

execute方法：

public voidexecute(Runnable command) {if (command == null)throw newNullPointerException();if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {if (runState == RUNNING &&workQueue.offer(command)) {if (runState != RUNNING || poolSize == 0)

ensureQueuedTaskHandled(command);

}else if (!addIfUnderMaximumPoolSize(command))

reject(command);//is shutdown or saturated

}

}

addIfUnderCorePoolSize方法檢查如果當前線程池的大小小于配置的核心線程數，說明還可以創建新線程，則啟動新的線程執行這個任務。

private booleanaddIfUnderCorePoolSize(Runnable firstTask) {

Thread t= null;final ReentrantLock mainLock = this.mainLock;

mainLock.lock();try{if (poolSize < corePoolSize && runState ==RUNNING)

t=addThread(firstTask);

}finally{

mainLock.unlock();

}return t != null;

}

addThread:

privateThread addThread(Runnable firstTask) {

Worker w= newWorker(firstTask);

Thread t=threadFactory.newThread(w);boolean workerStarted = false;if (t != null) {if (t.isAlive()) //precheck that t is startable

throw newIllegalThreadStateException();

w.thread=t;

workers.add(w);int nt = ++poolSize;if (nt >largestPoolSize)

largestPoolSize=nt;try{

t.start();

workerStarted= true;

}finally{if (!workerStarted)

workers.remove(w);

}

}returnt;

}

Worker，在ThreadPoolExecutor中的內部類

private final class Worker implementsRunnable {/*** The runLock is acquired and released surrounding each task

* execution. It mainly protects against interrupts that are

* intended to cancel the worker thread from instead

* interrupting the task being run.*/

private final ReentrantLock runLock = newReentrantLock();/*** Initial task to run before entering run loop. Possibly null.*/

privateRunnable firstTask;/*** Per thread completed task counter; accumulated

* into completedTaskCount upon termination.*/

volatile longcompletedTasks;/*** Thread this worker is running in. Acts as a final field,

* but cannot be set until thread is created.*/Thread thread;/*** Records that the thread assigned to this worker has actually

* executed our run() method. Such threads are the only ones

* that will be interrupted.*/

volatile boolean hasRun = false;

Worker(Runnable firstTask) {this.firstTask =firstTask;

}booleanisActive() {returnrunLock.isLocked();

}/*** Interrupts thread if not running a task.*/

voidinterruptIfIdle() {final ReentrantLock runLock = this.runLock;if(runLock.tryLock()) {try{if (hasRun && thread !=Thread.currentThread())

thread.interrupt();

}finally{

runLock.unlock();

}

}

}/*** Interrupts thread even if running a task.*/

voidinterruptNow() {if(hasRun)

thread.interrupt();

}/*** Runs a single task between before/after methods.*/

private voidrunTask(Runnable task) {final ReentrantLock runLock = this.runLock;

runLock.lock();try{/** If pool is stopping ensure thread is interrupted;

* if not, ensure thread is not interrupted. This requires

* a double-check of state in case the interrupt was

* cleared concurrently with a shutdownNow -- if so,

* the interrupt is re-enabled.*/

if ((runState >= STOP ||(Thread.interrupted()&& runState >= STOP)) &&hasRun)

thread.interrupt();/** Track execution state to ensure that afterExecute

* is called only if task completed or threw

* exception. Otherwise, the caught runtime exception

* will have been thrown by afterExecute itself, in

* which case we don't want to call it again.*/

boolean ran = false;

beforeExecute(thread, task);try{

task.run();

ran= true;

afterExecute(task,null);++completedTasks;

}catch(RuntimeException ex) {if (!ran)

afterExecute(task, ex);throwex;

}

}finally{

runLock.unlock();

}

}/*** Main run loop*/

public voidrun() {try{

hasRun= true;

Runnable task=firstTask;

firstTask= null;while (task != null || (task = getTask()) != null) {

runTask(task);

task= null;

}

}finally{

workerDone(this);

}

}

}

View Code

ensureQueuedTaskHandled:

判斷如果當前狀態不是RUNING，則當前任務不加入到任務隊列中，判斷如果狀態是停止，線程數小于允許的最大數，且任務隊列還不空，則加入一個新的工作線程到線程池來幫助處理還未處理完的任務。

private voidensureQueuedTaskHandled(Runnable command) {final ReentrantLock mainLock = this.mainLock;

mainLock.lock();boolean reject = false;

Thread t= null;try{int state =runState;if (state != RUNNING &&workQueue.remove(command))

reject= true;else if (state < STOP &&poolSize< Math.max(corePoolSize, 1) &&

!workQueue.isEmpty())

t= addThread(null);

}finally{

mainLock.unlock();

}if(reject)

reject(command);

}

voidreject(Runnable command) {

handler.rejectedExecution(command,this);

}

addIfUnderMaximumPoolSize:

addIfUnderMaximumPoolSize檢查如果線程池的大小小于配置的最大線程數，并且任務隊列已經滿了(就是execute方法試圖把當前線程加入任務隊列時不成功)，

說明現有線程已經不能支持當前的任務了，但線程池還有繼續擴充的空間，就可以創建一個新的線程來處理提交的任務。

private booleanaddIfUnderMaximumPoolSize(Runnable firstTask) {

Thread t= null;final ReentrantLock mainLock = this.mainLock;

mainLock.lock();try{if (poolSize < maximumPoolSize && runState ==RUNNING)

t=addThread(firstTask);

}finally{

mainLock.unlock();

}return t != null;

}

整個流程：

1、如果線程池的當前大小還沒有達到基本大小(poolSize < corePoolSize)，那么就新增加一個線程處理新提交的任務；

2、如果當前大小已經達到了基本大小，就將新提交的任務提交到阻塞隊列排隊，等候處理workQueue.offer(command)；

3、如果隊列容量已達上限，并且當前大小poolSize沒有達到maximumPoolSize，那么就新增線程來處理任務；

4、如果隊列已滿，并且當前線程數目也已經達到上限，那么意味著線程池的處理能力已經達到了極限，此時需要拒絕新增加的任務。至于如何拒絕處理新增的任務，取決于線程池的飽和策略RejectedExecutionHandler。

================================================

設置合適的線程池大小：

如果是CPU密集型的任務，那么良好的線程個數是實際CPU處理器的個數的1倍；

如果是I/O密集型的任務，那么良好的線程個數是實際CPU處理器個數的1.5倍到2倍

線程池中線程數量：

View Code

為什么+1,與CPU核數相等，表示滿核運行,+1的話表示在CPU上存在競爭，兩者的競爭力不一樣。稍微高一點負荷是不影響的。

==================================================================================

Java中提供了幾個Executors類的靜態方法：

public static ExecutorService newFixedThreadPool(intnThreads) {return newThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue());

}public staticExecutorService newSingleThreadExecutor() {return newFinalizableDelegatedExecutorService

(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue()));

}public staticExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue());

}

newFixedThreadPool創建的線程池corePoolSize和maximumPoolSize值是相等的，它使用的LinkedBlockingQueue；

newSingleThreadExecutor將corePoolSize和maximumPoolSize都設置為1，也使用的LinkedBlockingQueue；

newCachedThreadPool將corePoolSize設置為0，將maximumPoolSize設置為Integer.MAX_VALUE，使用的SynchronousQueue，也就是說來了任務就創建線程運行，當線程空閑超過60秒，就銷毀線程。

任務拒絕策略：

當線程池的任務緩存隊列已滿并且線程池中的線程數目達到maximumPoolSize，如果還有任務到來就會采取任務拒絕策略，通常有以下四種策略：

ThreadPoolExecutor.AbortPolicy:丟棄任務并拋出RejectedExecutionException異常。

ThreadPoolExecutor.DiscardPolicy：也是丟棄任務，但是不拋出異常。

ThreadPoolExecutor.DiscardOldestPolicy：丟棄隊列最前面的任務，然后重新嘗試執行任務(重復此過程)

ThreadPoolExecutor.CallerRunsPolicy：由調用線程處理該任務

demo:

importjava.util.concurrent.ArrayBlockingQueue;importjava.util.concurrent.ThreadPoolExecutor;importjava.util.concurrent.TimeUnit;public classMain {public static voidmain(String[] args) {

ThreadPoolExecutor executor= new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,new ArrayBlockingQueue(5));for(int i=0;i<15;i++){

MyTask myTask= newMyTask(i);

executor.execute(myTask);

System.out.println("線程池中線程數目："+executor.getPoolSize()+"，隊列中等待執行的任務數目："+executor.getQueue().size()+"，已執行玩別的任務數目："+executor.getCompletedTaskCount());

}

executor.shutdown();

}

}class MyTask implementsRunnable {private inttaskNum;public MyTask(intnum) {this.taskNum =num;

}

@Overridepublic voidrun() {

System.out.println("正在執行task "+taskNum);try{

Thread.currentThread().sleep(0);

}catch(InterruptedException e) {

e.printStackTrace();

}

System.out.println("task "+taskNum+"執行完畢");

}

}

線程池中線程數目：1，隊列中等待執行的任務數目：0，已執行玩別的任務數目：0線程池中線程數目：2，隊列中等待執行的任務數目：0，已執行玩別的任務數目：0線程池中線程數目：3，隊列中等待執行的任務數目：0，已執行玩別的任務數目：0正在執行task0線程池中線程數目：4，隊列中等待執行的任務數目：0，已執行玩別的任務數目：0正在執行task3正在執行task1task 3執行完畢

task 1執行完畢

線程池中線程數目：5，隊列中等待執行的任務數目：0，已執行玩別的任務數目：0task 0執行完畢

正在執行task5線程池中線程數目：5，隊列中等待執行的任務數目：1，已執行玩別的任務數目：2線程池中線程數目：5，隊列中等待執行的任務數目：1，已執行玩別的任務數目：3線程池中線程數目：5，隊列中等待執行的任務數目：2，已執行玩別的任務數目：3線程池中線程數目：5，隊列中等待執行的任務數目：3，已執行玩別的任務數目：3線程池中線程數目：5，隊列中等待執行的任務數目：4，已執行玩別的任務數目：3線程池中線程數目：5，隊列中等待執行的任務數目：5，已執行玩別的任務數目：3task 5執行完畢

正在執行task6task 6執行完畢

正在執行task7task 7執行完畢

正在執行task8task 8執行完畢

正在執行task9task 9執行完畢

正在執行task10task 10執行完畢

線程池中線程數目：6，隊列中等待執行的任務數目：0，已執行玩別的任務數目：9線程池中線程數目：6，隊列中等待執行的任務數目：1，已執行玩別的任務數目：9線程池中線程數目：6，隊列中等待執行的任務數目：2，已執行玩別的任務數目：9線程池中線程數目：6，隊列中等待執行的任務數目：3，已執行玩別的任務數目：9正在執行task12正在執行task14正在執行task13task 14執行完畢

task 13執行完畢

task 12執行完畢

正在執行task2task 2執行完畢

正在執行task4task 4執行完畢

正在執行task11task 11執行完畢

View Code