摘要：關(guān)于線程池中的疑問(wèn)對(duì)于線程池，相信大家都或多或少使用過(guò)。所以，阻塞隊(duì)列的作用是控制線程池中線程的生命周期。也就是說(shuō)，在的線程池，只有消費(fèi)者使用了阻塞的方法，生產(chǎn)者并沒(méi)有。

對(duì)于Java線程池，相信大家都或多或少使用過(guò)。關(guān)于其用法和原理介紹，網(wǎng)上已經(jīng)有很多非常精彩的文章，珠玉在前，我就不獻(xiàn)丑了。不了解的，可以參考這篇文章。今天我想講的，是關(guān)于我對(duì)Java線程次的兩個(gè)疑問(wèn)，當(dāng)然高手可以略過(guò)了。

1.為什么線程池要使用BlockingQueue，而不是ArrayList或別的什么列表？
2.既然使用了BlockingQueue，為什么還要設(shè)置拒絕策略，隊(duì)列滿的時(shí)候不是阻塞嗎？

要回答這個(gè)問(wèn)答，首先來(lái)看看不用線程池的時(shí)候怎么執(zhí)行異步任務(wù)

new Thread(() -> {
    // do something 
}).start();

也就是說(shuō)，每次需要執(zhí)行異步任務(wù)的時(shí)候，新建一個(gè)線程去執(zhí)行，執(zhí)行完就回收了。這會(huì)導(dǎo)致什么問(wèn)題呢，首先，是對(duì)資源的浪費(fèi)，線程的創(chuàng)建需要陷入內(nèi)核，需要分配?？臻g，需要執(zhí)行調(diào)度，等等，只使用一次就回收太浪費(fèi)資源。其次，當(dāng)異步任務(wù)比較多的時(shí)候，這種方式要?jiǎng)?chuàng)建大量的線程，這對(duì)于內(nèi)存資源也是一個(gè)很大的開(kāi)銷。我們知道，在jvm啟動(dòng)的時(shí)候可以設(shè)置線程棧大小的參數(shù)-Xss，默認(rèn)的大小是1M，如果同時(shí)啟動(dòng)1000個(gè)線程，就要占用1G的內(nèi)存，可想而知，這對(duì)內(nèi)存是一個(gè)多大的開(kāi)銷。而且，線程數(shù)太多，對(duì)于內(nèi)核的調(diào)度壓力也是相當(dāng)大的，而且，因?yàn)轭l繁的上下文切換而使程序的局部性喪失，也是一種消耗。線程池的作用，就是線程的復(fù)用，那么，怎么復(fù)用呢，來(lái)看一段代碼：

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

在ThreadPoolExecutor中，線程封裝在Worker中，Worker實(shí)現(xiàn)了Runnable，同時(shí)在run()方法中調(diào)用上面的runWorker()方法，只要runWorker()方法沒(méi)有執(zhí)行完，這個(gè)線程就不會(huì)被回收。而runWorker()方法要執(zhí)行下去，就要保證while (task != null || (task = getTask()) != null)的條件為真，第一次判斷時(shí)task為firstTask，即執(zhí)行的第一個(gè)任務(wù)，那么要點(diǎn)就成了getTask()必須不能為空，來(lái)看看getTask()的實(shí)現(xiàn)：

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.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            int wc = workerCountOf(c);

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

            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

核心邏輯是：

Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();

這里的workQueue就是阻塞隊(duì)列，timed表示是否會(huì)超時(shí)釋放，keepAliveTime是非核心線程允許的空閑時(shí)間；如果不超時(shí)，則調(diào)用BlockingQueue.take()，如果取不到值，就會(huì)一直阻塞直到程序提交了一個(gè)任務(wù)。所以，阻塞隊(duì)列的作用是控制線程池中線程的生命周期。

那么，如果不用阻塞隊(duì)列，有沒(méi)有別的方式可以實(shí)現(xiàn)線程池的功能？答案是，有，但是沒(méi)必要。比如我們可以使用wait/notify來(lái)控制線程的執(zhí)行和阻塞，但這里使用生產(chǎn)者/消費(fèi)者模式來(lái)實(shí)現(xiàn)是一種更優(yōu)雅的方式。

既然使用了阻塞隊(duì)列，那添加任務(wù)的時(shí)候如果隊(duì)列滿了不就阻塞了嗎，拒絕策略是干嘛用的？答案是添加任務(wù)調(diào)用的并不是阻塞的put()方法，而是非阻塞的offer()方法，看一下ThreadPoolExecutor的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.
         */
        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();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

至于為什么這么實(shí)現(xiàn)，應(yīng)該是不希望阻塞用戶進(jìn)程吧。

也就是說(shuō)，在Java的線程池，只有消費(fèi)者使用了阻塞的方法，生產(chǎn)者并沒(méi)有。

不過(guò)也有例外，調(diào)用ExecutorService executorService = Executors.newCachedThreadPool();
時(shí)，BlockingQueue的實(shí)現(xiàn)類是SynchronousQueue，顧名思義，這是一個(gè)同步隊(duì)列，其內(nèi)部沒(méi)有容量，使用SynchronousQueue，消費(fèi)者線程和生產(chǎn)者線程必須交替執(zhí)行，也就是說(shuō)，生產(chǎn)者和消費(fèi)者都必須等待對(duì)方就緒。這樣的話，不就阻塞用戶進(jìn)程了嗎。確實(shí)會(huì)，但是這個(gè)時(shí)間非常短，因?yàn)槭褂眠@種方式，每次通過(guò)execute()提交任務(wù)的時(shí)候，要么復(fù)用現(xiàn)有空閑的線程，要么新建一個(gè)線程，也就是說(shuō)線程數(shù)理論上沒(méi)有上界，所以可以當(dāng)作不會(huì)阻塞

https://stackoverflow.com/questions/7556465/why-threadpoolexecutor-has-blockingqueue-as-its-argument?utm_medium=organic&utm_source=google_rich_qa&utm_campaign=google_rich_qa

http://www.geek-programmer.com/java-blocking-queues-explained/