前言
在上一章節(jié)，我們了解到了多線程的一些基礎(chǔ)知識點(diǎn)——多線程的實(shí)現(xiàn)、CAS 是什么以及 AQS 的工作流程，接下來，在這章節(jié)中，我會(huì)通過有關(guān)于應(yīng)用性的多線程面試題的內(nèi)容，來給大家介紹相關(guān)的多線程知識點(diǎn)。
下面我們通過 2 道常問的面試題進(jìn)行相關(guān)知識點(diǎn)的學(xué)習(xí)：
1. BlockingQueue(阻塞隊(duì)列)原理是什么？手寫阻塞隊(duì)列的 take 和 put 方法
2. 如何定義線程池初始化參數(shù)？使用無界隊(duì)列的線程池會(huì)導(dǎo)致內(nèi)存飆升嗎？
這里面談到的阻塞隊(duì)列和 AQS 的阻塞隊(duì)列不是同一個(gè)東西，該文章談到的阻塞隊(duì)列全都是關(guān)乎于BlockingQueue接口
1. BlockingQueue 原理是什么？手寫阻塞隊(duì)列的 take 和 put 方法
阻塞隊(duì)列，從字面意義上看，就是一個(gè)帶阻塞功能的隊(duì)列，那么它會(huì)在什么時(shí)候出現(xiàn)阻塞呢？
- 如果隊(duì)列為空，那么想要隊(duì)列出隊(duì)即執(zhí)行 take() 方法的線程會(huì)被阻塞。
- 如果隊(duì)列已滿，那么想要隊(duì)列入隊(duì)即執(zhí)行 put() 方法的線程會(huì)被阻塞。
在上一個(gè)章節(jié)我們了解到了 AQS 以及它的一個(gè)實(shí)現(xiàn)類 ReentrantLock，而在 Java 的阻塞隊(duì)列當(dāng)中，則是借助了 ReentrantLock 類實(shí)現(xiàn)了阻塞的功能，除此之外，為了實(shí)現(xiàn)鎖的細(xì)膩度，還額外使用了另一個(gè)接口—— Condition。
在介紹阻塞隊(duì)列之前，我們先來了解下 AQS 中的 Conditon 接口。傳統(tǒng)多線程方式當(dāng)中，我們通常會(huì)使用 synchronized 關(guān)鍵字配合 Object 類的 wait()/notifyAll() 方法實(shí)現(xiàn)多線程的協(xié)同工作，相同的，為了配合 AQS ，AQS 自身通過內(nèi)部類實(shí)現(xiàn) Condition 接口來完成多線程的協(xié)同。
上文提到了一個(gè)鎖的細(xì)膩度，可能有的人無法理解這個(gè)鎖的細(xì)膩度是什么，那么接下來我便簡單舉一個(gè)例子。
假如我們使用 synchronized 關(guān)鍵字配合 Object 類的 wait()/notifyAll() 來實(shí)現(xiàn)一個(gè)生產(chǎn)者和消費(fèi)者模式的時(shí)候，由于你無法指定喚醒哪一類線程，因此只能使用 notifyAll() 方法將生產(chǎn)者和消費(fèi)者等待的線程都喚醒，這樣的壞處便是無需被喚醒的線程被喚醒了，導(dǎo)致了性能的浪費(fèi)。而所謂的細(xì)膩度，便是針對性的喚醒線程，每次喚醒都不會(huì)導(dǎo)致性能的浪費(fèi)，例如我生產(chǎn)者只能喚醒消費(fèi)者，而消費(fèi)者也只能喚醒生產(chǎn)者，這樣便不會(huì)出現(xiàn)無關(guān)線程被喚醒又重新進(jìn)入阻塞的問題。
Condition 類就是生成了一個(gè)與阻塞隊(duì)列隔離的條件隊(duì)列，每個(gè) Condition 對象都相當(dāng)于是一個(gè)隊(duì)列，里面由多個(gè) Node 元素以單向鏈表的形式組合而成，每個(gè)隊(duì)列元素出隊(duì)后都會(huì)加入到阻塞隊(duì)列的隊(duì)尾，我們通過一張圖來簡單了解下 Condition：
可以看出，條件隊(duì)列也是有序的，即在條件隊(duì)列中阻塞的線程如果沒有被中斷，都會(huì)根據(jù)隊(duì)列的順序進(jìn)行有序的喚醒出隊(duì)，即每次喚醒的都是在隊(duì)列中等待最久的線程。
有個(gè)注意點(diǎn)，Condition 對象的 await()/signal() 方法必須要線程獲取到鎖只能才會(huì)正常執(zhí)行，否則會(huì)拋出異常，即在沒有獲取鎖的環(huán)境調(diào)用這些方法，便會(huì)拋出 IllegalMonitorStateException 異常，這個(gè)和 Object 對象的 wait()/notify() 方法一樣。
接下來我簡單描述下 Condtion 條件隊(duì)列的 入隊(duì) await() / 出隊(duì) signal() 執(zhí)行流程。
await() 方法：
- 將當(dāng)前線程包裝為一個(gè) waitStatus 字段為 CONDITION(-2) 的 node 對象并加入到條件隊(duì)列尾部。
- 記錄當(dāng)前鎖的 state 字段值，并將當(dāng)前鎖的狀態(tài)值改為 0 (重入鎖也一樣改為0)。
- 通過 LockSupport.park() 方法進(jìn)入阻塞。
signal() 方法
- 斷開與條件隊(duì)列的關(guān)系，即將當(dāng)前 node 對象的 nextWaiter 指針字段設(shè)置為 null。
- 通過 CAS 將當(dāng)前 node 對象的 waitStatus 字段設(shè)置為 0，并將設(shè)置好的節(jié)點(diǎn)通過自旋 CAS 加入到阻塞隊(duì)列的隊(duì)尾
- 將當(dāng)前 node 對象的前驅(qū)節(jié)點(diǎn)的 waitStatus 字段設(shè)置為 SIGNAL(-1)，代表當(dāng)前節(jié)點(diǎn)需要這個(gè)節(jié)點(diǎn)進(jìn)行喚醒；如果這個(gè)前驅(qū)節(jié)點(diǎn)處于取消狀態(tài)，那么直接喚醒當(dāng)前 node 對象的線程，讓其去競爭鎖，相當(dāng)于一個(gè)新的線程進(jìn)入 AQS 一樣，接下來的步驟讀者可以參考上一篇 AQS 的運(yùn)行流程。
了解完 AQS(ReentrantLock)/Conditon 的內(nèi)容，接下來再看阻塞隊(duì)列就很簡單了，其實(shí)阻塞隊(duì)列的原理就是：Lock 配合多個(gè) Condtion 進(jìn)行阻塞控制，配合多個(gè) Conodtion 只是為了針對性喚醒來減少性能的損耗。開發(fā)者不需要擔(dān)心什么時(shí)候需要阻塞隊(duì)列，只需要往隊(duì)列存入/取出數(shù)據(jù)即可，因?yàn)橐磺杏嘘P(guān)阻塞的功能都在內(nèi)部進(jìn)行實(shí)現(xiàn)。
接下來，我通過一個(gè)簡單的例子來實(shí)現(xiàn)阻塞隊(duì)列的 take()/put(E e) 方法。
public class BlockQueue<E>{
//一個(gè) Lock 對應(yīng)多個(gè) Condition
Lock lock = new ReentrantLock();
Condition producer = lock.newCondition();
Condition consumer = lock.newCondition();
//有界隊(duì)列，即隊(duì)列有容量限制
int size;
//這里通過一個(gè)List存儲(chǔ)隊(duì)列元素
List<E> queue;
public BlockQueue(int size){
this.size = size;
queue = new ArrayList<>(size);
}
public void put(E e) throws InterruptedException {
try {
lock.lock();
//如果容量已滿，對生產(chǎn)者線程進(jìn)行阻塞
while(size == queue.size()){
producer.await();
}
queue.add(e);
//有新的商品生產(chǎn)了，可以喚醒被阻塞的消費(fèi)者進(jìn)行消費(fèi)
consumer.signal();
}finally {
lock.unlock();
}
}
public E take() throws InterruptedException {
try{
lock.lock();
//沒有商品便阻塞消費(fèi)者
while (queue.size() == 0){
consumer.await();
}
E e = queue.get(0);
queue.remove(0);
//商品被消費(fèi)了一個(gè)，可以喚醒生產(chǎn)者繼續(xù)生產(chǎn)
producer.signal();
return e;
}finally {
lock.unlock();
}
}
public static void main(String[] args) {
BlockQueue<Object> queue = new BlockQueue<>(2);
Thread producer = new Thread(()->{
while (true){
try {
queue.put(new Object());
System.out.println(Thread.currentThread().getName() + " 生產(chǎn)了一個(gè)商品");
} catch (InterruptedException e) {}
}
});
producer.setName("producer");
Thread consumer = new Thread(()->{
while (true){
try {
Object take = queue.take();
System.out.println(Thread.currentThread().getName() + " 消費(fèi)了一個(gè)商品");
//每1秒才會(huì)有一個(gè)商品被消費(fèi)
try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
} catch (InterruptedException e) {}
}
});
consumer.setName("consumer");
producer.start();
consumer.start();
}
}
從上面代碼可以看出，阻塞隊(duì)列的核心就在如何正確的使用 Lock 和 Condition，當(dāng)然，Java 提供的阻塞隊(duì)列的功能還不僅僅如此，但由于篇幅原因，這里不在闡述過多內(nèi)容，有興趣的讀者可以自行查閱相關(guān)資料。
2. 如何定義線程池初始化參數(shù)？使用無界隊(duì)列的線程池會(huì)導(dǎo)致內(nèi)存飆升嗎？
在了解如何定義初始化參數(shù)前，我們通過一個(gè)圖來了解 jdk5 提供的線程池的構(gòu)造參數(shù)：
從上圖可以看出，線程池的構(gòu)造參數(shù)分為三大類，分別是線程相關(guān)、時(shí)間相關(guān)、任務(wù)相關(guān)，下面我們主要根據(jù)線程相關(guān)以及任務(wù)相關(guān)進(jìn)行分析：
- 線程相關(guān)我們只需要根據(jù)業(yè)務(wù)的類型進(jìn)行最大線程數(shù)的設(shè)值
- 任務(wù)相關(guān)首先是禁止使用無界阻塞隊(duì)列，其次便是根據(jù)業(yè)務(wù)實(shí)現(xiàn)自定義的拒絕策略，例如任務(wù)允許丟失，則可以通過日志簡單記錄該任務(wù)然后丟棄，如果任務(wù)不允許丟失，可以考慮新建線程去處理（如果服務(wù)器性能夠強(qiáng)大），或者記錄詳細(xì)的任務(wù)信息，等線程池空閑的時(shí)候在進(jìn)行處理。
了解完構(gòu)造參數(shù)的定義規(guī)則，接下來我們來看看為什么線程池禁止使用無界隊(duì)列。
回到問題，該問題的答案便是存在使用無界隊(duì)列會(huì)導(dǎo)致內(nèi)存飆升的場景，至于為什么，這個(gè)便涉及到線程池的執(zhí)行過程，因此下面我們通過了解線程池的執(zhí)行過程來詳細(xì)解答這個(gè)問題。
- 首先將任務(wù)給常駐線程處理
- 如果常駐線程來不及處理，則將任務(wù)放入阻塞隊(duì)列
- 如果阻塞隊(duì)列滿了并且線程數(shù)量少于maximumPoolSize，則將線程數(shù)擴(kuò)展至maximumPoolSize
- 如果仍然處理不完，阻塞隊(duì)列滿了，則進(jìn)行拒絕策略。
- 如果任務(wù)被快速處理完，除常駐線程外的其他線程，在經(jīng)過keepAliveTime(unit)時(shí)間后自動(dòng)銷毀
從線程池執(zhí)行流程的第三步可以看出，創(chuàng)建非常駐線程的前提是阻塞隊(duì)列已滿，但是對于無界隊(duì)列來說（上限為Integer.MAX_VALUE），幾乎不可能出現(xiàn)阻塞隊(duì)列滿的情況，如果常駐線程無法快速執(zhí)行任務(wù)，任務(wù)便會(huì)一直往隊(duì)列中添加積累，最終導(dǎo)致內(nèi)存飆升的問題。
總結(jié)
在多線程方面，首先我們得先了解多線程的實(shí)現(xiàn)，實(shí)現(xiàn)完多線程后便考慮如何管理多線程的競爭，多線程的競爭可以用到基于 CAS 和 AQS 實(shí)現(xiàn)的鎖和阻塞隊(duì)列，最后通過線程池來進(jìn)行任務(wù)分配和線程的回收利用，如果讀者在看這兩章節(jié)內(nèi)容時(shí)理不清楚順序的話，可以基于這段總結(jié)在回看一次。
接下來，我們回到上面的兩個(gè)問題
1. BlockingQueue(阻塞隊(duì)列)原理是什么？手寫阻塞隊(duì)列的 take 和 put 方法
2. 如何定義線程池初始化參數(shù)？使用無界隊(duì)列的線程池會(huì)導(dǎo)致內(nèi)存飆升嗎？
如果你們可以很流暢的回答這些問題，那么恭喜你，該章節(jié)的內(nèi)容已經(jīng)全部掌握，如果不行，希望可以回到對應(yīng)問題講解的地方，或者對某個(gè)不了解的點(diǎn)進(jìn)行額外的知識搜索，盡量用自己組織的語言回答這些問題。
