摘要：當隊列滿時，存儲元素的線程會等待隊列可用。分別是一個由數組結構組成的有界阻塞隊列。一個支持優先級排序的無界阻塞隊列。此隊列的默認和最大長度為。此隊列按照先進先出的原則對元素進行排序。

最近在看數據結構的時候，看到了隊列這里，在實際的開發中我們很少會手動的去實現一個隊列，甚至很少直接用到隊列，但是在Java的包中有一些具有特殊屬性的隊列應用的比較廣泛，例如：阻塞隊列＆并發隊列.

阻塞隊列（BlockingQueue）是一個額外支持兩種操作的隊列。這兩個附加的操作是：
１、在隊列為空時，獲取元素的線程會等待隊列變為非空。
２、當隊列滿時，存儲元素的線程會等待隊列可用。
阻塞隊列常用于生產者和消費者的場景，生產者是往隊列里添加元素的線程，消費者是從隊列里拿元素的線程。阻塞隊列就是生產者存放元素的容器，而消費者也只從容器里拿元素。

阻塞隊列提供了四種處理方法:

拋出異常

add(e):在添加元素的時候如果隊列已滿，那么直接拋出異常。
remove(e):移除元素，如果隊列為空，那么拋出異常。
element():檢查方法。

public static void test() {
   ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(10);
   for (int i=0; i


返回特殊值
1、offer(e)　入隊的時候返回特殊值，在不同的阻塞隊列中實現有一定的差別
２、poll() 出隊的時候返回特殊的值
３、peek()　測試出隊能否成功



一直阻塞
１、put(e)　如果隊列已滿，那么會一直阻塞，直到成功
２、take()  如果隊列為空，那么出隊會一直阻塞，直到成功



阻塞，超時退出
１、offer(e,time,unit)
２、poll(time,unit)





Java中的阻塞隊列
JDK7提供了7個阻塞隊列。分別是

ArrayBlockingQueue ：一個由數組結構組成的有界阻塞隊列。
LinkedBlockingQueue ：一個由鏈表結構組成的有界阻塞隊列。
PriorityBlockingQueue ：一個支持優先級排序的無界阻塞隊列。
DelayQueue：一個使用優先級隊列實現的無界阻塞隊列。
SynchronousQueue：一個不存儲元素的阻塞隊列。
LinkedTransferQueue：一個由鏈表結構組成的無界阻塞隊列。
LinkedBlockingDeque：一個由鏈表結構組成的雙向阻塞隊列。


下面具體看一下每一種阻塞隊列的實現方式以及使用場景：
1. ArrayBlockingQueue
特性：用數組實現的實現的有界阻塞隊列，默認情況下不保證線程公平的訪問隊列（按照阻塞的先后順序訪問隊列），隊列可用的時候，阻塞的線程都可以爭奪隊列的訪問資格,當然也可以使用以下的構造方法創建一個公平的阻塞隊列。
ArrayBlockingQueue blockingQueue2 = new ArrayBlockingQueue<>(10, true);
下面通過源碼探究以下，這個阻塞隊列是如何實現的？如果實現公平與非公平的控制。
構造過程
public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}

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();
}
/**阻塞隊列的公平與非公平是通過可重入鎖來實現的，關于為什么可重入鎖可以實現線程訪問的公平非公平特性，我們晚一點分析一下ReentrantLock的實現原理。
【關于ReentrantLock的實現原理】https://segmentfault.com/a/11...
add(E) 操作，如果add失敗，那么拋出異常
public boolean add(E e) {
    return super.add(e);
}
/**AbstractQueue 父類的add方法*/
public boolean add(E e) {
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}
/**通過多態調用自己的offer(e)實現*/
public boolean offer(E e) {
    checkNotNull(e);
    //加鎖
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        //如果隊列滿了，那么返回false
        if (count == items.length)
            return false;
        else {
            //入隊
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}
private void enqueue(E var1) {
    Object[] var2 = this.items;
    //putIndex可以認為是隊列的隊尾后的一個位置，數據入隊對應的位置，如果隊列滿了，那么putIndex設置為０
    var2[this.putIndex] = var1;
    if (++this.putIndex == var2.length) {
        this.putIndex = 0;
    }

    ++this.count;
    //喚醒一個等待在condition上的線程
    this.notEmpty.signal();
}
put(E e)　put操作，如果隊列已滿，那么會一直阻塞，直到put成功
public void put(E var1) throws InterruptedException {
    checkNotNull(var1);
    ReentrantLock var2 = this.lock;
    //加鎖，可被線程中斷返回
    var2.lockInterruptibly();
    try {
        //如果隊列已經滿了，那么阻塞
        while(this.count == this.items.length) {
            this.notFull.await();
        }
        //進行入隊操作
        this.enqueue(var1);
    } finally {
        var2.unlock();
    }
}
/**
 * 在隊列滿的情況put操作被阻塞，那么什么時候該操作可以被喚醒呢？很顯然是隊列中出現空地的情況下，才會被喚醒在notFull這種條件下 
 * 阻塞的操作：
 * 所以在發生以下操作的時候，會被喚醒進行入隊的操作
 * １、dequeue()操作　２、removeAt(int var1)操作　３、clear()　４、drainTo
 */
take() 出隊操作，如果隊列為空，那么阻塞，直到隊列中包含對應元素喚醒
/**實現比較容易，和上面的操作異曲同工*/
public E take() throws InterruptedException {
    ReentrantLock var1 = this.lock;
    var1.lockInterruptibly();

    Object var2;
    try {
        while(this.count == 0) {
            this.notEmpty.await();
        }

        var2 = this.dequeue();
    } finally {
        var1.unlock();
    }

    return var2;
}
個人總結：實現阻塞操作和核心在于線程掛起以及線程的喚醒，在Java中提供了兩種線程等待以及線程喚醒的方式。一是基于對象監視器的wait(),notify()方法。　二是通過Condition.await()和signal()方法。
2. LinkedBlockingQueue
基于鏈表實現的有界阻塞隊列。此隊列的默認和最大長度為Integer.MAX_VALUE。此隊列按照先進先出的原則對元素進行排序。這個隊列的實現原理和ArrayBlockingQueue實現基本相同。可以看一下隊列的定義：
隊列的定義
/**默認的構造函數*/
public LinkedBlockingQueue() {
    this(2147483647);
}

public LinkedBlockingQueue(int var1) {
    this.count = new AtomicInteger();
    this.takeLock = new ReentrantLock();
    this.notEmpty = this.takeLock.newCondition();
    this.putLock = new ReentrantLock();
    this.notFull = this.putLock.newCondition();
    if (var1 <= 0) {
        throw new IllegalArgumentException();
    } else {
        this.capacity = var1;
        //鏈表的頭結點和尾節點，默認是空
        this.last = this.head = new LinkedBlockingQueue.Node((Object)null);
    }
}
3、PriorityBlockingQueue
一個支持優先級的無界隊列。默認情況下元素采取自然順序排列，也可以通過比較器comparator來指定元素的排序規則。元素按照升序排列。具體是如何實現的？
隊列的定義以及構造方法
/**定義和通常的阻塞隊列相同，AbstractQueue中定義了隊列的基本操作，BlockingQueue中定義可阻塞隊列的相關操作定義*/
public class PriorityBlockingQueue extends AbstractQueue implements BlockingQueue
/**構造方法，默認的無參構造方法，調用的是另一個構造方法，默認定義了一個隊列的容量，那為什么說他是無界隊列呢？接著向下*/
public PriorityBlockingQueue() {
    this(DEFAULT_INITIAL_CAPACITY, null);
}
/**所有的構造方法最后調用的構造方法， comparator是一個比較器，通過比較器可以確定隊列中元素的排列順序*/
public PriorityBlockingQueue(int initialCapacity, Comparator comparator) {
    if (initialCapacity < 1)
        throw new IllegalArgumentException();
    this.lock = new ReentrantLock();
    this.notEmpty = lock.newCondition();
    this.comparator = comparator;
    /**隊列是基于數組實現的*/
    this.queue = new Object[initialCapacity];
}
add 操作以及offer操作
public boolean add(E e) {
    return offer(e);
}
/**
 * offer操作
 */
public boolean offer(E e) {
    if (e == null)
        throw new NullPointerException();
    final ReentrantLock lock = this.lock;
    lock.lock();
    int n, cap;
    Object[] array;
    //如果隊列已滿，那么嘗試進行擴容（個人感覺這里使用 >= 并不是很合理）
    while ((n = size) >= (cap = (array = queue).length))
        tryGrow(array, cap);
    try {
        Comparator cmp = comparator;
        if (cmp == null)
            //使用默認的比較方法將e放到隊列中
            siftUpComparable(n, e, array);
        else
            //使用指定的比較順序將數據插入到隊列中
            siftUpUsingComparator(n, e, array, cmp);
        size = n + 1;
        //激活一個在notEmpty這個condition上等待的線程
        notEmpty.signal();
    } finally {
        lock.unlock();
    }
    return true;
}
/**tryGrow()實現*/
private void tryGrow(Object[] array, int oldCap) {
    //這里先釋放了鎖，最后需要重新獲取鎖，那么這個時候所有的add操作都會執行下面的代碼段
    lock.unlock(); // must release and then re-acquire main lock
    Object[] newArray = null;
    if (allocationSpinLock == 0 &&
        UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
                                 0, 1)) {
        try {
            int newCap = oldCap + ((oldCap < 64) ?
                                   (oldCap + 2) : // grow faster if small
                                   (oldCap >> 1));
            if (newCap - MAX_ARRAY_SIZE > 0) {    // possible overflow
                int minCap = oldCap + 1;
                if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
                    throw new OutOfMemoryError();
                newCap = MAX_ARRAY_SIZE;
            }
            if (newCap > oldCap && queue == array)
                newArray = new Object[newCap];
        } finally {
            allocationSpinLock = 0;
        }
    }
    if (newArray == null) // back off if another thread is allocating
        Thread.yield();
    lock.lock();
    if (newArray != null && queue == array) {
        queue = newArray;
        System.arraycopy(array, 0, newArray, 0, oldCap);
    }
}

并發隊列