併發王者課-鉑金5:致勝良器-無處不在的“阻塞佇列”究竟是何面目
歡迎來到《併發王者課》,本文是該系列文章中的第18篇。
線上程的同步中,阻塞佇列是一個繞不過去的話題,它是同步器底層的關鍵。所以,我們在本文中將為你介紹阻塞佇列的基本原理,以瞭解它的工作機制和它在Java中的實現。本文稍微有點長,建議先了解大綱再細看章節。
一、阻塞佇列介紹
在生活中,相信你一定見過下圖的人山人海,也見過其中的秩序井然。混亂,是失控的開始。想想看,在沒有秩序的情況下,擁擠的人流蜂擁而上十分危險,輕則擠出一身臭汗,重則造成踩踏事故。而秩序,則讓情況免於混亂,排好隊大家都舒服。
面對人流,我們通過排隊解決混亂。而面對多執行緒,我們也通過佇列讓執行緒間免於混亂,這就是阻塞佇列為何而存在。
所謂阻塞佇列,你可以理解它是這樣的一種佇列:
- 當執行緒試著往佇列裡放資料時,如果它已經滿了,那麼執行緒將進入等待;
- 而當執行緒試著從佇列裡取資料時,如果它已經空了,那麼執行緒將進入等待。
下面這張圖展示了多執行緒是如何通過阻塞佇列進行協作的:
從圖中可以看到,對於阻塞佇列資料的讀寫並不侷限於單個執行緒,往往存在多個執行緒的競爭。
二、實現簡單的阻塞佇列
接下來我們先拋開JUC中複雜的阻塞佇列,來設計一個簡單的阻塞佇列,以瞭解它的核心思想。
在下面的阻塞佇列中,我們設計一個佇列queue
,並通過limit
欄位限定它的容量。enqueue()
方法用於向佇列中放入資料,如果佇列已滿則等待;而dequeue()
public class BlockingQueue { private final List<Object> queue = new LinkedList<>(); private final int limit; public BlockingQueue(int limit) { this.limit = limit; } public synchronized void enqueue(Object item) throws InterruptedException { while (this.queue.size() == this.limit) { print("佇列已滿,等待中..."); wait(); } this.queue.add(item); if (this.queue.size() == 1) { notifyAll(); } print(item, "已經放入!"); } public synchronized Object dequeue() throws InterruptedException { while (this.queue.size() == 0) { print("佇列空的,等待中..."); wait(); } if (this.queue.size() == this.limit) { notifyAll(); } Object item = this.queue.get(0); print(item, "已經拿到!"); return this.queue.remove(0); } public static void print(Object... args) { StringBuilder message = new StringBuilder(getThreadName() + ":"); for (Object arg : args) { message.append(arg); } System.out.println(message); } public static String getThreadName() { return Thread.currentThread().getName(); } }
定義lanLingWang
執行緒向佇列中放入資料,niumo
執行緒從佇列中取出資料。
public static void main(String[] args) {
BlockingQueue blockingQueue = new BlockingQueue(1);
Thread lanLingWang = new Thread(() -> {
try {
String[] items = { "A", "B", "C", "D", "E" };
for (String item: items) {
Thread.sleep(500);
blockingQueue.enqueue(item);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
lanLingWang.setName("蘭陵王");
Thread niumo = new Thread(() -> {
try {
while (true) {
blockingQueue.dequeue();
Thread.sleep(1000);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
lanLingWang.setName("蘭陵王");
niumo.setName("牛魔王");
lanLingWang.start();
niumo.start();
}
執行結果如下:
牛魔王:佇列空的,等待中...
蘭陵王:A已經放入!
牛魔王:A已經拿到!
蘭陵王:B已經放入!
牛魔王:B已經拿到!
蘭陵王:C已經放入!
蘭陵王:佇列已滿,等待中...
牛魔王:C已經拿到!
蘭陵王:D已經放入!
蘭陵王:佇列已滿,等待中...
牛魔王:D已經拿到!
蘭陵王:E已經放入!
牛魔王:E已經拿到!
牛魔王:佇列空的,等待中...
從結果中可以看到,設計的阻塞佇列已經可以有效工作,你可以仔細地品一品輸出的結果。當然,這個阻塞是極其簡單的,在下面一節中,我們將介紹Java中的阻塞佇列設計。
三、Java中的BlockingQueue
Java中的阻塞佇列有兩個核心介面:BlockingQueue和BlockingDeque,相關的介面實現設繼承關係如下圖所示。相比於上一節中我們自定義的阻塞佇列,Java中的實現要複雜很多。不過,你不必為此擔心,理解阻塞佇列最重要的是理解它的思想和實現的思路,況且Java中的實現其實很有意思,讀起來也比較輕鬆。
從圖中可以看出,BlockingQueue介面繼承了Queue介面和Collection介面,並有LinkedBlockingQueue和ArrayBlockingQueue兩種實現。這裡有個有意思的地方,繼承Queue介面很容易理解,可以為什麼要繼承Collection介面?先賣個關子,你可以思考一會,稍後會給出答案。
1. 核心方法
BlockingQueue中義了關於阻塞佇列所需要的一系列方法,它們彼此之間看起來很像,從表面上看不出明顯的差別。對於這些方法,你不必死記硬背,下圖的表格中將這些方法分為了A、B、C、D這四種類型,分類之後再去理解它們會容易很多:
型別 | A 丟擲異常 | B 返回特定值 | C 阻塞 | D 超時限定 |
---|---|---|---|---|
Insert | add(e) |
offer(e) |
put(e) |
offer(e, time, unit) |
Remove | remove() |
poll() |
take( ) |
poll(time, unit) |
Examine | Element() |
peek() |
-- | -- |
其中部分關鍵方法的解釋如下:
add(E e)
:在不違反容量限制的前提下,向佇列中插入資料。如果成功,返回true,否則丟擲異常;offer(E e)
:在不違反容量限制的前提下,向佇列中插入資料。如果成功,返回true
,否則返回false
;offer(E e, long timeout, TimeUnit unit)
:如果佇列中沒有足夠的空間,將等待一段時間;put(E e)
:在不違反容量限制的前提下,向佇列中插入資料。如果沒有足夠的空間,將進入等待;poll(long timeout, TimeUnit unit)
:從佇列的頭部獲取資料,並移除資料。如果沒有資料的話,將會等待指定的時間;take()
:從佇列的頭部獲取資料並移除。如果沒有可用資料,將進入等待
將這些方法填入前面的那張圖,它應該長這樣:
2. LinkedBlockingQueue
LinkedBlockingQueue實現了BlockingQueue介面,遵從先進先出(FIFO)的原則,提供了可選的有界阻塞佇列( Optionally Bounded )的能力,並且是執行緒安全的。
- 核心資料結構
int capacity
: 設定佇列容量;Node<E> head
: 佇列的頭部元素;Node<E> last
: 佇列的尾部元素;AtomicInteger count
: 佇列中元素的總數統計。
LinkedBlockingQueue的資料結構並不複雜,不過需要注意的是,資料結構中並不包含List,僅有head
和last
兩個Node,設計上比較巧妙。
- 核心構造
LinkedBlockingQueue()
: 空構造;LinkedBlockingQueue(int capacity)
: 指定容量構造。
- 執行緒安全性
ReentrantLock takeLock
: 獲取元素時的鎖;ReentrantLock putLock
: 寫入元素時的鎖。
注意,LinkedBlockingQueue有兩把鎖,讀取和寫入的鎖是分離的!這和下面的ArrayBlockingQueue並不相同。
下面截取了LinkedBlockingQueue中讀寫的部分程式碼,值得你仔細品一品。品的時候,要重點關注兩把鎖的使用和讀寫時資料結構是如何變化的。
- 佇列插入示例程式碼分析
public boolean add(E e) {
addLast(e);
return true;
}
public void addLast(E e) {
if (!offerLast(e))
throw new IllegalStateException("Deque full");
}
public boolean offerFirst(E e) {
if (e == null) throw new NullPointerException();
Node<E> node = new Node<E>(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
return linkFirst(node);
} finally {
lock.unlock();
}
}
- 佇列讀取示例程式碼分析
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
return pollFirst(timeout, unit);
}
public E pollFirst(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
E x;
while ( (x = unlinkFirst()) == null) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
return x;
} finally {
lock.unlock();
}
}
最後說下LinkedBlockingQueue為什麼要繼承Collection介面。我們知道,Collection介面有remove()
這樣的移除方法,而這些方法在佇列中也是有使用場景的。比如,你把一個數據錯誤地放入了佇列,或者你需要移除已經失效的資料,那麼Collection的一些方法就派上了用場。
3. ArrayBlockingQueue
ArrayBlockingQueue是BlockingQueue介面的另外一種實現,它與LinkedBlockingQueue在設計目標上的的關鍵不同,在於它是有界的。
-
核心資料結構
Object[] items
: 佇列元素集合;int takeIndex
: 下次獲取資料時的索引位置;int putIndex
: 下次寫入資料時的索引位置;int count
: 佇列總量計數。
從資料結構中可以看出,ArrayBlockingQueue使用的是陣列,而陣列是有界的。
-
核心構造
ArrayBlockingQueue(int capacity)
: 限定容量的構造;ArrayBlockingQueue(int capacity, boolean fair)
: 限定容量和公平性,預設是不公平的;ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c)
:帶有初始化佇列元素的構造。
-
執行緒安全性
ReentrantLock lock
:佇列讀取和寫入的鎖。
在讀寫鎖方面,前面已經說過,LinkedBlockingQueue和ArrayBlockingQueue是不同的,ArrayBlockingQueue只有一把鎖,讀寫用的都是它。
- 佇列寫入示例程式碼分析
public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
return false;
else {
enqueue(e);
return true;
}
} finally {
lock.unlock();
}
}
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;
notEmpty.signal();
}
下面截取了ArrayBlockingQueue中讀寫的部分程式碼,值得你仔細品一品。品的時候,要重點關注讀寫鎖的使用和讀寫時資料結構是如何變化的。
- 佇列讀取示例程式碼分析
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
return dequeue();
} finally {
lock.unlock();
}
}
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal();
return x;
}
四、Java中的BlockingDeque
在Java中,BlockingDeque與BlockingQueue是一對孿生兄弟似的存在,它們長得實在太像了,不注意的話很容易混淆。
但是,BlockingDeque與BlockingQueue核心不同在於,BlockingQueue只能夠從尾部寫入、從頭部讀取,使用上很有限制。而BlockingDeque則支援從任意端讀寫,在讀寫時可以指定頭部和尾部,豐富了阻塞佇列的使用場景。
1. 核心方法
相較於BlockingQueue,BlockingDeque的方法顯然要更豐富一些,畢竟它支援了雙端的讀寫。但是,豐富歸豐富,在型別上仍然和BlockingQueue是一致的,你仍然可以參考上面的A、B、C、D四種類型來分類理解。為了節約篇幅,我們這裡就不再羅列,只選取了其中的部分方法作了解釋:
add(E e)
:在不違反容量限制的前提下,在對列的尾部插入資料;addFirst(E e)
:從頭部插入資料,容量不夠就拋錯;addLast(E e)
:從尾部插入資料,容量不夠就拋錯;getFirst()
:從頭部讀取資料;getLast()
:從尾部讀取資料,但不會移除資料;offer(E e)
:寫入資料;offerFirst(E e)
:從頭部寫入資料。
將BlockingDeue放入前面的那張圖,就是這樣:
2. LinkedBlockingDeue
LinkedBlockingDeue是BlockingDeque的核心實現。
-
核心資料結構
int capacity
:容量設定;Node<E> head
:佇列頭部;Node<E> last
:佇列尾部;int count
:佇列計數。
-
核心構造
LinkedBlockingDeque()
: 空的構造;LinkedBlockingDeque(int capacity)
: 指定容量的構造;LinkedBlockingDeque(Collection<? extends E> c)
:構造時初始化佇列。
-
執行緒安全性
ReentrantLock lock
:讀寫鎖。注意,讀寫用的是同一把鎖。
下面截取了LinkedBlockingDeue中讀寫的部分程式碼,值得你仔細品一品。品的時候,要重點關注讀寫鎖的使用和讀寫時資料結構是如何變化的
- 佇列插入示例程式碼分析
public void addFirst(E e) {
if (!offerFirst(e))
throw new IllegalStateException("Deque full");
}
public boolean offerFirst(E e) {
if (e == null) throw new NullPointerException();
Node < E > node = new Node < E > (e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
return linkFirst(node);
} finally {
lock.unlock();
}
}
- 佇列讀取示例程式碼分析
public E pollFirst() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return unlinkFirst();
} finally {
lock.unlock();
}
}
小結
以上就是關於阻塞佇列的全部內容,相較於前面的系列文章,這次的內容明顯增加了很多。看起來很簡單,但是不要小瞧它。理解阻塞佇列,首先要理解它所要解決的問題,以及它的介面設計。介面的設計往往表示的是它所提供的核心能力,所以理解了介面的設計,就成功了一半。
在Java中,從介面層面,阻塞佇列分為BlockingQueue和BlockingDeque的兩大類,其主要差異在於雙端讀寫的限制不同。其中,BlockingQueue有LinkedBlockingDeue和ArrayBlockingQueue兩種關鍵實現,而BlockingDeque則有LinkedBlockingDeue實現。
正文到此結束,恭喜你又上了一顆星✨
夫子的試煉
- 從資料機構、佇列的初始化、鎖、效能等方面比較LinkedBlockingDeue和ArrayBlockingQueue的不同。
延伸閱讀與參考資料
關於作者
關注公眾號【技術八點半】,及時獲取文章更新。傳遞有品質的技術文章,記錄平凡人的成長故事,偶爾也聊聊生活和理想。早晨8:30推送作者品質原創,晚上20:30推送行業深度好文。
如果本文對你有幫助,歡迎點贊、關注、監督,我們一起從青銅到王者。