常用阻塞隊列 BlockingQueue 有哪些？
- 2020 年 3 月 5 日
- 筆記
為什麼要使用阻塞隊列
之前，介紹了一下 ThreadPoolExecutor 的各參數的含義（並發編程之執行緒池ThreadPoolExecutor），其中有一個 BlockingQueue，它是一個阻塞隊列。那麼，小夥伴們有沒有想過，為什麼此處的執行緒池要用阻塞隊列呢？
我們知道隊列是先進先出的。當放入一個元素的時候，會放在隊列的末尾，取出元素的時候，會從隊頭取。那麼，當隊列為空或者隊列滿的時候怎麼辦呢。
這時，阻塞隊列，會自動幫我們處理這種情況。
當阻塞隊列為空的時候，從隊列中取元素的操作就會被阻塞。當阻塞隊列滿的時候，往隊列中放入元素的操作就會被阻塞。
而後，一旦空隊列有數據了，或者滿隊列有空餘位置時，被阻塞的執行緒就會被自動喚醒。
這就是阻塞隊列的好處，你不需要關心執行緒何時被阻塞，也不需要關心執行緒何時被喚醒，一切都由阻塞隊列自動幫我們完成。我們只需要關注具體的業務邏輯就可以了。
而這種阻塞隊列經常用在生產者消費者模式中。（可參看：面試官讓我手寫一個生產者消費者模式）
常用的阻塞隊列
那麼，一般我們用到的阻塞隊列有哪些呢。下面，通過idea的類圖，列出來常用的阻塞隊列，然後一個一個講解（不懂怎麼用的，可以參考這篇文章：怎麼用IDEA快速查看類圖關係）。
阻塞隊列中，所有常用的方法都在 BlockingQueue 介面中定義。如
插入元素的方法： put，offer，add。移除元素的方法： remove，poll，take。
它們有四種不同的處理方式，第一種是在失敗時拋出異常，第二種是在失敗時返回特殊值，第三種是一直阻塞當前執行緒，最後一種是在指定時間內阻塞，否則返回特殊值。（以上特殊值，是指在插入元素時，失敗返回false，在取出元素時，失敗返回null）
| 拋異常 | 特殊值 | 阻塞 | 超時 | |
|---|---|---|---|---|
| 插入 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
| 移除 | remove() | poll() | take() | poll(time,unit) |
1） ArrayBlockingQueue
這是一個由數組結構組成的有界阻塞隊列。首先看下它的構造方法，有三個。
第一個可以指定隊列的大小，第二個還可以指定隊列是否公平，不指定的話，默認是非公平。它是使用 ReentrantLock 的公平鎖和非公平鎖實現的（後續講解AQS時，會詳細說明）。
簡單理解就是，ReentrantLock 內部會維護一個有先後順序的等待隊列，假如有五個任務一起過來，都被阻塞了。如果是公平的，則等待隊列中等待最久的任務就會先進入阻塞隊列。如果是非公平的，那麼這五個執行緒就需要搶鎖，誰先搶到，誰就先進入阻塞隊列。
第三個構造方法，是把一個集合的元素初始化到阻塞隊列中。
另外，ArrayBlockingQueue 沒有實現讀寫分離，也就是說，讀和寫是不能同時進行的。因為，它讀寫時用的是同一把鎖，如下圖所示：
2) LinkedBlockingQueue
這是一個由鏈表結構組成的有界阻塞隊列。它的構造方法有三個。
可以看到和 ArrayBlockingQueue 的構造方法大同小異，不過是，LinkedBlockingQueue 可以不指定隊列的大小，默認值是 Integer.MAX_VALUE 。
但是，最好不要這樣做，建議指定一個固定大小。因為，如果生產者的速度比消費者的速度大的多的情況下，這會導致阻塞隊列一直膨脹，直到系統記憶體被耗盡（此時，還沒達到隊列容量的最大值）。
此外，LinkedBlockingQueue 實現了讀寫分離，可以實現數據的讀和寫互不影響，這在高並發的場景下，對於效率的提高無疑是非常巨大的。
3） SynchronousQueue
這是一個沒有緩衝的無界隊列。什麼意思，看一下它的 size 方法：
總是返回 0 ，因為它是一個沒有容量的隊列。
當執行插入元素的操作時，必須等待一個取出操作。也就是說，put元素的時候，必須等待 take 操作。
那麼，有的同學就好奇了，這沒有容量，還叫什麼隊列啊，這有什麼意義呢。
我的理解是，這適用於並發任務不大，而且生產者和消費者的速度相差不多的場景下，直接把生產者和消費者對接，不用經過隊列的入隊出隊這一系列操作。所以，效率上會高一些。
可以去查看一下 Excutors.newCachedThreadPool 方法用的就是這種隊列。
這個隊列有兩個構造方法，用於傳入是公平還是非公平，默認是非公平。
4）PriorityBlockingQueue
這是一個支援優先順序排序的無界隊列。有四個構造方法：
可以指定初始容量大小（注意初始容量並不代表最大容量），或者不指定，默認大小為 11。也可以傳入一個比較器，把元素按一定的規則排序，不指定比較器的話，默認是自然順序。
PriorityBlockingQueue 是基於二叉樹最小堆實現的，每當取元素的時候，就會把優先順序最高的元素取出來。我們測試一下：
public class Person { private int id; private String name; public int getId() { return id; } public void setId(int id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } @Override public String toString() { return "Person{" + "id=" + id + ", name='" + name + ''' + '}'; } public Person(int id, String name) { this.id = id; this.name = name; } public Person() { } } public class QueueTest { public static void main(String[] args) throws InterruptedException { PriorityBlockingQueue<Person> priorityBlockingQueue = new PriorityBlockingQueue<>(1, new Comparator<Person>() { @Override public int compare(Person o1, Person o2) { return o1.getId() - o2.getId(); } }); Person p2 = new Person(7, "李四"); Person p1 = new Person(9, "張三"); Person p3 = new Person(6, "王五"); Person p4 = new Person(2, "趙六"); priorityBlockingQueue.add(p1); priorityBlockingQueue.add(p2); priorityBlockingQueue.add(p3); priorityBlockingQueue.add(p4); //由於二叉樹最小堆實現，用這種方式直接列印元素，不能保證有序 System.out.println(priorityBlockingQueue); System.out.println(priorityBlockingQueue.take()); System.out.println(priorityBlockingQueue); System.out.println(priorityBlockingQueue.take()); System.out.println(priorityBlockingQueue); } }列印結果：
[Person{id=2, name='趙六'}, Person{id=6, name='王五'}, Person{id=7, name='李四'}, Person{id=9, name='張三'}] Person{id=2, name='趙六'} [Person{id=6, name='王五'}, Person{id=9, name='張三'}, Person{id=7, name='李四'}] Person{id=6, name='王五'} [Person{id=7, name='李四'}, Person{id=9, name='張三'}]可以看到，第一次取出的是 id 最小值 2， 第二次取出的是 6 。
5）DelayQueue
這是一個帶有延遲時間的無界阻塞隊列。隊列中的元素，只有等延時時間到了，才能取出來。此隊列一般用於過期數據的刪除，或任務調度。以下，模擬一下定長時間的數據刪除。
首先定義數據元素，需要實現 Delayed 介面，實現 getDelay 方法用於計算剩餘時間，和 CompareTo方法用於優先順序排序。
public class DelayData implements Delayed { private int id; private String name; //數據到期時間 private long endTime; private TimeUnit timeUnit = TimeUnit.MILLISECONDS; public int getId() { return id; } public void setId(int id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public long getEndTime() { return endTime; } public void setEndTime(long endTime) { this.endTime = endTime; } public DelayData(int id, String name, long endTime) { this.id = id; this.name = name; //需要把傳入的時間endTime 加上當前系統時間，作為數據的到期時間 this.endTime = endTime + System.currentTimeMillis(); } public DelayData() { } @Override public long getDelay(TimeUnit unit) { return this.endTime - System.currentTimeMillis(); } @Override public int compareTo(Delayed o) { return o.getDelay(this.timeUnit) - this.getDelay(this.timeUnit) < 0 ? 1: -1; } } 模擬三條數據，分別設置不同的過期時間：
public class ProcessData { public static void main(String[] args) throws InterruptedException { DelayQueue<DelayData> delayQueue = new DelayQueue<>(); DelayData a = new DelayData(5, "A", 5000); DelayData b = new DelayData(8, "B", 8000); DelayData c = new DelayData(2, "C", 2000); delayQueue.add(a); delayQueue.add(b); delayQueue.add(c); System.out.println("開始計時時間:" + System.currentTimeMillis()); for (int i = 0; i < 3; i++) { DelayData data = delayQueue.take(); System.out.println("id:"+data.getId()+"，數據:"+data.getName()+"被移除，當前時間:"+System.currentTimeMillis()); } } }最後結果：
開始計時時間:1583333583216 id:2，數據:C被移除，當前時間:1583333585216 id:5，數據:A被移除，當前時間:1583333588216 id:8，數據:B被移除，當前時間:1583333591216可以看到，數據是按過期時間長短，按順序移除的。C的時間最短 2 秒，然後過了 3 秒 A 也過期，再過 3 秒，B 過期。
