## 22 ArrayBlockingQueue 源碼解析 ## 引導語 本小節我們來介紹本章最后一個隊列：ArrayBlockingQueue。按照字面翻譯，中文叫做數組阻塞隊列，從名稱上看，我們就比較清楚此阻塞隊列底層使用的是數組。一說到數組，大家可能會想到 ArrayList 和 HashMap，舉新增場景來說 ArrayList 通過 size ++ 找到新增的數組下標位置，HashMap 通過 hash 算法計算出下標位置，那么 ArrayBlockingQueue 是不是也是這兩種方法呢？都不是，ArrayBlockingQueue 使用的是一種非常奇妙的方式，我們一起拭目以待。 全文為了方便說明，隊頭的說法就是數組頭，隊尾的說法就是數組尾。 ### 1 整體架構 我們從類注釋上可以得到一些有用的信息： #### 1.1 類注釋 1. 有界的阻塞數組，容量一旦創建，后續大小無法修改； 2. 元素是有順序的，按照先入先出進行排序，從隊尾插入數據數據，從隊頭拿數據； 3. 隊列滿時，往隊列中 put 數據會被阻塞，隊列空時，往隊列中拿數據也會被阻塞。 從類注釋上可以看出 ArrayBlockingQueue 和一般的數組結構的類不太一樣，是不能夠動態擴容的，如果隊列滿了或者空時，take 和 put 都會被阻塞。 #### 1.2 數據結構 ``` // 隊列存放在 object 的數組里面 // 數組大小必須在初始化的時候手動設置，沒有默認大小 final Object[] items; // 下次拿數據的時候的索引位置 int takeIndex; // 下次放數據的索引位置 int putIndex; // 當前已有元素的大小 int count; // 可重入的鎖 final ReentrantLock lock; // take的隊列 private final Condition notEmpty; // put的隊列 private final Condition notFull; ``` 以上代碼有兩個關鍵的字段，takeIndex 和 putIndex，分別表示下次拿數據和放數據的索引位置。所以說在新增數據和拿數據時，都無需計算，就能知道應該新增到什么位置，應該從什么位置拿數據。 ### 2 初始化 初始化時，有兩個重要的參數：數組的大小、是否是公平，源碼如下： ``` public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity <= 0) throw new IllegalArgumentException(); this.items = new Object[capacity]; lock = new ReentrantLock(fair); // 隊列不為空 Condition，在 put 成功時使用 notEmpty = lock.newCondition(); // 隊列不滿 Condition，在 take 成功時使用 notFull = lock.newCondition(); } ``` 從源碼中我們可以看出，第二個參數是否公平，主要用于讀寫鎖是否公平，如果是公平鎖，那么在鎖競爭時，就會按照先來先到的順序，如果是非公平鎖，鎖競爭時隨機的。 對于鎖公平和非公平，我們舉個例子：比如說現在隊列是滿的，還有很多線程執行 put 操作，必然會有很多線程阻塞等待，當有其它線程執行 take 時，會喚醒等待的線程，如果是公平鎖，會按照阻塞等待的先后順序，依次喚醒阻塞的線程，如果是非公平鎖，會隨機喚醒沉睡的線程。 所以說隊列滿很多線程執行 put 操作時，如果是公平鎖，數組元素新增的順序就是阻塞線程被釋放的先后順序，是有順序的，而非公平鎖，由于阻塞線程被釋放的順序是隨機的，所以元素插入到數組的順序也就不會按照插入的順序了。 隊列空時，也是一樣的道理。 ArrayBlockingQueue 通過鎖的公平和非公平，輕松實現了數組元素的插入順序的問題。如果要實現這個功能，你會怎么做呢？會想到利用鎖的功能么？其實這種思想我們在文中多次提到，當我們需要完成一件事情時，首先看看已有的 API 能不能滿足，如果可以的話，通過繼承和組合的方式來實現，ArrayBlockingQueue 就是組合了鎖的功能。 初始化時，如果給定了原始數據的話，一定要注意原始數據的大小一定要小于隊列的容量，否則會拋異常，如下圖所示：  我們寫了一個 demo，報錯如下：  ### 3 新增數據 數據新增都會按照 putIndex 的位置進行新增，源碼如下： ``` // 新增，如果隊列滿，無限阻塞 public void put(E e) throws InterruptedException { // 元素不能為空 checkNotNull(e); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { // 隊列如果是滿的，就無限等待 // 一直等待隊列中有數據被拿走時，自己被喚醒 while (count == items.length) notFull.await(); enqueue(e); } finally { lock.unlock(); } } private void enqueue(E x) { // assert lock.getHoldCount() == 1; 同一時刻只能一個線程進行操作此方法 // assert items[putIndex] == null; final Object[] items = this.items; // putIndex 為本次插入的位置 items[putIndex] = x; // ++ putIndex 計算下次插入的位置 // 如果下次插入的位置，正好等于隊尾，下次插入就從 0 開始 if (++putIndex == items.length) putIndex = 0; count++; // 喚醒因為隊列空導致的等待線程 notEmpty.signal(); } ``` 從源碼中，我們可以看出，其實新增就兩種情況： 1. 本次新增的位置居中，直接新增，下圖演示的是 putIndex 在數組下標為 5 的位置，還不到隊尾，那么可以直接新增，計算下次新增的位置應該是 6；  2. 新增的位置到隊尾了，那么下次新增時就要從頭開始了，示意圖如下：  上面這張圖演示的就是這行代碼：if (++putIndex == items.length) putIndex = 0; 可以看到當新增到隊尾時，下次新增會重新從隊頭重新開始。 ### 4 拿數據 拿數據都是從隊頭開始拿數據，源碼如下： ``` public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { // 如果隊列為空，無限等待 // 直到隊列中有數據被 put 后，自己被喚醒 while (count == 0) notEmpty.await(); // 從隊列中拿數據 return dequeue(); } finally { lock.unlock(); } } private E dequeue() { final Object[] items = this.items; // takeIndex 代表本次拿數據的位置，是上一次拿數據時計算好的 E x = (E) items[takeIndex]; // 幫助 gc items[takeIndex] = null; // ++ takeIndex 計算下次拿數據的位置 // 如果正好等于隊尾的話，下次就從 0 開始拿數據 if (++takeIndex == items.length) takeIndex = 0; // 隊列實際大小減 1 count--; if (itrs != null) itrs.elementDequeued(); // 喚醒被隊列滿所阻塞的線程 notFull.signal(); return x; } ``` 從源碼中可以看出，每次拿數據的位置就是 takeIndex 的位置，在找到本次該拿的數據之后，會把 takeIndex 加 1，計算下次拿數據時的索引位置，有個特殊情況是，如果本次拿數據的位置已經是隊尾了，那么下次拿數據的位置就要從頭開始，就是從 0 開始了。 ### 5 刪除數據 刪除數據很有意思，我們一起來看下核心源碼： ``` // 一共有兩種情況： // 1：刪除位置和 takeIndex 的關系：刪除位置和 takeIndex 一樣，比如 takeIndex 是 2， 而要刪除的位置正好也是 2，那么就把位置 2 的數據置為 null ,并重新計算 takeIndex 為 3。 // 2：找到要刪除元素的下一個，計算刪除元素和 putIndex 的關系 // 如果下一個元素不是 putIndex，就把下一個元素往前移動一位 // 如果下一個元素是 putIndex，把 putIndex 的值修改成刪除的位置 void removeAt(final int removeIndex) { final Object[] items = this.items; // 情況1 如果刪除位置正好等于下次要拿數據的位置 if (removeIndex == takeIndex) { // 下次要拿數據的位置直接置空 items[takeIndex] = null; // 要拿數據的位置往后移動一位 if (++takeIndex == items.length) takeIndex = 0; // 當前數組的大小減一 count--; if (itrs != null) itrs.elementDequeued(); // 情況 2 } else { final int putIndex = this.putIndex; for (int i = removeIndex;;) { // 找到要刪除元素的下一個 int next = i + 1; if (next == items.length) next = 0; // 下一個元素不是 putIndex if (next != putIndex) { // 下一個元素往前移動一位 items[i] = items[next]; i = next; // 下一個元素是 putIndex } else { // 刪除元素 items[i] = null; // 下次放元素時，應該從本次刪除的元素放 this.putIndex = i; break; } } count--; if (itrs != null) itrs.removedAt(removeIndex); } notFull.signal(); } ``` 刪除數據的情況比較復雜，一共有兩種情況，第一種情況是 takeIndex == removeIndex，我們畫個示意圖來看下處理方式：  第二種情況又分兩種： 1. 如果 removeIndex + 1 != putIndex 的話，就把下一個元素往前移動一位，示意圖如下：  2. 如果 removeIndex + 1 == putIndex 的話，就把 putIndex 的值修改成刪除的位置，示意圖如下：  ArrayBlockingQueue 的刪除方法其實還蠻復雜的，需要考慮到很多特殊的場景。 ### 6 總結 ArrayBlockingQueue 底層是有界的數組，整體來說，和其它隊列差別不多，需要注意的是，當 takeIndex、putIndex 到隊尾的時候，都會重新從 0 開始循環，這點是比較特殊的，在我們學習源碼時，需要特別注意。