前言

ArrayBlockingQueue 由數(shù)組支持的有界阻塞隊(duì)列，隊(duì)列基于數(shù)組實(shí)現(xiàn)，容量大小在創(chuàng)建 ArrayBlockingQueue 對(duì)象時(shí)已經(jīng)定義好。 此隊(duì)列按照先進(jìn)先出(FIFO)的原則對(duì)元素進(jìn)行排序。支持公平鎖和非公平鎖，默認(rèn)采用非公平鎖。其數(shù)據(jù)結(jié)構(gòu)如下圖：

• 注：每一個(gè)線程在獲取鎖的時(shí)候可能都會(huì)排隊(duì)等待，如果在等待時(shí)間上，先獲取鎖的線程和請(qǐng)求一定先被滿足，那么這個(gè)鎖就是公平的。反之，這個(gè)鎖就是不公平的。公平的獲取鎖，也就是當(dāng)前等待時(shí)間最長(zhǎng)的線程先獲取鎖

隊(duì)列創(chuàng)建

BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(5); 

應(yīng)用場(chǎng)景

在線程池中有比較多的應(yīng)用，生產(chǎn)者消費(fèi)者場(chǎng)景。

• 先進(jìn)先出隊(duì)列(隊(duì)列頭的是最先進(jìn)隊(duì)的元素;隊(duì)列尾的是最后進(jìn)隊(duì)的元素)
• 有界隊(duì)列(即初始化時(shí)指定的容量，就是隊(duì)列最大的容量，不會(huì)出現(xiàn)擴(kuò)容，容量滿，則阻塞進(jìn)隊(duì)操作;容量空，則阻塞出隊(duì)操作)
• 隊(duì)列不支持空元素
• 公平性 (fairness)可以在構(gòu)造函數(shù)中指定。

此類支持對(duì)等待的生產(chǎn)者線程和使用者線程進(jìn)行排序的可選公平策略。默認(rèn)情況下，不保證是這種排序。然而，通過(guò)在構(gòu)造函數(shù)將公平性 (fairness) 設(shè)置為 true 而構(gòu)造的隊(duì)列允許按照 FIFO 順序訪問(wèn)線程。公平性通常會(huì)降低吞吐量，但也減少了可變性和避免了“不平衡性”。

工作原理

ArrayBlockingQueue是對(duì)BlockingQueue的一個(gè)數(shù)組實(shí)現(xiàn)，它使用一把全局的鎖并行對(duì)queue的讀寫(xiě)操作，同時(shí)使用兩個(gè)Condition阻塞容量為空時(shí)的取操作和容量滿時(shí)的寫(xiě)操作。

基于 ReentrantLock 保證線程安全，根據(jù) Condition 實(shí)現(xiàn)隊(duì)列滿時(shí)的阻塞。

final ReentrantLock lock; 
 
private final Condition notEmpty; 
 
private final Condition notFull; 

 Lock的作用是提供獨(dú)占鎖機(jī)制，來(lái)保護(hù)競(jìng)爭(zhēng)資源;而Condition是為了更加精細(xì)地對(duì)鎖進(jìn)行控制，它依賴于Lock，通過(guò)某個(gè)條件對(duì)多線程進(jìn)行控制。

notEmpty表示“鎖的非空條件”。當(dāng)某線程想從隊(duì)列中取數(shù)據(jù)時(shí)，而此時(shí)又沒(méi)有數(shù)據(jù)，則該線程通過(guò)notEmpty.await()進(jìn)行等待;當(dāng)其它線程向隊(duì)列中插入了元素之后，就調(diào)用notEmpty.signal()喚醒“之前通過(guò)notEmpty.await()進(jìn)入等待狀態(tài)的線程”。 同理，notFull表示“鎖的滿條件”。當(dāng)某線程想向隊(duì)列中插入元素，而此時(shí)隊(duì)列已滿時(shí)，該線程等待;當(dāng)其它線程從隊(duì)列中取出元素之后，就喚醒該等待的線程。

• 試圖向已滿隊(duì)列中放入元素會(huì)導(dǎo)致放入操作受阻塞，直到BlockingQueue里有新的喚空間才會(huì)被醒繼續(xù)操作; 試圖從空隊(duì)列中檢索元素將導(dǎo)致類似阻塞，直到BlocingkQueue進(jìn)了新貨才會(huì)被喚醒。

源碼分析

以下源碼分析基于JDK1.8

定義

ArrayBlockingQueue的類繼承關(guān)系如下：

  其包含的方法定義如下：

 成員屬性

/** 真正存入數(shù)據(jù)的數(shù)組 */ 
   final Object[] items; 
 
   /** take，poll，peek or remove 的下一個(gè)索引 */ 
   int takeIndex; 
 
   /** put，offer，or add 下一個(gè)索引 */ 
   int putIndex; 
 
   /** 隊(duì)列中元素個(gè)數(shù) */ 
   int count; 
 
   /** 可重入鎖 */ 
   final ReentrantLock lock; 
 
   /** 如果數(shù)組是空的，在該Condition上等待 */ 
   private final Condition notEmpty; 
 
   /** 如果數(shù)組是滿的，在該Condition上等待 */ 
   private final Condition notFull; 
 
   /** 遍歷器實(shí)現(xiàn) */ 
   transient Itrs itrs = null; 

 構(gòu)造函數(shù) 

/** 
    * 構(gòu)造函數(shù)，設(shè)置隊(duì)列的初始容量 
    */ 
   public ArrayBlockingQueue(int capacity) { 
       this(capacity, false); 
   } 
 
   /** 
    * 構(gòu)造函數(shù)， 
    * capacity and the specified access policy. 
    * 
    * @param capacity 設(shè)置數(shù)組大小 
    * @param fair  設(shè)置是否為公平鎖 
    * @throws IllegalArgumentException if {@code capacity < 1} 
    */ 
   public ArrayBlockingQueue(int capacity, boolean fair) { 
       if (capacity <= 0) 
           throw new IllegalArgumentException(); 
       this.items = new Object[capacity]; 
       // 是否為公平鎖，如果是的話，那么先到的線程先獲得鎖對(duì)象 
       // 否則，由操作系統(tǒng)調(diào)度由哪個(gè)線程獲得鎖，一般為false，性能會(huì)比較高 
       lock = new ReentrantLock(fair);  
       notEmpty = lock.newCondition(); 
       notFull =  lock.newCondition(); 
   } 
 
   /** 
    * 構(gòu)造函數(shù)，帶有初始內(nèi)容的隊(duì)列 
    */ 
   public ArrayBlockingQueue(int capacity, boolean fair, 
                             Collection<? extends E> c) { 
       this(capacity, fair); 
 
       final ReentrantLock lock = this.lock; 
       //加鎖的目的是為了其他CPU能夠立即看到修改 
       //加鎖和解鎖底層都是CAS，會(huì)強(qiáng)制修改寫(xiě)回主存，對(duì)其他CPU可見(jiàn) 
       lock.lock(); // 要給數(shù)組設(shè)置內(nèi)容，先上鎖 
       try { 
           int i = 0; 
           try { 
               for (E e : c) { 
                   checkNotNull(e); 
                   items[i++] = e; // 依次拷貝內(nèi)容 
               } 
           } catch (ArrayIndexOutOfBoundsException ex) { 
               throw new IllegalArgumentException(); 
           } 
           count = i; 
           putIndex = (i == capacity) ? 0 : i; // 如果 putIndex大于數(shù)組大小，那么從0重寫(xiě)開(kāi)始 
       } finally { 
           lock.unlock(); // 最后一定要釋放鎖 
       } 
   } 

 入隊(duì)方法

add / offer / put，這三個(gè)方法都是往隊(duì)列中添加元素，說(shuō)明如下：

• add方法依賴于offer方法，如果隊(duì)列滿了則拋出異常，否則添加成功返回true;
• offer方法有兩個(gè)重載版本，只有一個(gè)參數(shù)的版本，如果隊(duì)列滿了就返回false，否則加入到隊(duì)列中，返回true，add方法就是調(diào)用此版本的offer方法;另一個(gè)帶時(shí)間參數(shù)的版本，如果隊(duì)列滿了則等待，可指定等待的時(shí)間，如果這期間中斷了則拋出異常，如果等待超時(shí)了則返回false，否則加入到隊(duì)列中返回true;
• put方法跟帶時(shí)間參數(shù)的offer方法邏輯一樣，不過(guò)沒(méi)有等待的時(shí)間限制，會(huì)一直等待直到隊(duì)列有空余位置了，再插入到隊(duì)列中，返回true

/** 
     * 添加一個(gè)元素，其實(shí)super.add里面調(diào)用了offer方法 
     */ 
    public boolean add(E e) { 
        return super.add(e); 
    } 
 
    /** 
     * 加入成功返回 true，否則返回 false 
     */ 
    public boolean offer(E e) { 
     // 創(chuàng)建插入的元素是否為null，是的話拋出NullPointerException異常 
        checkNotNull(e); 
        // 獲取“該阻塞隊(duì)列的獨(dú)占鎖” 
        final ReentrantLock lock = this.lock; 
        lock.lock(); // 上鎖 
        try { 
         // 如果隊(duì)列已滿，則返回false。 
            if (count == items.length) // 超過(guò)數(shù)組的容量 
                return false; 
            else { 
             // 如果隊(duì)列未滿，則插入e，并返回true。 
                enqueue(e);  
                return true; 
            } 
        } finally { 
         // 釋放鎖 
            lock.unlock(); 
        } 
    } 
 
    /** 
     * 如果隊(duì)列已滿的話，就會(huì)等待 
     */ 
    public void put(E e) throws InterruptedException { 
        checkNotNull(e); 
        final ReentrantLock lock = this.lock; 
        lock.lockInterruptibly(); //和lock方法的區(qū)別是讓它在阻塞時(shí)可以拋出異常跳出 
        try { 
            while (count == items.length) 
                notFull.await(); // 這里就是阻塞了，要注意：如果運(yùn)行到這里，那么它會(huì)釋放上面的鎖，一直等到 notify 
            enqueue(e); 
        } finally { 
            lock.unlock(); 
        } 
    } 
 
    /** 
     * 帶有超時(shí)事件的插入方法，unit 表示是按秒、分、時(shí)哪一種 
     */ 
    public boolean offer(E e, long timeout, TimeUnit unit) 
        throws InterruptedException { 
 
        checkNotNull(e); 
        long nanos = unit.toNanos(timeout); 
        final ReentrantLock lock = this.lock; 
        lock.lockInterruptibly(); 
        try { 
            while (count == items.length) { 
                if (nanos <= 0) 
                    return false; 
                nanos = notFull.awaitNanos(nanos); // 帶有超時(shí)等待的阻塞方法 
            } 
            enqueue(e); // 入隊(duì) 
            return true; 
        } finally { 
            lock.unlock(); 
        } 
    } 

 出隊(duì)方法

poll / take / peek，這幾個(gè)方法都是獲取隊(duì)列頂?shù)脑?，具體說(shuō)明如下：

• poll方法有兩個(gè)重載版本，第一個(gè)版本，如果隊(duì)列是空的，返回null，否則移除并返回隊(duì)列頭部元素;另一個(gè)帶時(shí)間參數(shù)的版本，如果棧為空則等待，可以指定等待的時(shí)間，如果等待超時(shí)了則返回null，如果被中斷了則拋出異常，否則移除并返回棧頂元素
• take方法同帶時(shí)間參數(shù)的poll方法，但是不能指定等待時(shí)間，會(huì)一直等待直到隊(duì)列中有元素為止，然后移除并返回棧頂元素
• peek方法只是返回隊(duì)列頭部元素，不移除

// 實(shí)現(xiàn)的方法，如果當(dāng)前隊(duì)列為空，返回null 
   public E poll() { 
       final ReentrantLock lock = this.lock; 
       lock.lock(); 
       try { 
           return (count == 0) ? null : dequeue(); 
       } finally { 
           lock.unlock(); 
       } 
   } 
// 實(shí)現(xiàn)的方法，如果當(dāng)前隊(duì)列為空，一直阻塞 
   public E take() throws InterruptedException { 
       final ReentrantLock lock = this.lock; 
       lock.lockInterruptibly(); 
       try { 
           while (count == 0) 
               notEmpty.await(); // 隊(duì)列為空，阻塞方法 
           return dequeue(); 
       } finally { 
           lock.unlock(); 
       } 
   } 
// 帶有超時(shí)事件的取元素方法，否則返回null 
   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); // 超時(shí)等待 
           } 
           return dequeue(); // 取得元素 
       } finally { 
           lock.unlock(); 
       } 
   } 
 
   // 只是看一個(gè)隊(duì)列最前面的元素，取出是不擅長(zhǎng)隊(duì)列中原來(lái)的元素，隊(duì)列為空時(shí)返回null 
   public E peek() { 
       final ReentrantLock lock = this.lock; 
       lock.lock(); 
       try { 
           return itemAt(takeIndex); // 隊(duì)列為空時(shí)返回null 
       } finally { 
           lock.unlock(); 
       } 
   } 

 刪除元素方法

remove / clear /drainT，這三個(gè)方法用于從隊(duì)列中移除元素，具體說(shuō)明如下：

• remove方法用于移除某個(gè)元素，如果棧為空或者沒(méi)有找到該元素則返回false，否則從棧中移除該元素;移除時(shí)，如果該元素位于棧頂則直接移除，如果位于棧中間，則需要將該元素后面的其他元素往前面挪動(dòng)，移除后需要喚醒因?yàn)闂M了而阻塞的線程
• clear方法用于整個(gè)棧，同時(shí)將takeIndex置為putIndex，保證棧中的元素先進(jìn)先出;最后會(huì)喚醒最多count個(gè)線程，因?yàn)檎Ｒ粋€(gè)線程插入一個(gè)元素，如果喚醒超過(guò)count個(gè)線程，可能導(dǎo)致部分線程因?yàn)闂M了又再次被阻塞
• drainTo方法有兩個(gè)重載版本，一個(gè)是不帶個(gè)數(shù)，將所有的元素都移除并拷貝到指定的集合中;一個(gè)帶個(gè)數(shù)，將指定個(gè)數(shù)的元素移除并拷貝到指定的集合中，兩者的底層實(shí)現(xiàn)都是同一個(gè)方法。移除后需要重置takeIndex和count，并喚醒最多移除個(gè)數(shù)的因?yàn)闂M而阻塞的線程。

/** 
    * 從隊(duì)列中刪除一個(gè)元素的方法。刪除成功返回true，否則返回false 
    */ 
   public boolean remove(Object o) { 
       if (o == null) return false; 
       final Object[] items = this.items; 
       final ReentrantLock lock = this.lock; 
       lock.lock(); 
       try { 
           if (count > 0) { 
               final int putIndex = this.putIndex; 
               int i = takeIndex; 
               //從takeIndex開(kāi)始往后遍歷直到等于putIndex 
               do { 
                   if (o.equals(items[i])) { 
                       removeAt(i); // 真正刪除的方法 
                       return true; 
                   } 
                   //走到數(shù)組末尾了又從頭開(kāi)始，put時(shí)也按照這個(gè)規(guī)則來(lái) 
                   if (++i == items.length) 
                       i = 0; 
               } while (i != putIndex); // 一直不斷的循環(huán)取出來(lái)做判斷 
           } 
           //如果數(shù)組為空，返回false 
           return false; 
       } finally { 
           lock.unlock(); 
       } 
   } 
 
/** 
    * 指定刪除索引上的元素. 
    */ 
   void removeAt(final int removeIndex) { 
       // assert lock.getHoldCount() == 1; 
       // assert items[removeIndex] != null; 
       // assert removeIndex >= 0 && removeIndex < items.length; 
       final Object[] items = this.items; 
       if (removeIndex == takeIndex) { 
           //如果移除的就是棧頂?shù)脑?nbsp;
           items[takeIndex] = null; 
           if (++takeIndex == items.length) 
               takeIndex = 0; 
           //元素個(gè)數(shù)減1 
           count--; 
           if (itrs != null) 
               itrs.elementDequeued(); 
       } else { 
           // an "interior" remove 
 
           // 如果移除的是棧中間的某個(gè)元素，需要將該元素后面的元素往前挪動(dòng) 
           final int putIndex = this.putIndex; 
           for (int i = removeIndex;;) { 
               int next = i + 1; 
               //到數(shù)組末尾了，從頭開(kāi)始 
               if (next == items.length) 
                   next = 0; 
               if (next != putIndex) { 
                //將后面一個(gè)元素復(fù)制到前面來(lái) 
                   items[i] = items[next]; 
                   i = next; 
               } else { 
                //如果下一個(gè)元素的索引等于putIndex，說(shuō)明i就是棧中最后一個(gè)元素了，直接將該元素置為null 
                   items[i] = null; 
                   //重置putIndex為i 
                   this.putIndex = i; 
                   break; 
               } 
           } 
           count--; 
           if (itrs != null) 
            //通知itrs節(jié)點(diǎn)移除了 
               itrs.removedAt(removeIndex); 
       } 
       //喚醒因?yàn)闂M了而等待的線程 
       notFull.signal(); // 有一個(gè)元素刪除成功，那肯定隊(duì)列不滿 
   } 
 
/** 
    * 清空隊(duì)列 
    */ 
   public void clear() { 
       final Object[] items = this.items; 
       final ReentrantLock lock = this.lock; 
       lock.lock(); 
       try { 
           int k = count; 
           if (k > 0) { 
               final int putIndex = this.putIndex; 
               int i = takeIndex; 
               //從takeIndex開(kāi)始遍歷直到i等于putIndex，將數(shù)組元素置為null 
               do { 
                   items[i] = null; 
                   if (++i == items.length) 
                       i = 0; 
               } while (i != putIndex); 
               //注意此處沒(méi)有將這兩個(gè)index置為0，只是讓他們相等，因?yàn)橹灰嗟染涂梢詫?shí)現(xiàn)棧先進(jìn)先出了 
               takeIndex = putIndex; 
               count = 0; 
               if (itrs != null) 
                   itrs.queueIsEmpty(); 
               //如果有因?yàn)闂M了而等待的線程，則將其喚醒 
               //注意這里沒(méi)有使用signalAll而是通過(guò)for循環(huán)來(lái)signal多次，單純從喚醒線程來(lái)看是可以使用signalAll的，效果跟這里的for循環(huán)是一樣的 
               //如果有等待的線程，說(shuō)明count就是當(dāng)前線程的最大容量了，這里清空了，最多只能put count次，一個(gè)線程只能put 1次，只喚醒最多count個(gè)線程就避免了 
               //線程被喚醒后再次因?yàn)闂M了而阻塞 
               for (; k > 0 && lock.hasWaiters(notFull); k--) 
                   notFull.signal(); 
           } 
       } finally { 
           lock.unlock(); 
       } 
   } 
 
   /** 
    * 取出所有元素到集合 
    */ 
   public int drainTo(Collection<? super E> c) { 
       return drainTo(c, Integer.MAX_VALUE); 
   } 
 
   /** 
    * 取出所有元素到集合 
    */ 
   public int drainTo(Collection<? super E> c, int maxElements) { 
       //校驗(yàn)參數(shù)合法 
       checkNotNull(c); 
       if (c == this) 
           throw new IllegalArgumentException(); 
       if (maxElements <= 0) 
           return 0; 
       final Object[] items = this.items; 
       final ReentrantLock lock = this.lock; 
       lock.lock(); 
       try { 
        //取兩者之間的最小值 
           int n = Math.min(maxElements, count); 
           int take = takeIndex; 
           int i = 0; 
           try { 
            //從takeIndex開(kāi)始遍歷，取出元素然后添加到c中，直到滿足個(gè)數(shù)要求為止 
               while (i < n) { 
                   @SuppressWarnings("unchecked") 
                   E x = (E) items[take]; 
                   c.add(x); 
                   items[take] = null; 
                   if (++take == items.length) 
                       take = 0; 
                   i++; 
               } 
               return n; 
           } finally { 
               // Restore invariants even if c.add() threw 
               if (i > 0) { 
                //取完了，修改count減去i 
                   count -= i; 
                   takeIndex = take; 
                   if (itrs != null) { 
                       if (count == 0) 
                        //通知itrs ?？樟?nbsp;
                           itrs.queueIsEmpty(); 
                       else if (i > take) 
                        //說(shuō)明take中間變成0了，通知itrs 
                           itrs.takeIndexWrapped(); 
                   } 
                   //喚醒在因?yàn)闂M而等待的線程，最多喚醒i個(gè)，同上避免線程被喚醒了因?yàn)闂Ｓ譂M了而阻塞 
                   for (; i > 0 && lock.hasWaiters(notFull); i--) 
                       notFull.signal(); 
               } 
           } 
       } finally { 
           lock.unlock(); 
       } 
   } 

 iterator / Itr / Itrs

Itr和Itrs都是ArrayBlockingQueue的兩個(gè)內(nèi)部類，如下：

 iterator方法返回一個(gè)迭代器實(shí)例，用于實(shí)現(xiàn)for循環(huán)遍歷和部分Collection接口，該方法的實(shí)現(xiàn)如下： 

public Iterator<E> iterator() { 
 return new Itr(); 
} 
 
Itr() { 
 // assert lock.getHoldCount() == 0; 
 lastRet = NONE; 
 final ReentrantLock lock = ArrayBlockingQueue.this.lock; 
    lock.lock(); 
    try { 
     if (count == 0) { 
         //NONE和DETACHED都是常量 
            cursor = NONE; 
            nextIndex = NONE; 
            prevTakeIndex = DETACHED; 
        } else { 
         //初始化各屬性 
            final int takeIndex = ArrayBlockingQueue.this.takeIndex; 
            prevTakeIndex = takeIndex; 
            nextItem = itemAt(nextIndex = takeIndex); 
            cursor = incCursor(takeIndex); 
            if (itrs == null) { 
             itrs = new Itrs(this); 
            } else { 
             //初始化Itrs，將當(dāng)前線程注冊(cè)到Itrs 
                itrs.register(this); // in this order 
                itrs.doSomeSweeping(false); 
            } 
            prevCycles = itrs.cycles; 
            // assert takeIndex >= 0; 
            // assert prevTakeIndex == takeIndex; 
            // assert nextIndex >= 0; 
            // assert nextItem != null; 
     } 
 } finally { 
    lock.unlock(); 
    } 
} 
 
Itrs(Itr initial) { 
 register(initial); 
} 
 
//根據(jù)index計(jì)算cursor 
private int incCursor(int index) { 
 // assert lock.getHoldCount() == 1; 
 if (++index == items.length) 
  index = 0; 
 if (index == putIndex) 
  index = NONE; 
 return index; 
} 
 
/** 
* 創(chuàng)建一個(gè)新的Itr實(shí)例時(shí)，會(huì)調(diào)用此方法將該實(shí)例添加到Node鏈表中 
*/ 
void register(Itr itr) { 
 //創(chuàng)建一個(gè)新節(jié)點(diǎn)將其插入到head節(jié)點(diǎn)的前面 
 head = new Node(itr, head); 
} 

 小結(jié)

ArrayBlockingQueue是一個(gè)阻塞隊(duì)列，內(nèi)部由ReentrantLock來(lái)實(shí)現(xiàn)線程安全，由Condition的await和signal來(lái)實(shí)現(xiàn)等待喚醒的功能。它的數(shù)據(jù)結(jié)構(gòu)是數(shù)組，準(zhǔn)確地說(shuō)是一個(gè)循環(huán)數(shù)組(可以類比一個(gè)圓環(huán))，所有的下標(biāo)在到達(dá)最大長(zhǎng)度時(shí)自動(dòng)從0繼續(xù)開(kāi)始。

PS：以上代碼提交在 Github ：

https://github.com/Niuh-Study/niuh-juc-final.git