成人无码视频,亚洲精品久久久久av无码,午夜精品久久久久久毛片,亚洲 中文字幕 日韩 无码

資訊專(zhuān)欄INFORMATION COLUMN

ThreadPoolExecutor線(xiàn)程池源碼分析

stormzhang / 686人閱讀

摘要:線(xiàn)程池技術(shù)旨在解決兩個(gè)不同的問(wèn)題在處理大量異步任務(wù)時(shí)可以提高性能,因?yàn)闇p少了線(xiàn)程的銷(xiāo)毀,新建,切換等消耗性能的操作。線(xiàn)程池還有能力統(tǒng)一管理,調(diào)度,監(jiān)控,調(diào)優(yōu)線(xiàn)程等,還提供了一下基本的統(tǒng)計(jì),比如已完成的任務(wù)數(shù)量。線(xiàn)程數(shù)量,線(xiàn)程池的狀態(tài)。

了解ThreadPoolExecutor

先看一下線(xiàn)程池類(lèi)的類(lèi)圖關(guān)系:

Executor接口

Executor作者描述的是Executor提供了一種解耦方式將任務(wù)的提交和任務(wù)以何種技術(shù)執(zhí)行分離;
Executor接口只有一個(gè)方法:

void execute(Runnable command);

execute方法接收一個(gè)Runnable對(duì)象,方法的描述是在未來(lái)的某個(gè)時(shí)間執(zhí)行command。不管是在一個(gè)新的線(xiàn)程中執(zhí)行,還是在線(xiàn)程池中執(zhí)行,甚至在調(diào)用者線(xiàn)程中立即執(zhí)行。

ExecutorService接口

ExecutorService繼承了Executor接口,ExecutorService可以被關(guān)閉,關(guān)閉以后不再接收新的任務(wù)。ExecutorService提供了兩個(gè)不同的方法關(guān)閉ExecutorService。shutdown方法會(huì)等待之前還未執(zhí)行的任務(wù)執(zhí)行完畢再關(guān)閉,而shutdownNow則不會(huì)再啟動(dòng)新的任務(wù),還會(huì)中斷正在執(zhí)行的任務(wù)。一旦關(guān)閉后,ExecutorService就不會(huì)有正在執(zhí)行的任務(wù),也不會(huì)有等待被執(zhí)行的任務(wù),更不會(huì)有新的任務(wù)被提交。ExecutorService關(guān)閉后應(yīng)該處理好一些資源的回收。

ThreadPoolExecutor

線(xiàn)程池技術(shù)旨在解決兩個(gè)不同的問(wèn)題:

在處理大量異步任務(wù)時(shí)可以提高性能,因?yàn)闇p少了線(xiàn)程的銷(xiāo)毀,新建,切換等消耗性能的操作。

線(xiàn)程池還有能力統(tǒng)一管理,調(diào)度,監(jiān)控,調(diào)優(yōu)線(xiàn)程等,還提供了一下基本的統(tǒng)計(jì),比如已完成的任務(wù)數(shù)量。

重要的狀態(tài)和狀態(tài)判斷的方法
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//高3位和低29位分別表示狀態(tài)和線(xiàn)程數(shù)
private static final int COUNT_BITS = Integer.SIZE - 3;
//1左移29位減一得到低29位都是1,即線(xiàn)程的最大數(shù)量,大概5億多
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;//111
private static final int SHUTDOWN   =  0 << COUNT_BITS;//000
private static final int STOP       =  1 << COUNT_BITS;//001
private static final int TIDYING    =  2 << COUNT_BITS;//010
private static final int TERMINATED =  3 << COUNT_BITS;//011

// Packing and unpacking ctl
//獲得狀態(tài)
private static int runStateOf(int c)     { return c & ~CAPACITY; }
//獲得線(xiàn)程數(shù)量
private static int workerCountOf(int c)  { return c & CAPACITY; }
//通過(guò)狀態(tài)和線(xiàn)程數(shù)量組裝ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }

/*
 * Bit field accessors that don"t require unpacking ctl.
 * These depend on the bit layout and on workerCount being never negative.
 */
//c狀態(tài)是否小于s狀態(tài)
private static boolean runStateLessThan(int c, int s) {
    return c < s;
}
//c狀態(tài)是否大于等于s狀態(tài)
private static boolean runStateAtLeast(int c, int s) {
    return c >= s;
}
//線(xiàn)程池是否是運(yùn)行狀態(tài)
private static boolean isRunning(int c) {
    return c < SHUTDOWN;
}

整個(gè)類(lèi)最重要的一個(gè)狀態(tài)標(biāo)志ctl是一個(gè)AtomicInteger,它包含了兩個(gè)字段的含義。workerCount線(xiàn)程數(shù)量,runState線(xiàn)程池的狀態(tài)。
這一個(gè)字段是如何包含兩個(gè)字段的含義的呢,Doug Lea大牛使用了一個(gè)int的32位bits的高三位保存了狀態(tài)值,低29位保存了線(xiàn)程數(shù)量。

其中五個(gè)狀態(tài):
RUNNING:接收新的任務(wù),處理隊(duì)列中的任務(wù);
SHUTDOWN:不接收新的任務(wù),但處理隊(duì)列中的任務(wù);
STOP:不接收新的任務(wù),不處理隊(duì)列中的任務(wù),中斷正在執(zhí)行的任務(wù);
TIDYING:所有任務(wù)都終止,線(xiàn)程數(shù)為0, 線(xiàn)程過(guò)度到TIDYING時(shí)會(huì)調(diào)用terminated鉤子方法;
TERMINATED:terminated執(zhí)行完畢;

狀態(tài)之間的轉(zhuǎn)換:
RUNNING -> SHUTDOWN:調(diào)用shutdown方法;
(RUNNING or SHUTDOWN) -> STOP:調(diào)用shutdownNow方法;
SHUTDOWN -> TIDYING:當(dāng)線(xiàn)程池和任務(wù)隊(duì)列都為空;
STOP -> TIDYING:當(dāng)線(xiàn)程池為空;
TIDYING -> TERMINATED:當(dāng)terminated方法執(zhí)行完畢;

Worker介紹

Worker類(lèi)主要包含了線(xiàn)程運(yùn)行任務(wù)時(shí)的終端控制狀態(tài),同時(shí)還有一些少量的信息記錄。Worker適時(shí)的繼承了AQS,讓線(xiàn)程在任務(wù)執(zhí)行之間獲取鎖和釋放鎖變得簡(jiǎn)單。這確保了中斷是喚醒一個(gè)等待任務(wù)的線(xiàn)程,而不是中斷一個(gè)正在運(yùn)行的任務(wù)線(xiàn)程。

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;

    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;

    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }

    // Lock methods
    //
    // The value 0 represents the unlocked state.
    // The value 1 represents the locked state.

    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }

    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

Worker繼承了AQS,實(shí)現(xiàn)了Runnable接口;在構(gòu)造函數(shù)中,初始化了它的第一次仍無(wú),使用threadFactory創(chuàng)建一個(gè)新的線(xiàn)程;
Worker繼承AQS,目的是想使用獨(dú)占鎖來(lái)表示線(xiàn)程是否正在執(zhí)行任務(wù),Worker的線(xiàn)程獲取了獨(dú)占鎖就說(shuō)明它在執(zhí)行任務(wù),不能被中斷。從tryAcquire方法可以看出,它實(shí)現(xiàn)的是不可重入鎖,因?yàn)槭欠瘾@得鎖在這里表示一個(gè)狀態(tài),如果可以重入的話(huà),獨(dú)占鎖就失去了只表示一個(gè)狀態(tài)的含義。在構(gòu)造函數(shù)初始化時(shí),Worker將state設(shè)置為-1,因?yàn)樵趖ryAcquire中CAS操作compareAndSetState(0, 1),表示state在-1時(shí)不能被中斷。在runWorker中將state設(shè)置為0.

ThreadPooleExecutor構(gòu)造方法
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

說(shuō)明一下各參數(shù)的含義:
corePoolSize:核心線(xiàn)程數(shù)量,即使線(xiàn)程是空閑的也保持在線(xiàn)程池中,除非allowCoreThreadTimeOut參數(shù)被設(shè)置;

maximumPoolSize:最大線(xiàn)程數(shù)量;

keepAliveTime:當(dāng)線(xiàn)程數(shù)量超過(guò)核心線(xiàn)程數(shù)量時(shí),超出的空閑線(xiàn)程等待新任務(wù)的最大時(shí)長(zhǎng);

unit:時(shí)間單位;

workQueue:存放將要被執(zhí)行的任務(wù)的隊(duì)列;

threadFactory:創(chuàng)建線(xiàn)程的線(xiàn)程工廠;

handler:當(dāng)任務(wù)隊(duì)列滿(mǎn)且沒(méi)有空閑的線(xiàn)程時(shí)處理任務(wù)的handler,線(xiàn)程池提供了四種策略:

AbortPolicy:直接拋出異常,默認(rèn);

CallerRunsPolicy:使用調(diào)用者的線(xiàn)程執(zhí)行;

DiscardOldestPolicy:拋棄隊(duì)列最前的任務(wù),執(zhí)行當(dāng)前任務(wù);

DiscardPolicy:直接丟棄任務(wù);

這些參數(shù)對(duì)整個(gè)線(xiàn)程池運(yùn)行非常重要;

execute方法
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.
     */
    //獲取ctl
    int c = ctl.get();
    //如果線(xiàn)程數(shù)小于核心線(xiàn)程數(shù)
    if (workerCountOf(c) < corePoolSize) {
        //添加線(xiàn)程并執(zhí)行任務(wù)
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //線(xiàn)程數(shù)大于核心線(xiàn)程數(shù)
    //如果線(xiàn)程池running狀態(tài)且添加任務(wù)到隊(duì)列成功
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        //如果線(xiàn)程池不是運(yùn)行狀態(tài),隊(duì)列移除任務(wù),使用拒絕策略處理任務(wù)
        if (! isRunning(recheck) && remove(command))
            reject(command);
        //如果這時(shí)線(xiàn)程數(shù)為0,添加任務(wù)
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    //隊(duì)列滿(mǎn),添加線(xiàn)程失敗,使用拒絕策略處理任務(wù)
    else if (!addWorker(command, false))
        reject(command);
}

在線(xiàn)程池添

數(shù)量如果小于核心線(xiàn)程數(shù),則添加新的線(xiàn)程并執(zhí)行當(dāng)前任務(wù),否則判斷如果隊(duì)列是否未滿(mǎn),則添加當(dāng)前任務(wù)到隊(duì)列,否則判斷線(xiàn)程數(shù)量如果小于最大線(xiàn)程數(shù),則添加新的線(xiàn)程并執(zhí)行,否則使用拒絕策略處理當(dāng)前任務(wù)。

addWorker方法

addWorker方法主要是添加線(xiàn)程并執(zhí)行任務(wù):

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        //獲取線(xiàn)程池運(yùn)行狀態(tài)
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        //如果運(yùn)行狀態(tài)大于等于SHUTDOWN,不再接受新的任務(wù),返回false
        //如果運(yùn)行狀態(tài)等于SHUTDOWN且firstTask不為空,繼續(xù)執(zhí)行下去,如果firstTask為空,queue為空,返回false,否則繼續(xù)執(zhí)行;只要SHUTDOWN狀態(tài)下還有任務(wù)在,就需要往下執(zhí)行,可能需要新建worker執(zhí)行
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            //獲得線(xiàn)程數(shù)量
            int wc = workerCountOf(c);
            //如果線(xiàn)程數(shù)量大于容量或者當(dāng)core為true時(shí)wc大于等于核心線(xiàn)程數(shù),當(dāng)core為falsewc大于等于最大線(xiàn)程數(shù)量時(shí),返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            //CAS線(xiàn)程數(shù)加一,成功則中斷循環(huán)
            if (compareAndIncrementWorkerCount(c))
                break retry;
            //如果CAS失敗,重新獲取ctl,線(xiàn)程池運(yùn)行狀態(tài)沒(méi)變的話(huà)繼續(xù)loop
            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 {
        //新建一個(gè)worker
        w = new Worker(firstTask);
        //能得到worker的thread
        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());

                //如果rs是RUNNING或者SHUTDOWN且firstTask為null
                //因?yàn)镾HUTDOWN時(shí)還需要執(zhí)行queue中的任務(wù)
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    //往線(xiàn)程池中添加worker
                    workers.add(w);
                    int s = workers.size();
                    //記錄線(xiàn)程池出現(xiàn)的最大線(xiàn)程數(shù)量
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                //啟動(dòng)worker
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

worker的run方法調(diào)用的是runWorker;

runWorker方法
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    //保存worker的第一個(gè)任務(wù)
    Runnable task = w.firstTask;
    //清空worker的第一個(gè)任務(wù)
    w.firstTask = null;
    //這里將worker的state設(shè)置為0,允許中斷
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        //如果task為空,則從隊(duì)列中獲取任務(wù)
        while (task != null || (task = getTask()) != null) {
            //開(kāi)始執(zhí)行任務(wù),不允許中斷
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            //如果當(dāng)前狀態(tài)大于等于STOP要保持當(dāng)前線(xiàn)程中斷
            //如果當(dāng)前線(xiàn)程小于STOP即RUNNING或者SHUTDOWN,調(diào)用Thread.interrupted()清空中斷標(biāo)志,如果這時(shí)調(diào)用了shutdownNow狀態(tài)為STOP,還是要保持中斷狀態(tài)
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                //執(zhí)行任務(wù)前做的事
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    //執(zhí)行任務(wù)
                    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 {
                    //執(zhí)行任務(wù)之后做的事
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                //worker的完成任務(wù)數(shù)量加一,此時(shí)是線(xiàn)程安全的
                w.completedTasks++;
                //釋放鎖
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        //線(xiàn)程退出
        processWorkerExit(w, completedAbruptly);
    }
}

每個(gè)task在調(diào)用runWorker后會(huì)一直循環(huán)執(zhí)行任務(wù),直到queue中沒(méi)有任務(wù)了,循環(huán)結(jié)束,worker生命周期結(jié)束。

getTask

上面runWorker時(shí)調(diào)用了getTask去獲取隊(duì)列中的任務(wù),下面我們看一下這個(gè)方法:

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        //如果rs大于等于SHUTDOWN,當(dāng)RS大于等于STOP說(shuō)明線(xiàn)程池已經(jīng)不處理隊(duì)列中的任務(wù)了,當(dāng)rs為SHUTDOWN時(shí),如果隊(duì)列是空的,返回null
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            //線(xiàn)程數(shù)減一
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        //是否超時(shí)控制,allowCoreThreadTimeOut默認(rèn)false,代表不允許核心線(xiàn)程超時(shí),對(duì)于超出核心線(xiàn)程的線(xiàn)程需要控制超時(shí)
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        //當(dāng)線(xiàn)程數(shù)大于最大線(xiàn)程數(shù),或者需要超時(shí)控制且上次獲取任務(wù)超時(shí)
        //且線(xiàn)程數(shù)大于1或者隊(duì)列為空,嘗試將線(xiàn)程數(shù)減一并返回null
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            //失敗重試
            continue;
        }

        try {
            //當(dāng)需要超時(shí)控制時(shí),在keepAliveTime時(shí)間內(nèi)沒(méi)有獲取到任務(wù)的話(huà)返回null,否則調(diào)用take獲取任務(wù),此時(shí)線(xiàn)程時(shí)阻塞的
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

getTask方法在線(xiàn)程數(shù)量大于核心線(xiàn)程數(shù)時(shí)會(huì)判斷在獲取task時(shí)進(jìn)行超時(shí)判斷(poll),超時(shí)返回null這時(shí)getTask返回null,那當(dāng)前worker的loop結(jié)束即run方法結(jié)束,線(xiàn)程生命周期結(jié)束。而核心線(xiàn)程則會(huì)調(diào)用take方法,當(dāng)沒(méi)有任務(wù)時(shí)會(huì)阻塞。

processWorkerExit

runTask方法最后會(huì)調(diào)用processWorkerExit方法進(jìn)行一些cleanup工作。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    //completedAbruptly為true時(shí)代表發(fā)生了異常,線(xiàn)程數(shù)減一
    if (completedAbruptly) // If abrupt, then workerCount wasn"t adjusted
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //統(tǒng)計(jì)完成任務(wù)數(shù)
        completedTaskCount += w.completedTasks;
        //線(xiàn)程池移除當(dāng)前worker
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    // 根據(jù)線(xiàn)程池狀態(tài)進(jìn)行判斷是否結(jié)束線(xiàn)程池
    tryTerminate();

    int c = ctl.get();
    //當(dāng)線(xiàn)程池狀態(tài)為RUNNING或者SHUTDOWN時(shí)
    //如果發(fā)生異常,重新加入一個(gè)worker replacement
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            //當(dāng)allowCoreThreadTimeOut為true,最少要一個(gè)worker
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            //當(dāng)線(xiàn)程數(shù)大于等于最少需要的線(xiàn)程數(shù),則不需要add新的worker
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

tryTerminate方法

上面我們跳過(guò)了tryTerminate方法,該方法判斷是否要結(jié)束線(xiàn)程池,這里看一下

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        //當(dāng)線(xiàn)程池狀態(tài)時(shí)RUNNING或者已經(jīng)TIDYING或者已經(jīng)TERMINATED或者SHUTDOWN且還有任務(wù)沒(méi)有被執(zhí)行,直接返回
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        // 如果線(xiàn)程數(shù)不為0,則中斷一個(gè)空閑的工作線(xiàn)程
        if (workerCountOf(c) != 0) { // Eligible to terminate
            //workQueue.take()時(shí)如果queue一直為空的話(huà),線(xiàn)程會(huì)一直阻塞
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //如果狀態(tài)設(shè)置成功為T(mén)IDYING,調(diào)用勾子方法terminated,該方法留給了子類(lèi)實(shí)現(xiàn)
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    //設(shè)置狀態(tài)為T(mén)ERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}
interruptIdleWorkers

上面說(shuō)為了當(dāng)隊(duì)列一直為空的時(shí)候,核心線(xiàn)程會(huì)一直阻塞,所以調(diào)用了interruptIdleWorkers,我們看一下執(zhí)行了什么:

private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

遍歷線(xiàn)程池中所有的線(xiàn)程,若線(xiàn)程沒(méi)有被中斷tryLock成功,就中斷該線(xiàn)程,LockSupport.park()能響應(yīng)中斷信號(hào),阻塞的線(xiàn)程被中斷喚醒。

文章版權(quán)歸作者所有,未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為,您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請(qǐng)注明本文地址:http://m.hztianpu.com/yun/70949.html

相關(guān)文章

  • 使用 Executors,ThreadPoolExecutor,創(chuàng)建線(xiàn)程源碼分析理解

    摘要:源碼分析創(chuàng)建可緩沖的線(xiàn)程池。源碼分析使用創(chuàng)建線(xiàn)程池源碼分析的構(gòu)造函數(shù)構(gòu)造函數(shù)參數(shù)核心線(xiàn)程數(shù)大小,當(dāng)線(xiàn)程數(shù),會(huì)創(chuàng)建線(xiàn)程執(zhí)行最大線(xiàn)程數(shù),當(dāng)線(xiàn)程數(shù)的時(shí)候,會(huì)把放入中保持存活時(shí)間,當(dāng)線(xiàn)程數(shù)大于的空閑線(xiàn)程能保持的最大時(shí)間。 之前創(chuàng)建線(xiàn)程的時(shí)候都是用的 newCachedThreadPoo,newFixedThreadPool,newScheduledThreadPool,newSingleThr...

    Chiclaim 評(píng)論0 收藏0
  • 后端ing

    摘要:當(dāng)活動(dòng)線(xiàn)程核心線(xiàn)程非核心線(xiàn)程達(dá)到這個(gè)數(shù)值后,后續(xù)任務(wù)將會(huì)根據(jù)來(lái)進(jìn)行拒絕策略處理。線(xiàn)程池工作原則當(dāng)線(xiàn)程池中線(xiàn)程數(shù)量小于則創(chuàng)建線(xiàn)程,并處理請(qǐng)求。當(dāng)線(xiàn)程池中的數(shù)量等于最大線(xiàn)程數(shù)時(shí)默默丟棄不能執(zhí)行的新加任務(wù),不報(bào)任何異常。 spring-cache使用記錄 spring-cache的使用記錄,坑點(diǎn)記錄以及采用的解決方案 深入分析 java 線(xiàn)程池的實(shí)現(xiàn)原理 在這篇文章中,作者有條不紊的將 ja...

    roadtogeek 評(píng)論0 收藏0
  • 線(xiàn)程源碼分析

    摘要:線(xiàn)程池的作用線(xiàn)程池能有效的處理多個(gè)線(xiàn)程的并發(fā)問(wèn)題,避免大量的線(xiàn)程因?yàn)榛ハ鄰?qiáng)占系統(tǒng)資源導(dǎo)致阻塞現(xiàn)象,能夠有效的降低頻繁創(chuàng)建和銷(xiāo)毀線(xiàn)程對(duì)性能所帶來(lái)的開(kāi)銷(xiāo)。固定的線(xiàn)程數(shù)由系統(tǒng)資源設(shè)置。線(xiàn)程池的排隊(duì)策略與有關(guān)。線(xiàn)程池的狀態(tài)值分別是。 線(xiàn)程池的作用 線(xiàn)程池能有效的處理多個(gè)線(xiàn)程的并發(fā)問(wèn)題,避免大量的線(xiàn)程因?yàn)榛ハ鄰?qiáng)占系統(tǒng)資源導(dǎo)致阻塞現(xiàn)象,能夠有效的降低頻繁創(chuàng)建和銷(xiāo)毀線(xiàn)程對(duì)性能所帶來(lái)的開(kāi)銷(xiāo)。 線(xiàn)程池的...

    enda 評(píng)論0 收藏0
  • 一看就懂的Java線(xiàn)程分析詳解

    摘要:任務(wù)性質(zhì)不同的任務(wù)可以用不同規(guī)模的線(xiàn)程池分開(kāi)處理。線(xiàn)程池在運(yùn)行過(guò)程中已完成的任務(wù)數(shù)量。如等于線(xiàn)程池的最大大小,則表示線(xiàn)程池曾經(jīng)滿(mǎn)了。線(xiàn)程池的線(xiàn)程數(shù)量。獲取活動(dòng)的線(xiàn)程數(shù)。通過(guò)擴(kuò)展線(xiàn)程池進(jìn)行監(jiān)控??蚣馨ň€(xiàn)程池,,,,,,等。 Java線(xiàn)程池 [toc] 什么是線(xiàn)程池 線(xiàn)程池就是有N個(gè)子線(xiàn)程共同在運(yùn)行的線(xiàn)程組合。 舉個(gè)容易理解的例子:有個(gè)線(xiàn)程組合(即線(xiàn)程池,咱可以比喻為一個(gè)公司),里面有3...

    Yangder 評(píng)論0 收藏0
  • Java ThreadPoolExecutor 線(xiàn)程源碼分析

    摘要:線(xiàn)程池常見(jiàn)實(shí)現(xiàn)線(xiàn)程池一般包含三個(gè)主要部分調(diào)度器決定由哪個(gè)線(xiàn)程來(lái)執(zhí)行任務(wù)執(zhí)行任務(wù)所能夠的最大耗時(shí)等線(xiàn)程隊(duì)列存放并管理著一系列線(xiàn)程這些線(xiàn)程都處于阻塞狀態(tài)或休眠狀態(tài)任務(wù)隊(duì)列存放著用戶(hù)提交的需要被執(zhí)行的任務(wù)一般任務(wù)的執(zhí)行的即先提交的任務(wù)先被執(zhí)行調(diào)度 線(xiàn)程池常見(jiàn)實(shí)現(xiàn) 線(xiàn)程池一般包含三個(gè)主要部分: 調(diào)度器: 決定由哪個(gè)線(xiàn)程來(lái)執(zhí)行任務(wù), 執(zhí)行任務(wù)所能夠的最大耗時(shí)等 線(xiàn)程隊(duì)列: 存放并管理著一系列線(xiàn)...

    greatwhole 評(píng)論0 收藏0
  • Java調(diào)度線(xiàn)程ScheduledThreadPoolExecutor源碼分析

    摘要:當(dāng)面試官問(wèn)線(xiàn)程池時(shí),你應(yīng)該知道些什么一執(zhí)行流程與不同,向中提交任務(wù)的時(shí)候,任務(wù)被包裝成對(duì)象加入延遲隊(duì)列并啟動(dòng)一個(gè)線(xiàn)程。當(dāng)我們創(chuàng)建出一個(gè)調(diào)度線(xiàn)程池以后,就可以開(kāi)始提交任務(wù)了。 最近新接手的項(xiàng)目里大量使用了ScheduledThreadPoolExecutor類(lèi)去執(zhí)行一些定時(shí)任務(wù),之前一直沒(méi)有機(jī)會(huì)研究這個(gè)類(lèi)的源碼,這次趁著機(jī)會(huì)好好研讀一下。 原文地址:http://www.jianshu....

    kohoh_ 評(píng)論0 收藏0

發(fā)表評(píng)論

0條評(píng)論

閱讀需要支付1元查看
<