java執行緒池

合理使用執行緒池的好處

1. 降低資源消耗。通過重複利用已建立的執行緒降低執行緒建立和銷燬造成的消耗；
2. 提高執行速度。當任務到達時，任務可以不需要等待執行緒建立就能立即執行；
3. 提高執行緒的可管理性。執行緒是系統重要資源，執行緒池可以控制執行緒數量，避免獨自建立執行緒，引發系統問題。

核心類-ThreadPoolExecutor

java.uitl.concurrent.ThreadPoolExecutor類是執行緒池中最核心的一個類，因此如果要透徹地瞭解Java中的執行緒池，必須先了解這個類。

ThreadPoolExecutor類的具體實現原始碼：

public class ThreadPoolExecutor extends AbstractExecutorService {
    .....
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
        BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);
    ...
}
 

ThreadPoolExecutor繼承了AbstractExecutorService類，並提供了四個構造器。

核心構造器為

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
        BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);

主要引數：

• corePoolSize：核心池的大小
• maximumPoolSize：執行緒池最大執行緒數
• keepAliveTime：表示執行緒沒有任務執行時最多保持多久時間會終止
• unit：引數keepAliveTime的時間單位，有7種取值，在TimeUnit類中有7種靜態屬性
• workQueue：一個阻塞佇列，用來儲存等待執行的任務
• threadFactory：執行緒工廠，主要用來建立執行緒，設定執行緒名
• handler：表示當拒絕處理任務時的策略

unit的值

TimeUnit.DAYS;               //天
TimeUnit.HOURS;             //小時
TimeUnit.MINUTES;           //分鐘
TimeUnit.SECONDS;           //秒
TimeUnit.MILLISECONDS;      //毫秒
TimeUnit.MICROSECONDS;      //微妙
TimeUnit.NANOSECONDS;       //納秒
 

workQueue的值

ArrayBlockingQueue;
LinkedBlockingQueue;
SynchronousQueue;

handler的值-任務飽和處理策略

ThreadPoolExecutor.AbortPolicy:丟棄任務並丟擲RejectedExecutionException異常。 
ThreadPoolExecutor.DiscardPolicy：也是丟棄任務，但是不丟擲異常。 
ThreadPoolExecutor.DiscardOldestPolicy：丟棄佇列最前面的任務，然後重新嘗試執行任務（重複此過程）
ThreadPoolExecutor.CallerRunsPolicy：由呼叫執行緒處理該任務 

任務執行邏輯

ThreadPoolExecutor執行execute()方法來提交任務；

執行緒池節點到任務後有四種處理場景：

1. 判斷核心執行緒池是否有空閒執行緒。如果有空閒執行緒，直接讓任務搭車執行；如果核心執行緒已滿，進入下一個流程；
2. 判斷任務佇列是否已滿。如果沒有滿，就進入佇列，否則進入下一個流程；
3. 判斷整個執行緒池的執行緒數是否達到上限(BlockingQueue已滿)。如果沒有就開啟一個新的執行緒(需要獲取全域性鎖)，否則(危險狀態)交給飽和策略來處理；
4. RejectedExecutionHandler.rejectedExecution()方法來處理多出來的任務。

以上步驟的的設計思路是為了避免使用全域性鎖。

ThreadPoolExecutor執行execute()的流程

原始碼分析

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();
        //再次檢查執行緒池狀態，如果執行緒池處於非執行狀態（死掉了，或者執行緒進入方法後就關閉了  (because existing ones died since last checking) or that  the pool shut down since entry into this method）
        //就執行回滾策略(移除佇列元素)，然後  執行任務淘汰策略
        if (! isRunning(recheck) && remove(command))
            reject(command);
            // 如果當前空閒執行緒為0，則建立建一個新的執行緒
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 如果新增任務失敗，則執行淘汰策略
    else if (!addWorker(command, false))
        reject(command);
}

建立執行緒時加鎖原始碼

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);
}

任務執行：

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);
    }
}

工作執行緒：執行緒池建立執行緒時，會將執行緒封裝成工作執行緒Worker，Worker在執行完任務後，還會迴圈獲取工作佇列裡的任務來執行。

每個執行緒執行流程：

• 建立執行緒時，執行擁有的任務，然後取佇列中的任務執行。

配置執行緒池的引數

根據任務特性來分析：

• cpu密集型
• IO密集型
• 混合型

CPU密集型任務應配置儘可能小的執行緒，如配置Ncpu+1個執行緒的執行緒池。

由於IO密集型任務執行緒並不是一直在執行任務，則應配置儘可能多的執行緒，如2*Ncpu

可以通過Runtime.getRuntime().availableProcessors()方法獲得當前裝置的CPU個數。

優先順序不同的任務可以使用優先順序佇列PriorityBlockingQueue來處理。它可以讓優先順序高的任務先執行。

執行緒池的監控

可以根據api提供的方法來獲取執行緒池的一些資料來對執行緒池進行監控；
taskCount：執行緒池需要執行的任務數量。

• completedTaskCount：執行緒池在執行過程中已完成的任務數量，小於或等於taskCount。
• largestPoolSize：執行緒池裡曾經建立過的最大執行緒數量。通過這個資料可以知道執行緒池是否曾經滿過。如該數值等於執行緒池的最大大小，則表示執行緒池曾經滿過。
• getPoolSize：執行緒池的執行緒數量。如果執行緒池不銷燬的話，執行緒池裡的執行緒不會自動銷燬，所以這個大小隻增不減。
• getActiveCount：獲取活動的執行緒數。