今天要講的是ArrayBlockQueue，ArrayBlockQueue是JUC提供的線程安全的有界的阻塞隊(duì)列，一看到Array，第一反應(yīng)：這貨肯定和數(shù)組有關(guān)，既然是數(shù)組，那自然是有界的了，我們先來(lái)看看ArrayBlockQueue的基本使用方法，然后再看看ArrayBlockQueue的源碼。

ArrayBlockQueue基本使用

public static void main(String[] args) throws InterruptedException {
    ArrayBlockingQueue<Integer> arrayBlockingQueue=new ArrayBlockingQueue(5);
    arrayBlockingQueue.offer(10);
    arrayBlockingQueue.offer(50);
    arrayBlockingQueue.add(20);
    arrayBlockingQueue.add(60);
    System.out.println(arrayBlockingQueue);

    System.out.println(arrayBlockingQueue.poll());
    System.out.println(arrayBlockingQueue);

    System.out.println(arrayBlockingQueue.take());
    System.out.println(arrayBlockingQueue);

    System.out.println(arrayBlockingQueue.peek());
    System.out.println(arrayBlockingQueue);
  }

運(yùn)行結(jié)果：

1. 創(chuàng)建了一個(gè)長(zhǎng)度為5的ArrayBlockQueue。
2. 用offer方法，向ArrayBlockQueue添加了兩個(gè)元素，分別是10，50。
3. 用put方法，向ArrayBlockQueue添加了兩個(gè)元素，分別是20，60。
4. 打印出ArrayBlockQueue，結(jié)果是10，50，20，60。
5. 用poll方法，彈出ArrayBlockQueue第一個(gè)元素，并且打印出來(lái)：10。
6. 打印出ArrayBlockQueue，結(jié)果是50，20，60。
7. 用take方法，彈出ArrayBlockQueue第一個(gè)元素，并且打印出來(lái)：50。
8. 打印出ArrayBlockQueue，結(jié)果是20，60。
9. 用peek方法，彈出ArrayBlockQueue第一個(gè)元素，并且打印出來(lái)：20。
10. 打印出ArrayBlockQueue，結(jié)果是20，60。

 代碼比較簡(jiǎn)單，但是你肯定會(huì)有疑問(wèn)

1. offer/add（在上面的代碼中沒(méi)有演示）/put都是往隊(duì)列里面添加元素，區(qū)別是什么？
2. poll/take/peek都是彈出隊(duì)列的元素，區(qū)別是什么？
3. 底層代碼是如何保證線程安全的？
4. 數(shù)據(jù)保存在哪里？

要解決上面幾個(gè)疑問(wèn)，最好的辦法當(dāng)然是看下源碼，通過(guò)親自閱讀源碼所產(chǎn)生的印象遠(yuǎn)遠(yuǎn)要比看視頻，看博客，死記硬背最后的結(jié)論要深刻的多。就算真的忘記了，只要再看看源碼，瞬間可以回憶起來(lái)。

ArrayBlockQueue源碼解析

構(gòu)造方法

ArrayBlockQueue提供了三個(gè)構(gòu)造方法，如下圖所示：

ArrayBlockingQueue(int capacity)

 public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
  }

這是最常用的構(gòu)造方法，傳入capacity，capacity是容量的意思，也就是ArrayBlockingQueue的最大長(zhǎng)度，方法內(nèi)部直接調(diào)用了第二個(gè)構(gòu)造方法，傳入的第二個(gè)參數(shù)為false。

ArrayBlockingQueue(int capacity, boolean fair)

 public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
      throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull = lock.newCondition();
  }

這個(gè)構(gòu)造方法接受兩個(gè)參數(shù)，分別是capacity和fair，fair是boolean類型的，代表是公平鎖，還是非公平鎖，可以看出如果我們用第一個(gè)構(gòu)造方法來(lái)創(chuàng)建ArrayBlockingQueue的話，采用的是非公平鎖，因?yàn)楣芥i會(huì)損失一定的性能，在沒(méi)有充足的理由的情況下，是沒(méi)有必要采用公平鎖的。

方法內(nèi)部做了幾件事情：

1. 創(chuàng)建Object類型的數(shù)組，容量為capacity，并且賦值給當(dāng)前類對(duì)象的items。
2. 創(chuàng)建排他鎖。
3. 創(chuàng)建條件變量notEmpty 。
4. 創(chuàng)建條件變量notFull。

至于排他鎖和兩個(gè)條件變量是做什么用的，看到后面就明白了。

ArrayBlockingQueue(int capacity, boolean fair,Collection<? extends E> c)

public ArrayBlockingQueue(int capacity, boolean fair,
               Collection<? extends E> c) {
    //調(diào)用第二個(gè)構(gòu)造方法，方法內(nèi)部就是初始化數(shù)組，排他鎖，兩個(gè)條件變量
    this(capacity, fair);

    final ReentrantLock lock = this.lock;
    lock.lock(); // 開(kāi)啟排他鎖
    try {
      int i = 0;
      try {
        // 循環(huán)傳入的集合，把集合中的元素賦值給items數(shù)組，其中i會(huì)自增
        for (E e : c) {
          checkNotNull(e);
          items[i++] = e;
        }
      } catch (ArrayIndexOutOfBoundsException ex) {
        throw new IllegalArgumentException();
      }
      count = i;//把i賦值給count 
      //如果i==capacity，也就是到了最大容量，把0賦值給putIndex，否則把i賦值給putIndex
      putIndex = (i == capacity) ? 0 : i;
    } finally {
      lock.unlock();//釋放排他鎖
    }
  }

1. 調(diào)用第二個(gè)構(gòu)造方法，方法內(nèi)部就是初始化數(shù)組items，排他鎖lock，以及兩個(gè)條件變量。
2. 開(kāi)啟排他鎖。
3. 循環(huán)傳入的集合，將集合中的元素賦值給items數(shù)組，其中i會(huì)自增。
4. 把i賦值給count。
5. 如果i==capacity，說(shuō)明到了最大的容量，就把0賦值給putIndex，否則把i賦值給putIndex。
6. 在finally中釋放排他鎖。

看到這里，我們應(yīng)該明白這個(gè)構(gòu)造方法的作用是什么了，就是把傳入的集合作為ArrayBlockingQueuede初始化數(shù)據(jù)，但是我們又會(huì)有一個(gè)新的疑問(wèn)：count，putIndex 是做什么用的。

offer(E e)

 public boolean offer(E e) {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock();//開(kāi)啟排他鎖
    try {
      if (count == items.length)//如果count==items.length，返回false
        return false;
      else {
        enqueue(e);//入隊(duì)
        return true;//返回true
      }
    } finally {
      lock.unlock();//釋放鎖
    }
  }

1. 開(kāi)啟排他鎖。
2. 如果count==items.length，也就是到了最大的容量，返回false。
3. 如果count<items.length，執(zhí)行入隊(duì)方法，并且返回true。
4. 釋放排他鎖。

看到這里，我們應(yīng)該可以明白了，ArrayBlockQueue是如何保證線程安全的，還是利用了ReentrantLock排他鎖，count就是用來(lái)保存數(shù)組的當(dāng)前大小的。我們?cè)賮?lái)看看enqueue方法。

 private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
      putIndex = 0;
    count++;
    notEmpty.signal();
  }

這方法比較簡(jiǎn)單，在代碼里面就不寫注釋了，做了如下的操作：

1. 把x賦值給items[putIndex] 。
2. 將putIndex進(jìn)行自增，如果自增后的值 == items.length，把0賦值給putIndex 。
3. 執(zhí)行count++操作。
4. 調(diào)用條件變量notEmpty的signal方法，說(shuō)明在某個(gè)地方，必定調(diào)用了notEmpty的await方法，這里就是喚醒因?yàn)檎{(diào)用notEmpty的await方法而被阻塞的線程。

這里就解答了一個(gè)疑問(wèn)：putIndex是做什么的，就是入隊(duì)元素的下標(biāo)。

add(E e)

 public boolean add(E e) {
    return super.add(e);
  }

 public boolean add(E e) {
    if (offer(e))
      return true;
    else
      throw new IllegalStateException("Queue full");
  }

這個(gè)方法內(nèi)部最終還是調(diào)用的offer方法。

put(E e)

 public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();//開(kāi)啟響應(yīng)中斷的排他鎖
    try {
      while (count == items.length)//如果隊(duì)列滿了，調(diào)用notFull的await
        notFull.await();
      enqueue(e);//入隊(duì)
    } finally {
      lock.unlock();//釋放排他鎖
    }
  }

1. 開(kāi)啟響應(yīng)中斷的排他鎖，如果在獲取鎖的過(guò)程中，當(dāng)前的線程被中斷，會(huì)拋出異常。
2. 如果隊(duì)列滿了，調(diào)用notFull的await方法，說(shuō)明在某個(gè)地方，必定調(diào)用了notFull的signal方法來(lái)喚醒當(dāng)前線程，這里用while循環(huán)是為了防止虛假喚醒。
3. 執(zhí)行入隊(duì)操作。
4. 釋放排他鎖。

可以看到put方法和 offer/add方法的區(qū)別了：

1. offer/add：如果隊(duì)列滿了，直接返回false。
2. put：如果隊(duì)列滿了，當(dāng)前線程被阻塞，等待喚醒。

poll()

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
      return (count == 0) ? null : dequeue();
    } finally {
      lock.unlock();
    }
  }

1. 開(kāi)啟排他鎖。
2. 如果count==0，直接返回null，否則執(zhí)行dequeue出隊(duì)操作。
3. 釋放排他鎖。

我們來(lái)看dequeue方法：

 private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];//獲得元素的值
    items[takeIndex] = null;//把null賦值給items[takeIndex] 
    if (++takeIndex == items.length)//如果takeIndex自增后的值== items.length，就把0賦值給takeIndex
      takeIndex = 0;
    count--;
    if (itrs != null)
      itrs.elementDequeued();
    notFull.signal();//喚醒因?yàn)檎{(diào)用notFull的await方法而被阻塞的線程
    return x;
  }

1. 獲取元素的值，takeIndex保存的是出隊(duì)的下標(biāo)。
2. 把null賦值給items[takeIndex]，也就是清空被彈出的元素。
3. 如果takeIndex自增后的值== items.length，就把0賦值給takeIndex。
4. count--。
5. 喚醒因?yàn)檎{(diào)用notFull的await方法而被阻塞的線程。

這里調(diào)用了notFull的signal方法來(lái)喚醒因?yàn)檎{(diào)用notFull的await方法而被阻塞的線程，那到底在哪里調(diào)用了notFull的await方法呢，還記不記得在put方法中調(diào)用了notFull的await方法，我們?cè)倏纯矗?/p>

  while (count == items.length)
        notFull.await();

當(dāng)隊(duì)列滿了，就調(diào)用 notFull.await()來(lái)等待，在出隊(duì)操作中，又調(diào)用了notFull.signal()來(lái)喚醒。

take()

 public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
      while (count == 0)
        notEmpty.await();
      return dequeue();
    } finally {
      lock.unlock();
    }
  }

1. 開(kāi)啟排他鎖。
2. 如果count==0，代表隊(duì)列是空的，則調(diào)用notEmpty的await方法，用while循環(huán)是為了防止虛假喚醒。
3. 執(zhí)行出隊(duì)操作。
4. 釋放排他鎖。

這里調(diào)用了notEmpty的await方法，那么哪里調(diào)用了notEmpty的signal方法呢？在enqueue入隊(duì)方法里。

我們可以看到take和poll的區(qū)別：

1. take：如果隊(duì)列為空，會(huì)阻塞，直到被喚醒了。
2. poll： 如果隊(duì)列為空，直接返回null。

peek()

public E peek() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
      return itemAt(takeIndex); 
    } finally {
      lock.unlock();
    }
  }

 final E itemAt(int i) {
    return (E) items[i];
  }

1. 開(kāi)啟排他鎖。
2. 獲得元素。
3. 釋放排他鎖。

我們可以看到peek和poll/take的區(qū)別：

1. peek，只是獲取元素，不會(huì)清空元素。
2. poll/take，獲取并清空元素。

size()

 public int size() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
      return count;
    } finally {
      lock.unlock();
    }
  }

1. 開(kāi)啟排他鎖。
2. 返回count。
3. 釋放排他鎖。

總結(jié)

至此，ArrayBlockQueue的核心源碼就分析完畢了，我們來(lái)做一個(gè)總結(jié)：

1. ArrayBlockQueue有幾個(gè)比較重要的字段，分別是items，保存的是隊(duì)列的數(shù)據(jù)，putIndex保存的是入隊(duì)的下標(biāo)，takeIndex保存的是出隊(duì)的下標(biāo)，count用來(lái)統(tǒng)計(jì)隊(duì)列元素的個(gè)數(shù)，lock用來(lái)保證線程的安全性，notEmpty和notFull兩個(gè)條件變量實(shí)現(xiàn)喚醒和阻塞。
2. offer和add是一樣的，其中add方法內(nèi)部調(diào)用的就是offer方法，如果隊(duì)列滿了，直接返回false。
3. put，如果隊(duì)列滿了，會(huì)被阻塞。
4. peek，只是彈出元素，不會(huì)清空元素。
5. poll，彈出并清空元素，如果隊(duì)列為空，直接返回null。
6. take，彈出并清空元素，如果隊(duì)列為空，會(huì)被阻塞。

以上所述是小編給大家介紹的ArrayBlockQueue源碼解析詳解整合，希望對(duì)大家有所幫助，如果大家有任何疑問(wèn)請(qǐng)給我留言，小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對(duì)腳本之家網(wǎng)站的支持！

最新評(píng)論