摘要：關于線程池中的疑問對于線程池，相信大家都或多或少使用過。所以，阻塞隊列的作用是控制線程池中線程的生命周期。也就是說，在的線程池，只有消費者使用了阻塞的方法，生產者并沒有。

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

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

要回答這個問答，首先來看看不用線程池的時候怎么執行異步任務

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

也就是說，每次需要執行異步任務的時候，新建一個線程去執行，執行完就回收了。這會導致什么問題呢，首先，是對資源的浪費，線程的創建需要陷入內核，需要分配棧空間，需要執行調度，等等，只使用一次就回收太浪費資源。其次，當異步任務比較多的時候，這種方式要創建大量的線程，這對于內存資源也是一個很大的開銷。我們知道，在jvm啟動的時候可以設置線程棧大小的參數-Xss，默認的大小是1M，如果同時啟動1000個線程，就要占用1G的內存，可想而知，這對內存是一個多大的開銷。而且，線程數太多，對于內核的調度壓力也是相當大的，而且，因為頻繁的上下文切換而使程序的局部性喪失，也是一種消耗。線程池的作用，就是線程的復用，那么，怎么復用呢，來看一段代碼：

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實現了Runnable，同時在run()方法中調用上面的runWorker()方法，只要runWorker()方法沒有執行完，這個線程就不會被回收。而runWorker()方法要執行下去，就要保證while (task != null || (task = getTask()) != null)的條件為真，第一次判斷時task為firstTask，即執行的第一個任務，那么要點就成了getTask()必須不能為空，來看看getTask()的實現：

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就是阻塞隊列，timed表示是否會超時釋放，keepAliveTime是非核心線程允許的空閑時間；如果不超時，則調用BlockingQueue.take()，如果取不到值，就會一直阻塞直到程序提交了一個任務。所以，阻塞隊列的作用是控制線程池中線程的生命周期。

那么，如果不用阻塞隊列，有沒有別的方式可以實現線程池的功能？答案是，有，但是沒必要。比如我們可以使用wait/notify來控制線程的執行和阻塞，但這里使用生產者/消費者模式來實現是一種更優雅的方式。

既然使用了阻塞隊列，那添加任務的時候如果隊列滿了不就阻塞了嗎，拒絕策略是干嘛用的？答案是添加任務調用的并不是阻塞的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);
    }

至于為什么這么實現，應該是不希望阻塞用戶進程吧。

也就是說，在Java的線程池，只有消費者使用了阻塞的方法，生產者并沒有。

不過也有例外，調用ExecutorService executorService = Executors.newCachedThreadPool();
時，BlockingQueue的實現類是SynchronousQueue，顧名思義，這是一個同步隊列，其內部沒有容量，使用SynchronousQueue，消費者線程和生產者線程必須交替執行，也就是說，生產者和消費者都必須等待對方就緒。這樣的話，不就阻塞用戶進程了嗎。確實會，但是這個時間非常短，因為使用這種方式，每次通過execute()提交任務的時候，要么復用現有空閑的線程，要么新建一個線程，也就是說線程數理論上沒有上界，所以可以當作不會阻塞

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/