Java線程池中BlockingQueue的做用

關於線程池中BlockingQueue的疑問

對於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的線程池，只有消費者使用了阻塞的方法，生產者並無。

SynchronousQueue

不過也有例外，調用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/