其實java線程池的實現原理很簡單，說白了就是一個線程集合workerSet和一個Java阻塞隊列workQueue。當用戶向線程池提交一個任務(也就是線程)時，線程池會先將任務放入workQueue中。workerSet中的線程會不斷的從workQueue中獲取線程然后執行。當workQueue中沒有任務的時候，worker就會阻塞，直到隊列中有任務了就取出來繼續執行。

線程池的幾個主要參數的作用

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

corePoolSize: 規定線程池有幾個線程(worker)在運行。

maximumPoolSize: 當workQueue滿了,不能添加任務的時候，這個參數才會生效。規定線程池最多只能有多少個線程(worker)在執行。

keepAliveTime: 超出corePoolSize大小的那些線程的生存時間,這些線程如果長時間沒有執行任務并且超過了keepAliveTime設定的時間，就會消亡。

unit: 生存時間對于的單位

workQueue: 存放任務的隊列

threadFactory: 創建線程的工廠

handler: 當workQueue已經滿了，并且線程池線程數已經達到maximumPoolSize，將執行拒絕策略。

任務提交后的流程分析

用戶通過submit提交一個任務。線程池會執行如下流程:

判斷當前運行的worker數量是否超過corePoolSize,如果不超過corePoolSize。就創建一個worker直接執行該任務。—— 線程池最開始是沒有worker在運行的

如果正在運行的worker數量超過或者等于corePoolSize,那么就將該任務加入到workQueue隊列中去。

如果workQueue隊列滿了,也就是offer方法返回false的話，就檢查當前運行的worker數量是否小于maximumPoolSize,如果小于就創建一個worker直接執行該任務。

如果當前運行的worker數量是否大于等于maximumPoolSize，那么就執行RejectedExecutionHandler來拒絕這個任務的提交。

源碼解析

我們先來看一下ThreadPoolExecutor中的幾個關鍵屬性。

//這個屬性是用來存放 當前運行的worker數量以及線程池狀態的
//int是32位的，這里把int的高3位拿來充當線程池狀態的標志位,后29位拿來充當當前運行worker的數量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//存放任務的阻塞隊列
private final BlockingQueue<Runnable> workQueue;
//worker的集合,用set來存放
private final HashSet<Worker> workers = new HashSet<Worker>();
//歷史達到的worker數最大值
private int largestPoolSize;
//當隊列滿了并且worker的數量達到maxSize的時候,執行具體的拒絕策略
private volatile RejectedExecutionHandler handler;
//超出coreSize的worker的生存時間
private volatile long keepAliveTime;
//常駐worker的數量
private volatile int corePoolSize;
//最大worker的數量,一般當workQueue滿了才會用到這個參數
private volatile int maximumPoolSize;

1. 提交任務相關源碼

下面是execute方法的源碼

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        //workerCountOf(c)會獲取當前正在運行的worker數量
        if (workerCountOf(c) < corePoolSize) {
            //如果workerCount小于corePoolSize,就創建一個worker然后直接執行該任務
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //isRunning(c)是判斷線程池是否在運行中,如果線程池被關閉了就不會再接受任務
        //后面將任務加入到隊列中
        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);
        }
        //如果加入隊列失敗,就嘗試直接創建worker來執行任務
        else if (!addWorker(command, false))
            //如果創建worker失敗,就執行拒絕策略
            reject(command);
}

添加worker的方法addWorker源碼

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        //使用自旋+cas失敗重試來保證線程競爭問題
        for (;;) {
            //先獲取線程池的狀態
            int c = ctl.get();
            int rs = runStateOf(c);
            // 如果線程池是關閉的,或者workQueue隊列非空,就直接返回false,不做任何處理
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
            for (;;) {
                int wc = workerCountOf(c);
                //根據入參core 來判斷可以創建的worker數量是否達到上限,如果達到上限了就拒絕創建worker
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //沒有的話就嘗試修改ctl添加workerCount的值。這里用了cas操作,如果失敗了下一個循環會繼續重試,直到設置成功
                if (compareAndIncrementWorkerCount(c))
                    //如果設置成功了就跳出外層的那個for循環
                    break retry;
                //重讀一次ctl,判斷如果線程池的狀態改變了,會再重新循環一次
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
            }
        }
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            final ReentrantLock mainLock = this.mainLock;
            //創建一個worker,將提交上來的任務直接交給worker
            w = new Worker(firstTask);
            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)) {
                        //如果worker的線程已經啟動了,會拋出異常
                        if (t.isAlive()) 
                              throw new IllegalThreadStateException();
                        //添加新建的worker到線程池中
                        workers.add(w);
                        int s = workers.size();
                        //更新歷史worker數量的最大值
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        //設置新增標志位
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                //如果worker是新增的,就啟動該線程
                if (workerAdded) {
                    t.start();
                     //成功啟動了線程,設置對應的標志位
                    workerStarted = true;
                }
            }
        } finally {
            //如果啟動失敗了,會觸發執行相應的方法
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
}

2. Worker的結構

Worker是ThreadPoolExecutor內部定義的一個Java內部類。我們先看一下Worker的繼承關系

private final class Worker extends AbstractQueuedSynchronizer implements Runnable

它實現了Runnable接口,所以可以拿來當線程用。同時它還繼承了AbstractQueuedSynchronizer同步器類,主要用來實現一個不可重入的鎖。

一些屬性還有構造方法:

//運行的線程,前面addWorker方法中就是直接通過啟動這個線程來啟動這個worker
final Thread thread;
//當一個worker剛創建的時候,就先嘗試執行這個任務
Runnable firstTask;
//記錄完成任務的數量
volatile long completedTasks;
Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            //創建一個Thread,將自己設置給他,后面這個thread啟動的時候,也就是執行worker的run方法
            this.thread = getThreadFactory().newThread(this);
}

worker的run方法

public void run() {
            //這里調用了ThreadPoolExecutor的runWorker方法
            runWorker(this);
}

ThreadPoolExecutor的runWorker方法

final void runWorker(Worker w) {
        //獲取當前線程
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        //執行unlock方法,允許其他線程來中斷自己
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            //如果前面的firstTask有值,就直接執行這個任務
            //如果沒有具體的任務,就執行getTask()方法從隊列中獲取任務
            //這里會不斷執行循環體,除非線程中斷或者getTask()返回null才會跳出這個循環
            while (task != null || (task = getTask()) != null) {
                //執行任務前先鎖住,這里主要的作用就是給shutdown方法判斷worker是否在執行中的
                //shutdown方法里面會嘗試給這個線程加鎖,如果這個線程在執行,就不會中斷它
                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);
        }
}

下面來看一下getTask()方法，這里面涉及到keepAliveTime的使用，從這個方法我們可以看出先吃池是怎么讓超過corePoolSize的那部分worker銷毀的。

private Runnable getTask() {
        boolean timedOut = false; 
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            // 如果線程池已經關閉了,就直接返回null,
            //如果這里返回null,調用的那個worker就會跳出while循環,然后執行完銷毀線程
            //SHUTDOWN狀態表示執行了shutdown()方法
            //STOP表示執行了shutdownNow()方法
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
            //獲取當前正在運行中的worker數量
            int wc = workerCountOf(c);
            // 如果設置了核心worker也會超時或者當前正在運行的worker數量超過了corePoolSize,就要根據時間判斷是否要銷毀線程了
            //其實就是從隊列獲取任務的時候要不要設置超時間時間,如果超過這個時間隊列還沒有任務進來,就會返回null
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;            
            //如果上一次循環從隊列獲取到的未null,這時候timedOut就會為true了
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                //通過cas來設置WorkerCount,如果多個線程競爭,只有一個可以設置成功
                //最后如果沒設置成功,就進入下一次循環,說不定下一次worker的數量就沒有超過corePoolSize了,也就不用銷毀worker了
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
            try {
                //如果要設置超時時間,就設置一下咯
                //過了這個keepAliveTime時間還沒有任務進隊列就會返回null,那worker就會銷毀
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                //如果r為null,就設置timedOut為true
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
}

3. 添加Callable任務的實現源碼

public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
}

要添加一個有返回值的任務的實現也很簡單。其實就是對任務做了一層封裝，將其封裝成Future，然后提交給線程池執行，最后返回這個future。

這里的 newTaskFor(task) 方法會將其封裝成一個FutureTask類。

外部的線程拿到這個future，執行get()方法的時候,如果任務本身沒有執行完，執行線程就會被阻塞，直到任務執行完。

下面是FutureTask的get方法

public V get() throws InterruptedException, ExecutionException {
        int s = state;
        //判斷狀態,如果任務還沒執行完,就進入休眠,等待喚醒
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        //返回值
        return report(s);
}

FutureTask中通過一個state狀態來判斷任務是否完成。當run方法執行完后,會將state狀態置為完成，同時喚醒所有正在等待的線程。我們可以看一下FutureTask的run方法

public void run() {
        //判斷線程的狀態
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    //執行call方法
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                    //這個方法里面會設置返回內容,并且喚醒所以等待中的線程
                    set(result);
            }
        } finally {
            runner = null;
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
}

4. shutdown和shutdownNow方法的實現

shutdown方法會將線程池的狀態設置為SHUTDOWN,線程池進入這個狀態后,就拒絕再接受任務,然后會將剩余的任務全部執行完

public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //檢查是否可以關閉線程
            checkShutdownAccess();
            //設置線程池狀態
            advanceRunState(SHUTDOWN);
            //嘗試中斷worker
            interruptIdleWorkers();
             //預留方法,留給子類實現
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
}
private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
}
private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //遍歷所有的worker
            for (Worker w : workers) {
                Thread t = w.thread;
                //先嘗試調用w.tryLock(),如果獲取到鎖,就說明worker是空閑的,就可以直接中斷它
                //注意的是,worker自己本身實現了AQS同步框架,然后實現的類似鎖的功能
                //它實現的鎖是不可重入的,所以如果worker在執行任務的時候,會先進行加鎖,這里tryLock()就會返回false
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        w.unlock();
                    }
                }
                if (onlyOne)
                    break;
            }
        } finally {
            mainLock.unlock();
        }
}

shutdownNow做的比較絕，它先將線程池狀態設置為STOP，然后拒絕所有提交的任務。最后中斷左右正在運行中的worker,然后清空任務隊列。

public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            //檢測權限
            advanceRunState(STOP);
            //中斷所有的worker
            interruptWorkers();
            //清空任務隊列
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
}
private void interruptWorkers() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //遍歷所有worker，然后調用中斷方法
            for (Worker w : workers)
                w.interruptIfStarted();
        } finally {
            mainLock.unlock();
        }
}

以上就是關于“Java線程池原理解析”的介紹，大家如果想了解更多相關知識，不妨來關注一下本站的Java在線學習，里面的課程內容從入門到精通，細致全面，很適合沒有基礎的小伙伴學習，希望對大家能夠有所幫助哦。