並發王者課-鉑金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推送行業深度好文。

如果本文對你有幫助，歡迎點贊、關注、監督，我們一起從青銅到王者。

Tags:

VirMach 便宜 VPS