轉載自? ?集合框架 Queue---ArrayBlockingQueue

?

• 摘要：java.util.concurrent類
• java.util.concurrent?
  類 ArrayBlockingQueue<E>?
  java.lang.Object?
  ? java.util.AbstractCollection<E>?
  ????? java.util.AbstractQueue<E>?
  ????????? java.util.concurrent.ArrayBlockingQueue<E>?
  類型參數:?
  E - 在此 collection 中保持的元素類型?
  所有已實現的接口:?
  Serializable, Iterable<E>, Collection<E>, BlockingQueue<E>, Queue<E>?
  public class ArrayBlockingQueue<E>?
  extends AbstractQueue<E>?
  implements BlockingQueue<E>, Serializable?
  一個由數組支持的有界阻塞隊列。此隊列按 FIFO(先進先出)原則對元素進行排序。隊列的頭部 是在隊列中存在時間最長的元素。隊列的尾部 是在隊列中存在時間最短的元素。新元素插入到隊列的尾部,隊列獲取操作則是從隊列頭部開始獲得元素。?
  這是一個典型的“有界緩存區(qū)”,固定大小的數組在其中保持生產者插入的元素和使用者提取的元素。一旦創(chuàng)建了這樣的緩存區(qū),就不能再增加其容量。試圖向已滿隊列中放入元素會導致操作受阻塞;試圖從空隊列中提取元素將導致類似阻塞。?
  此類支持對等待的生產者線程和使用者線程進行排序的可選公平策略。默認情況下,不保證是這種排序。然而,通過將公平性 (fairness) 設置為 true 而構造的隊列允許按照 FIFO 順序訪問線程。公平性通常會降低吞吐量,但也減少了可變性和避免了“不平衡性”。?
  此類及其迭代器實現了 Collection 和 Iterator 接口的所有可選 方法。?
  此類是 Java Collections Framework 的成員。?
  實現了Serializable可知道是其實線程安全的?
  源代碼:?

public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { /** 隊列元素 數組 */ private final E[] items; /** 獲取、刪除元素時的索引(take, poll 或 remove操作) */ private int takeIndex; /** 添加元素時的索引(put, offer或 add操作) */ private int putIndex; /** 隊列元素的數目*/ private int count; /** 鎖 */ private final ReentrantLock lock; /** 獲取操作時的條件 */ private final Condition notEmpty; /** 插入操作時的條件 */ private final Condition notFull; //超出數組長度時,重設為0 final int inc(int i) { return (++i == items.length)? 0 : i; } /** * 插入元素(在獲得鎖的情況下才調用) */ private void insert(E x) { items[putIndex] = x; putIndex = inc(putIndex); ++count; notEmpty.signal(); } /** * 獲取并移除元素(在獲得鎖的情況下才調用) */ private E extract() { final E[] items = this.items; E x = items[takeIndex]; items[takeIndex] = null; takeIndex = inc(takeIndex);//移到下一個位置 --count; notFull.signal(); return x; } /** * 刪除i位置的元素 */ void removeAt(int i) { final E[] items = this.items; // if removing front item, just advance if (i == takeIndex) { items[takeIndex] = null; takeIndex = inc(takeIndex); } else { // 把i后面的直到putIndex的元素都向前移動一個位置 for (;;) { int nexti = inc(i); if (nexti != putIndex) { items[i] = items[nexti]; i = nexti; } else { items[i] = null; putIndex = i; break; } } } --count; notFull.signal(); } /** *構造方法,指定容量,默認策略(不是按照FIFO的順序訪問) */ public ArrayBlockingQueue(int capacity) { this(capacity, false); } /** *構造方法,指定容量及策略 */ public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity <= 0) throw new IllegalArgumentException(); this.items = (E[]) new Object[capacity]; lock = new ReentrantLock(fair); notEmpty = lock.newCondition(); notFull = lock.newCondition(); } /** * 通過集合構造 */ public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) { this(capacity, fair); if (capacity < c.size()) throw new IllegalArgumentException(); for (Iterator<? extends E> it = c.iterator(); it.hasNext();) add(it.next()); } /** * 插入元素到隊尾(super調用offer方法) * public boolean add(E e) { * if (offer(e)) * return true; * else * throw new IllegalStateException("Queue full"); * } * 將指定的元素插入到此隊列的尾部(如果立即可行且不會超過該隊列的容量), * 在成功時返回 true,如果此隊列已滿,則拋出 IllegalStateException。 */ public boolean add(E e) { return super.add(e); } /** * 將指定的元素插入到此隊列的尾部(如果立即可行且不會超過該隊列的容量), * 在成功時返回 true,如果此隊列已滿,則返回 false。 */ public boolean offer(E e) { if (e == null) throw new NullPointerException(); final ReentrantLock lock = this.lock; lock.lock(); try { if (count == items.length) return false; else { insert(e); return true; } } finally { lock.unlock(); } } /** * 將指定的元素插入此隊列的尾部,如果該隊列已滿,則等待可用的空間。 */ public void put(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); final E[] items = this.items; final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { try { while (count == items.length) notFull.await(); } catch (InterruptedException ie) { notFull.signal(); // propagate to non-interrupted thread throw ie; } insert(e); } finally { lock.unlock(); } } /** * 將指定的元素插入此隊列的尾部,如果該隊列已滿,則在到達指定的等待時間之前等待可用的空間。 */ public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { if (e == null) throw new NullPointerException(); long nanos = unit.toNanos(timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { for (;;) { if (count != items.length) { insert(e); return true; } if (nanos <= 0)//如果時間到了就返回 return false; try { nanos = notFull.awaitNanos(nanos); } catch (InterruptedException ie) { notFull.signal(); // propagate to non-interrupted thread throw ie; } } } finally { lock.unlock(); } } //獲取并移除此隊列的頭,如果此隊列為空,則返回 null。 public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { if (count == 0) return null; E x = extract(); return x; } finally { lock.unlock(); } } //獲取并移除此隊列的頭部,在元素變得可用之前一直等待(如果有必要)。 public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { try { while (count == 0) notEmpty.await(); } catch (InterruptedException ie) { notEmpty.signal(); // propagate to non-interrupted thread throw ie; } E x = extract(); return x; } finally { lock.unlock(); } } //獲取并移除此隊列的頭部,在指定的等待時間前等待可用的元素(如果有必要)。 public E poll(long timeout, TimeUnit unit) throws InterruptedException { long nanos = unit.toNanos(timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { for (;;) { if (count != 0) { E x = extract(); return x; } if (nanos <= 0) return null; try { nanos = notEmpty.awaitNanos(nanos); } catch (InterruptedException ie) { notEmpty.signal(); // propagate to non-interrupted thread throw ie; } } } finally { lock.unlock(); } } //獲取但不移除此隊列的頭;如果此隊列為空,則返回 null。 public E peek() { final ReentrantLock lock = this.lock; lock.lock(); try { return (count == 0) ? null : items[takeIndex]; } finally { lock.unlock(); } } /** * 返回此隊列中元素的數量。 */ public int size() { final ReentrantLock lock = this.lock; lock.lock(); try { return count; } finally { lock.unlock(); } } /** *返回在無阻塞的理想情況下(不存在內存或資源約束)此隊列能接受的其他元素數量。 */ public int remainingCapacity() { final ReentrantLock lock = this.lock; lock.lock(); try { return items.length - count; } finally { lock.unlock(); } } /** * 從此隊列中移除指定元素的單個實例(如果存在)。 */ public boolean remove(Object o) { if (o == null) return false; final E[] items = this.items; final ReentrantLock lock = this.lock; lock.lock(); try { int i = takeIndex; int k = 0; for (;;) { if (k++ >= count) return false; if (o.equals(items[i])) { removeAt(i); return true; } i = inc(i); } } finally { lock.unlock(); } } /** * 如果此隊列包含指定的元素,則返回 true。 */ public boolean contains(Object o) { if (o == null) return false; final E[] items = this.items; final ReentrantLock lock = this.lock; lock.lock(); try { int i = takeIndex; int k = 0; while (k++ < count) { if (o.equals(items[i])) return true; i = inc(i); } return false; } finally { lock.unlock(); } } }

?

總結

以上是生活随笔為你收集整理的集合框架 Queue---ArrayBlockingQueue的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。