麻豆小视频在线观看_中文黄色一级片_久久久成人精品_成片免费观看视频大全_午夜精品久久久久久久99热浪潮_成人一区二区三区四区

首頁 > 編程 > Java > 正文

Java線程池執(zhí)行原理分析

2019-11-11 07:46:08
字體:
來源:轉載
供稿:網(wǎng)友

學習java的同學注意了!!! 學習過程中遇到什么問題或者想獲取學習資源的話,歡迎加入Java學習交流群,群號碼:183993990  我們一起學Java!

本文將會圍繞線程池的生命周期,分析線程池執(zhí)行任務的過程。

線程池狀態(tài)

首先認識兩個貫穿線程池代碼的參數(shù):

runState:線程池運行狀態(tài)workerCount:工作線程的數(shù)量

線程池用一個32位的int來同時保存runState和workerCount,其中高3位是runState,其余29位是workerCount。代碼中會反復使用runStateOf和workerCountOf來獲取runState和workerCount。

PRivate final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));private static final int COUNT_BITS = Integer.SIZE - 3;private static final int CAPACITY   = (1 << COUNT_BITS) - 1;// 線程池狀態(tài)private static final int RUNNING    = -1 << COUNT_BITS;private static final int SHUTDOWN   =  0 << COUNT_BITS;private static final int STOP       =  1 << COUNT_BITS;private static final int TIDYING    =  2 << COUNT_BITS;private static final int TERMINATED =  3 << COUNT_BITS;// ctl操作private static int runStateOf(int c)     { return c & ~CAPACITY; }private static int workerCountOf(int c)  { return c & CAPACITY; }private static int ctlOf(int rs, int wc) { return rs | wc; }RUNNING:可接收新任務,可執(zhí)行等待隊列里的任務SHUTDOWN:不可接收新任務,可執(zhí)行等待隊列里的任務STOP:不可接收新任務,不可執(zhí)行等待隊列里的任務,并且嘗試終止所有在運行任務TIDYING:所有任務已經終止,執(zhí)行terminated()TERMINATED:terminated()執(zhí)行完成

線程池狀態(tài)默認從RUNNING開始流轉,到狀態(tài)TERMINATED結束,中間不需要經過每一種狀態(tài),但不能讓狀態(tài)回退。下面是狀態(tài)變化可能的路徑和變化條件:

圖1 線程池狀態(tài)變化路徑

Worker的創(chuàng)建

線程池是由Worker類負責執(zhí)行任務,Worker繼承了AbstractQueuedSynchronizer,引出了Java并發(fā)框架的核心AQS。

AbstractQueuedSynchronizer,簡稱AQS,是Java并發(fā)包里一系列同步工具的基礎實現(xiàn),原理是根據(jù)狀態(tài)位來控制線程的入隊阻塞、出隊喚醒來處理同步。

AQS不會在這里展開討論,只需要知道Worker包裝了Thread,由它去執(zhí)行任務。

調用execute將會根據(jù)線程池的情況創(chuàng)建Worker,可以歸納出下圖四種情況:

圖2 worker在線程池里的四種可能

public void execute(Runnable command) {    if (command == null)        throw new NullPointerException();    int c = ctl.get();    //1    if (workerCountOf(c) < corePoolSize) {        if (addWorker(command, true))            return;        c = ctl.get();    }    //2    if (isRunning(c) && workQueue.offer(command)) {        int recheck = ctl.get();        if (! isRunning(recheck) && remove(command))            //3            reject(command);        else if (workerCountOf(recheck) == 0)            //4            addWorker(null, false);    }    //5    else if (!addWorker(command, false))        //6        reject(command);}

標記1對應第一種情況,要留意addWorker傳入了core,core=true為corePoolSize,core=false為maximumPoolSize,新增時需要檢查workerCount是否超過允許的最大值。

標記2對應第二種情況,檢查線程池是否在運行,并且將任務加入等待隊列。標記3再檢查一次線程池狀態(tài),如果線程池忽然處于非運行狀態(tài),那就將等待隊列剛加的任務刪掉,再交給RejectedExecutionHandler處理。標記4發(fā)現(xiàn)沒有worker,就先補充一個空任務的worker。

標記5對應第三種情況,等待隊列不能再添加任務了,調用addWorker添加一個去處理。

標記6對應第四種情況,addWorker的core傳入false,返回調用失敗,代表workerCount已經超出maximumPoolSize,那就交給RejectedExecutionHandler處理。

private boolean addWorker(Runnable firstTask, boolean core) {        //1        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            }        }        //2        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;    }

標記1的第一段代碼,目的很簡單,是為workerCount加一。至于為什么代碼寫了這么長,是因為線程池的狀態(tài)在不斷變化,并發(fā)環(huán)境下需要保證變量的同步性。外循環(huán)判斷線程池狀態(tài)、任務非空和隊列非空,內循環(huán)使用CAS機制保證workerCount正確地遞增。不了解CAS可以看認識非阻塞的同步機制CAS,后續(xù)增減workerCount都會使用CAS。

標記2的第二段代碼,就比較簡單。創(chuàng)建一個新Worker對象,將Worker添加進workers里(Set集合)。成功添加后,啟動worker里的線程。在finally里判斷線程是否啟動成功,不成功直接調用addWorkerFailed。

private void addWorkerFailed(Worker w) {        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            if (w != null)                workers.remove(w);            decrementWorkerCount();            tryTerminate();        } finally {            mainLock.unlock();        }    }

addWorkerFailed將減少已經遞增的workerCount,并且調用tryTerminate結束線程池。

Worker的執(zhí)行

Worker(Runnable firstTask) {    setState(-1); // inhibit interrupts until runWorker    this.firstTask = firstTask;    this.thread = getThreadFactory().newThread(this);}public void run() {    runWorker(this);}

Worker在構造函數(shù)里采用ThreadFactory創(chuàng)建Thread,在run方法里調用了runWorker,看來是真正執(zhí)行任務的地方。

final void runWorker(Worker w) {    Thread wt = Thread.currentThread();    Runnable task = w.firstTask;    w.firstTask = null;    w.unlock(); // allow interrupts    boolean completedAbruptly = true;    try {       //1        while (task != null || (task = getTask()) != null) {            w.lock();           //2            if ((runStateAtLeast(ctl.get(), STOP) ||                 (Thread.interrupted() &&                  runStateAtLeast(ctl.get(), STOP))) &&                !wt.isInterrupted())                wt.interrupt();            try {               //3                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;                 //4                w.completedTasks++;                w.unlock();            }        }        completedAbruptly = false;       //5    } finally {        //6        processWorkerExit(w, completedAbruptly);    }}

標記1進入循環(huán),從getTask獲取要執(zhí)行的任務,直到返回null。這里達到了線程復用的效果,讓線程處理多個任務。

標記2是一個比較復雜的判斷,保證了線程池在STOP狀態(tài)下線程是中斷的,非STOP狀態(tài)下線程沒有被中斷。如果你不了解Java的中斷機制,看如何正確結束Java線程這篇。

標記3調用了run方法,真正執(zhí)行了任務。執(zhí)行前后提供了beforeExecute和afterExecute兩個方法,由子類實現(xiàn)。

標記4里的completedTasks統(tǒng)計worker執(zhí)行了多少任務,最后累加進completedTaskCount變量,可以調用相應方法返回一些統(tǒng)計信息。

標記5的變量completedAbruptly表示worker是否異常終止,執(zhí)行到這里代表執(zhí)行正常,后續(xù)的方法需要這個變量。

標記6調用processWorkerExit結束,后面會分析。

接著來看worker從等待隊列獲取任務的getTask方法:

private Runnable getTask() {    boolean timedOut = false; // Did the last poll() time out?    for (;;) {        int c = ctl.get();        int rs = runStateOf(c);        //1        // Check if queue empty only if necessary.        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {            decrementWorkerCount();            return null;        }        int wc = workerCountOf(c);        //2        // Are workers subject to culling?        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;        if ((wc > maximumPoolSize || (timed && timedOut))            && (wc > 1 || workQueue.isEmpty())) {            if (compareAndDecrementWorkerCount(c))                return null;            continue;        }       //3        try {            Runnable r = timed ?                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                workQueue.take();            if (r != null)                return r;            timedOut = true;        } catch (InterruptedException retry) {            timedOut = false;        }    }}

標記1檢查線程池的狀態(tài),這里就體現(xiàn)出SHUTDOWN和STOP的區(qū)別。如果線程池是SHUTDOWN狀態(tài),還會先處理完等待隊列的任務;如果是STOP狀態(tài),就不再處理等待隊列里的任務了。

標記2先看allowCoreThreadTimeOut這個變量,false時worker空閑,也不會結束;true時,如果worker空閑超過keepAliveTime,就會結束。接著是一個很復雜的判斷,好難轉成文字描述,自己看吧。注意一下wc>maximumPoolSize,出現(xiàn)這種可能是在運行中調用setMaximumPoolSize,還有wc>1,在等待隊列非空時,至少保留一個worker。

標記3是從等待隊列取任務的邏輯,根據(jù)timed分為等待keepAliveTime或者阻塞直到有任務。

最后來看結束worker需要執(zhí)行的操作:

private void processWorkerExit(Worker w, boolean completedAbruptly) {   //1    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted        decrementWorkerCount();  //2    final ReentrantLock mainLock = this.mainLock;    mainLock.lock();    try {        completedTaskCount += w.completedTasks;        workers.remove(w);    } finally {        mainLock.unlock();    }   //3    tryTerminate();    int c = ctl.get();    //4    if (runStateLessThan(c, STOP)) {        if (!completedAbruptly) {            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;            if (min == 0 && ! workQueue.isEmpty())                min = 1;            if (workerCountOf(c) >= min)                return; // replacement not needed        }        addWorker(null, false);    }}

正常情況下,在getTask里就會將workerCount減一。標記1處用變量completedAbruptly判斷worker是否異常退出,如果是,需要補充對workerCount的減一。

標記2將worker處理任務的數(shù)量累加到總數(shù),并且在集合workers中去除。

標記3嘗試終止線程池,后續(xù)會研究。

標記4處理線程池還是RUNNING或SHUTDOWN狀態(tài)時,如果worker是異常結束,那么會直接addWorker。如果allowCoreThreadTimeOut=true,并且等待隊列有任務,至少保留一個worker;如果allowCoreThreadTimeOut=false,workerCount不少于corePoolSize。

總結一下worker:線程池啟動后,worker在池內創(chuàng)建,包裝了提交的Runnable任務并執(zhí)行,執(zhí)行完就等待下一個任務,不再需要時就結束。

線程池的關閉

線程池的關閉不是一關了事,worker在池里處于不同狀態(tài),必須安排好worker的”后事”,才能真正釋放線程池。ThreadPoolExecutor提供兩種方法關閉線程池:

shutdown:不能再提交任務,已經提交的任務可繼續(xù)運行;shutdownNow:不能再提交任務,已經提交但未執(zhí)行的任務不能運行,在運行的任務可繼續(xù)運行,但會被中斷,返回已經提交但未執(zhí)行的任務。
public void shutdown() {    final ReentrantLock mainLock = this.mainLock;    mainLock.lock();    try {        checkShutdownaccess();   //1 安全策略機制        advanceRunState(SHUTDOWN);   //2        interruptIdleWorkers();   //3        onShutdown(); //4 空方法,子類實現(xiàn)    } finally {        mainLock.unlock();    }    tryTerminate();   //5}

shutdown將線程池切換到SHUTDOWN狀態(tài),并調用interruptIdleWorkers請求中斷所有空閑的worker,最后調用tryTerminate嘗試結束線程池。

public List<Runnable> shutdownNow() {    List<Runnable> tasks;    final ReentrantLock mainLock = this.mainLock;    mainLock.lock();    try {        checkShutdownAccess();        advanceRunState(STOP);        interruptWorkers();        tasks = drainQueue();  //1    } finally {        mainLock.unlock();    }    tryTerminate();    return tasks;}

shutdownNow和shutdown類似,將線程池切換為STOP狀態(tài),中斷目標是所有worker。drainQueue會將等待隊列里未執(zhí)行的任務返回。

interruptIdleWorkers和interruptWorkers實現(xiàn)原理都是遍歷workers集合,中斷條件符合的worker。

上面的代碼多次出現(xiàn)調用tryTerminate,這是一個嘗試將線程池切換到TERMINATED狀態(tài)的方法。

final void tryTerminate() {    for (;;) {        int c = ctl.get();        //1        if (isRunning(c) ||            runStateAtLeast(c, TIDYING) ||            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))            return;        //2        if (workerCountOf(c) != 0) { // Eligible to terminate            interruptIdleWorkers(ONLY_ONE);            return;        }       //3        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {                try {                    terminated();                } finally {                    ctl.set(ctlOf(TERMINATED, 0));                    termination.signalAll();                }                return;            }        } finally {            mainLock.unlock();        }        // else retry on failed CAS    }}

標記1檢查線程池狀態(tài),下面幾種情況,后續(xù)操作都沒有必要,直接return。

RUNNING(還在運行,不能停)TIDYING或TERMINATED(已經沒有在運行的worker)SHUTDOWN并且等待隊列非空(執(zhí)行完才能停)

標記2在worker非空的情況下又調用了interruptIdleWorkers,你可能疑惑在shutdown時已經調用過了,為什么又調用,而且每次只中斷一個空閑worker?你需要知道,shutdown時worker可能在執(zhí)行中,執(zhí)行完阻塞在隊列的take,不知道要結束,所有要補充調用interruptIdleWorkers。每次只中斷一個是因為processWorkerExit時,還會執(zhí)行tryTerminate,自動中斷下一個空閑的worker。

標記3是最終的狀態(tài)切換。線程池會先進入TIDYING狀態(tài),再進入TERMINATED狀態(tài),中間提供了terminated這個空方法供子類實現(xiàn)。

調用關閉線程池方法后,需要等待線程池切換到TERMINATED狀態(tài)。awaitTermination檢查限定時間內線程池是否進入TERMINATED狀態(tài),代碼如下:

public boolean awaitTermination(long timeout, TimeUnit unit)    throws InterruptedException {    long nanos = unit.toNanos(timeout);    final ReentrantLock mainLock = this.mainLock;    mainLock.lock();    try {        for (;;) {            if (runStateAtLeast(ctl.get(), TERMINATED))                return true;            if (nanos <= 0)                return false;            nanos = termination.awaitNanos(nanos);        }    } finally {        mainLock.unlock();    }}

后言

以上過了一遍線程池主要的邏輯,總體來看線程池的設計是很清晰的。如有錯誤或不足,歡迎指出,也歡迎留言交流。今次介紹了線程池運行的生命周期,下篇會研究更細粒度地控制任務的生命周期,也就是submit和Future。

學習Java的同學注意了!!! 學習過程中遇到什么問題或者想獲取學習資源的話,歡迎加入Java學習交流群,群號碼:183993990  我們一起學Java!


發(fā)表評論 共有條評論
用戶名: 密碼:
驗證碼: 匿名發(fā)表
主站蜘蛛池模板: 在线播放中文 | 97精品国产高清在线看入口 | 欧美综合在线观看视频 | 国产亚洲精品久久 | 黄色av电影在线 | 九九热免费观看 | caoporn国产一区二区 | 亚洲精品一区二区三区在线看 | 成人午夜精品 | 成人9禁啪啪无遮挡免费 | 最新国产毛片 | 蜜桃91丨九色丨蝌蚪91桃色 | 最新久久免费视频 | 亚洲欧美国产高清 | 狠狠久久伊人中文字幕 | 精品国产欧美一区二区 | 国产九色在线播放九色 | 久久综合伊人 | 国产一区二区三区四 | 久久丝袜脚交足黄网站免费 | 国产在线a| 成人做爰高潮片免费视频美国 | 一级黄片毛片免费看 | 日韩色视频在线观看 | 九九热视频在线免费观看 | 欧美a∨亚洲欧美亚洲 | 依依成人综合 | 欧美亚洲一级 | 毛片视频网站 | 国产精品www | 日本a在线观看 | 羞羞的网站 | 欧美在线观看视频一区二区 | 99在线免费观看视频 | av电影网站在线 | 亚洲一区二区中文字幕在线观看 | 美女很黄很黄免费的 | 成人h精品动漫一区二区三区 | 亚洲网站在线 | 久久精品中文字幕一区 | 亚洲午夜久久久精品一区二区三区 |