在并發(fā)編程中使用生產(chǎn)者和消費(fèi)者模式能夠解決絕大多數(shù)并發(fā)問題。該模式通過平衡生產(chǎn)線程和消費(fèi)線程的工作能力來提高程序整體處理數(shù)據(jù)的速度。

在線程世界里，生產(chǎn)者就是生產(chǎn)數(shù)據(jù)的線程，消費(fèi)者就是消費(fèi)數(shù)據(jù)的線程。在多線程開發(fā)中，如果生產(chǎn)者處理速度很快，而消費(fèi)者處理速度很慢，那么生產(chǎn)者就必須等待消費(fèi)者處理完，才能繼續(xù)生產(chǎn)數(shù)據(jù)。同樣的道理，如果消費(fèi)者的處理能力大于生產(chǎn)者，那么消費(fèi)者就必須等待生產(chǎn)者。

為了解決這種生產(chǎn)消費(fèi)能力不均衡的問題，便有了生產(chǎn)者和消費(fèi)者模式。生產(chǎn)者和消費(fèi)者模式是通過一個(gè)容器來解決生產(chǎn)者和消費(fèi)者的強(qiáng)耦合問題。生產(chǎn)者和消費(fèi)者彼此之間不直接通信，而是通過阻塞隊(duì)列來進(jìn)行通信，所以生產(chǎn)者生產(chǎn)完數(shù)據(jù)之后不用等待消費(fèi)者處理，直接扔給阻塞隊(duì)列，消費(fèi)者不找生產(chǎn)者要數(shù)據(jù)，而是直接從阻塞隊(duì)列里取，阻塞隊(duì)列就相當(dāng)于一個(gè)緩沖區(qū)，平衡了生產(chǎn)者和消費(fèi)者的處理能力。

阻塞隊(duì)列常用于生產(chǎn)者和消費(fèi)者的場景，生產(chǎn)者是向隊(duì)列里添加元素的線程，消費(fèi)者是從隊(duì)列里取元素的線程。阻塞隊(duì)列就是生產(chǎn)者用來存放元素、消費(fèi)者用來獲取元素的容器。

在Android開發(fā)中阻塞隊(duì)列也是常見的 —— Handler機(jī)制中的MessageQueue就是優(yōu)先級(jí)阻塞隊(duì)列

二、阻塞隊(duì)列有什么用？

解耦 在生產(chǎn)者和消費(fèi)者之間解除了耦合

平衡兩者性能差異 平衡了生產(chǎn)者消費(fèi)者之間的性能差異

三、阻塞隊(duì)列的簡單實(shí)用

①常見的阻塞隊(duì)列主要有那些？

常見的阻塞隊(duì)列主要有一下7中：

• ArrayBlockingQueue
• LinkedBlockingQueue
• PriorityBlockingQueue
• DelayQueue
• SynchronousQueue
• LinkedTransferQueue
• LinkedBlockingDeque

②阻塞隊(duì)列常見的幾種處理方式（并非所有方式都阻塞）

• 其中只有put和take方法時(shí)阻塞的
• 拋出異常：當(dāng)隊(duì)列滿時(shí)，如果再往隊(duì)列里插入元素，會(huì)拋出IllegalStateException（“Queuefull”）異常。當(dāng)隊(duì)列空時(shí)，從隊(duì)列里獲取元素會(huì)拋出NoSuchElementException異常。
• -返回特殊值：當(dāng)往隊(duì)列插入元素時(shí)，會(huì)返回元素是否插入成功，成功返回true。如果是移除方法，則是從隊(duì)列里取出一個(gè)元素，如果沒有則返回null。
• 一直阻塞：當(dāng)阻塞隊(duì)列滿時(shí)，如果生產(chǎn)者線程往隊(duì)列里put元素，隊(duì)列會(huì)一直阻塞生產(chǎn)者線程，直到隊(duì)列可用或者響應(yīng)中斷退出。當(dāng)隊(duì)列空時(shí)，如果消費(fèi)者線程從隊(duì)列里take元素，隊(duì)列會(huì)阻塞住消費(fèi)者線程，直到隊(duì)列不為空。
• 超時(shí)退出：當(dāng)阻塞隊(duì)列滿時(shí)，如果生產(chǎn)者線程往隊(duì)列里插入元素，隊(duì)列會(huì)阻塞生產(chǎn)者線程一段時(shí)間，如果超過了指定的時(shí)間，生產(chǎn)者線程就會(huì)退出。

③阻塞隊(duì)列簡單使用

• 三個(gè)線程添加數(shù)據(jù)
• 三個(gè)線程消費(fèi)數(shù)據(jù)

public class MyBlockingQueue {
    static ArrayBlockingQueue<String> abq = new ArrayBlockingQueue(3);
    public static void main(String[] args) {
        // 生產(chǎn)者線程
        for (int i = 0; i < 3; i++) {
            new Thread(() -> producer(), "producerThread" + i).start();
        }
        // 消費(fèi)者線程
        for (int i = 0; i < 3; i++) {
            new Thread(() -> consumer(), "consumerThread" + i).start();
        }
    }

    private static void consumer() {
        while (true) {
            try {
                String msg = abq.take();
                System.out.println(Thread.currentThread().getName() + " ->receive msg:" + msg);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private static void producer() {
        for (int i = 0; i < 100; i++) {
            try {
                abq.put("[" + i + "]");
                System.out.println(Thread.currentThread().getName() + " ->send msg:" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

執(zhí)行結(jié)果：

producerThread1 ->send msg:0
producerThread2 ->send msg:0
producerThread0 ->send msg:0
consumerThread1 ->receive msg:[0]
producerThread1 ->send msg:1
consumerThread2 ->receive msg:[0]
producerThread1 ->send msg:2
producerThread2 ->send msg:1
consumerThread1 ->receive msg:[0]
consumerThread0 ->receive msg:[1]
...

【2】Java 線程池

一、我們?yōu)槭裁葱枰狫ava 線程池？使用它的好處是什么？

①降低資源消耗。

通過重復(fù)利用已創(chuàng)建的線程降低線程創(chuàng)建和銷毀造成的消耗。

②提高響應(yīng)速度。

通常我們在Java程序中執(zhí)行一個(gè)任務(wù)的到結(jié)果分為以下步驟：

1.創(chuàng)建線程 ——> 2.執(zhí)行任務(wù) ——> 3.銷毀線程

當(dāng)任務(wù)到達(dá)時(shí)，任務(wù)可以不需要等到線程創(chuàng)建就能立即執(zhí)行。假設(shè)一個(gè)服務(wù)器完成一項(xiàng)任務(wù)所需時(shí)間為：T1 創(chuàng)建線程時(shí)間，T2 在線程中執(zhí)行任務(wù)的時(shí)間，T3 銷毀線程時(shí)間。 如果：T1 + T3 遠(yuǎn)大于 T2，則可以采用線程池，以提高服務(wù)器性能。線程池技術(shù)正是關(guān)注如何縮短或調(diào)整T1,T3時(shí)間的技術(shù)，從而提高服務(wù)器程序性能的。它把T1，T3分別安排在服務(wù)器程序的啟動(dòng)和結(jié)束的時(shí)間段或者一些空閑的時(shí)間段，這樣在服務(wù)器程序處理客戶請求時(shí)，不會(huì)有T1，T3的開銷了。

③提高線程的可管理性。

線程是稀缺資源，如果無限制地創(chuàng)建，不僅會(huì)消耗系統(tǒng)資源，還會(huì)降低系統(tǒng)的穩(wěn)定性，使用線程池可以進(jìn)行統(tǒng)一分配、調(diào)優(yōu)和監(jiān)控。

二、Java中主要提供了哪幾種線程的線程池？

Java中主要提供了一下4中線程池：

1、newCachedThreadPool：用來創(chuàng)建一個(gè)可以無限擴(kuò)大的線程池，適用于負(fù)載較輕的場景，執(zhí)行短期異步任務(wù)。（可以使得任務(wù)快速得到執(zhí)行，因?yàn)槿蝿?wù)時(shí)間執(zhí)行短，可以很快結(jié)束，也不會(huì)造成cpu過度切換）

2、newFixedThreadPool：創(chuàng)建一個(gè)固定大小的線程池，因?yàn)椴捎脽o界的阻塞隊(duì)列，所以實(shí)際線程數(shù)量永遠(yuǎn)不會(huì)變化，適用于負(fù)載較重的場景，對(duì)當(dāng)前線程數(shù)量進(jìn)行限制。（保證線程數(shù)可控，不會(huì)造成線程過多，導(dǎo)致系統(tǒng)負(fù)載更為嚴(yán)重）

3、newSingleThreadExecutor：創(chuàng)建一個(gè)單線程的線程池，適用于需要保證順序執(zhí)行各個(gè)任務(wù)。

4、newScheduledThreadPool：適用于執(zhí)行延時(shí)或者周期性任務(wù)。

三、線程類的繼承關(guān)系

• ThreadPoolExecutor 的類關(guān)系
• Executor是一個(gè)接口，它是Executor框架的基礎(chǔ)，它將任務(wù)的提交與任務(wù)的執(zhí)行分離開來。
• ExecutorService接口繼承了Executor，在其上做了一些shutdown()、submit()的擴(kuò)展，可以說是真正的線程池接口；
• AbstractExecutorService抽象類實(shí)現(xiàn)了ExecutorService接口中的大部分方法；
• ThreadPoolExecutor是線程池的核心實(shí)現(xiàn)類，用來執(zhí)行被提交的任務(wù)。
• ScheduledExecutorService接口繼承了ExecutorService接口，提供了帶"周期執(zhí)行"功能ExecutorService；
• ScheduledThreadPoolExecutor是一個(gè)實(shí)現(xiàn)類，可以在給定的延遲后運(yùn)行命令，或者定期執(zhí)行命令。ScheduledThreadPoolExecutor比Timer更靈活，功能更強(qiáng)大。

Executor——>ExecutorService——>AbstractExecutorService——>ThreadPoolExecutor

二中常用的幾種線程池都是源自ThreadPoolExecutor，所以我們來分析一下這個(gè)類

四、ThreadPoolExecutor參數(shù)的含義 corePoolSize

• corePoolSize

線程池中的核心線程數(shù)，當(dāng)提交一個(gè)任務(wù)時(shí)，線程池創(chuàng)建一個(gè)新線程執(zhí)行任務(wù)，直到當(dāng)前線程數(shù)等于corePoolSize；如果當(dāng)前線程數(shù)為corePoolSize，繼續(xù)提交的任務(wù)被保存到阻塞隊(duì)列中，等待被執(zhí)行；

如果執(zhí)行了線程池的prestartAllCoreThreads()方法，線程池會(huì)提前創(chuàng)建并啟動(dòng)所有核心線程。

• maximumPoolSize

線程池中允許的最大線程數(shù)。如果當(dāng)前阻塞隊(duì)列滿了，且繼續(xù)提交任務(wù)，則創(chuàng)建新的線程執(zhí)行任務(wù)，前提是當(dāng)前線程數(shù)小于maximumPoolSize

• keepAliveTime

線程空閑時(shí)的存活時(shí)間，即當(dāng)線程沒有任務(wù)執(zhí)行時(shí)，繼續(xù)存活的時(shí)間。默認(rèn)情況下，該參數(shù)只在線程數(shù)大于corePoolSize時(shí)才有用

• TimeUnit

keepAliveTime的時(shí)間單位

• workQueue

workQueue必須是BlockingQueue阻塞隊(duì)列。當(dāng)線程池中的線程數(shù)超過它的corePoolSize的時(shí)候，線程會(huì)進(jìn)入阻塞隊(duì)列進(jìn)行阻塞等待。通過workQueue，線程池實(shí)現(xiàn)了阻塞功能。

一般來說，我們應(yīng)該盡量使用有界隊(duì)列，因?yàn)槭褂脽o界隊(duì)列作為工作隊(duì)列會(huì)對(duì)線程池帶來如下影響。

1）當(dāng)線程池中的線程數(shù)達(dá)到corePoolSize后，新任務(wù)將在無界隊(duì)列中等待，因此線程池中的線程數(shù)不會(huì)超過corePoolSize。

2）由于1，使用無界隊(duì)列時(shí)maximumPoolSize將是一個(gè)無效參數(shù)。

3）由于1和2，使用無界隊(duì)列時(shí)keepAliveTime將是一個(gè)無效參數(shù)。

4）更重要的，使用無界queue可能會(huì)耗盡系統(tǒng)資源，有界隊(duì)列則有助于防止資源耗盡，同時(shí)即使使用有界隊(duì)列，也要盡量控制隊(duì)列的大小在一個(gè)合適的范圍。

• threadFactory

創(chuàng)建線程的工廠，通過自定義的線程工廠可以給每個(gè)新建的線程設(shè)置一個(gè)具有識(shí)別度的線程名，當(dāng)然還可以更加自由的對(duì)線程做更多的設(shè)置，比如設(shè)置所有的線程為守護(hù)線程。

Executors靜態(tài)工廠里默認(rèn)的threadFactory，線程的命名規(guī)則是“pool-數(shù)字-thread-數(shù)字”。

• RejectedExecutionHandler

線程池的飽和策略，當(dāng)阻塞隊(duì)列滿了，且沒有空閑的工作線程，如果繼續(xù)提交任務(wù)，必須采取一種策略處理該任務(wù)，線程池提供了4種策略：

（1）AbortPolicy：直接拋出異常，默認(rèn)策略；

（2）CallerRunsPolicy：用調(diào)用者所在的線程來執(zhí)行任務(wù)；

（3）DiscardOldestPolicy：丟棄阻塞隊(duì)列中靠最前的任務(wù)，并執(zhí)行當(dāng)前任務(wù)；

（4）DiscardPolicy：直接丟棄任務(wù)；

當(dāng)然也可以根據(jù)應(yīng)用場景實(shí)現(xiàn)RejectedExecutionHandler接口，自定義飽和策略，如記錄日志或持久化存儲(chǔ)不能處理的任務(wù)。

五、線程池工作流程（機(jī)制）

1. 如果當(dāng)前運(yùn)行的線程少于corePoolSize，則創(chuàng)建新線程來執(zhí)行任務(wù)（注意，執(zhí)行這一步驟需要獲取全局鎖）。

2. 如果運(yùn)行的線程等于或多于corePoolSize，則將任務(wù)加入BlockingQueue。

3. 如果無法將任務(wù)加入BlockingQueue（隊(duì)列已滿），則創(chuàng)建新的線程來處理任務(wù)。

4. 如果創(chuàng)建新線程將使當(dāng)前運(yùn)行的線程超出maximumPoolSize，任務(wù)將被拒絕，并調(diào)用RejectedExecutionHandler.rejectedExecution()方法。

六、關(guān)于兩種提交方法的比較

execute()方法用于提交不需要返回值的任務(wù)，所以無法判斷任務(wù)是否被線程池執(zhí)行成功。

submit()方法用于提交需要返回值的任務(wù)。線程池會(huì)返回一個(gè)future類型的對(duì)象，通過這個(gè)future對(duì)象可以判斷任務(wù)是否執(zhí)行成功，并且可以通過future的get()方法來獲取返回值，get()方法會(huì)阻塞當(dāng)前線程直到任務(wù)完成，而使用get（long timeout，TimeUnit unit）方法則會(huì)阻塞當(dāng)前線程一段時(shí)間后立即返回，這時(shí)候有可能任務(wù)沒有執(zhí)行完。