上一篇我們看了一下這個隊列ConcurrentLinkedQueue，那就是一個無界非阻塞鏈表，我們這次來看看LinkedBlockingQueue，這個隊列看名字就知道是一個阻塞式隊列(也就是一個單向鏈表)，基於獨占鎖實現的，比較簡單；
一.LinkedBlockingQueue基本結構
內部也是有一個Node類，下圖所示，item存 實際數據，next指向下一個節點，一個有參構造器，沒啥好說的；
我們可以看看這個隊列有的一些屬性，其實大概能猜出來就是生產者消費者模型：
//隊列實際容量 private final int capacity; //這個原子變量記錄節點數量 private final AtomicInteger count = new AtomicInteger(); //頭節點 transient Node<E> head; //尾節點 private transient Node<E> last; //這個鎖用於控制多個線程從隊列頭部獲取元素 private final ReentrantLock takeLock = new ReentrantLock(); //當隊列為空，執行出隊操作的線程就放到這條件變量里來 private final Condition notEmpty = takeLock.newCondition(); //用於控制多個線程往隊列尾部添加元素 private final ReentrantLock putLock = new ReentrantLock(); //如果隊列滿了，執行入隊操作的線程就丟到這里面來 private final Condition notFull = putLock.newCondition();
構造器中可以看到，默認最大數量就是65536個，雖然說也可以指定大小，我們一定程度上可以說這是一個有界阻塞隊列；
//默認隊列最大的數量就是65536個 public LinkedBlockingQueue() { this(Integer.MAX_VALUE); } //也可以指定隊列大小，默認頭節點和尾節點都是指向哨兵節點 public LinkedBlockingQueue(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); this.capacity = capacity; last = head = new Node<E>(null); } //也可以傳一個實現Collection接口的類，比如List，然后遍歷將其中的元素封裝成節點丟到隊列中去 //注意這里獲取鎖和釋放鎖 public LinkedBlockingQueue(Collection<? extends E> c) { this(Integer.MAX_VALUE); final ReentrantLock putLock = this.putLock; putLock.lock(); // Never contended, but necessary for visibility try { int n = 0; for (E e : c) { if (e == null) throw new NullPointerException(); if (n == capacity) throw new IllegalStateException("Queue full"); enqueue(new Node<E>(e)); ++n; } count.set(n); } finally { putLock.unlock(); } }
我們簡單的介紹了這個隊列的基本結構，現在我們可以看看一些重要的方法；
二.offer方法
這個方法向隊列最后添加一個元素，插入成功返回true，如果隊列滿了，就拋棄當前元素返回false；
public boolean offer(E e) { //如果為null，就直接拋錯 if (e == null) throw new NullPointerException(); //count表示隊列中實際數量 final AtomicInteger count = this.count; //如果實際數量和隊列最大容量相同，那么就不能再添加了，返回false if (count.get() == capacity) return false; int c = -1; //將元素封裝成Node節點 Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; //獲取鎖 putLock.lock(); try { //隊列沒有滿，就把新的節點丟進去，遞增計數器，為什么這里要進行判斷呢？上面不是進行判斷了么？ //還是由於並發，如果在上面那里判斷容量之后，當前線程還沒有獲取鎖，此時一個其他線程先獲取鎖然后執行offer方法然后釋放鎖，那么 //這里需要再判斷一次 if (count.get() < capacity) { enqueue(node);
//注意getAndIncrement和incrementAndGet方法的區別，前者是返回自增之前的值，后者是返回自增之后的值 c = count.getAndIncrement(); //這里判斷如果隊列還沒有滿，就喚醒之前notFully條件隊列中的線程，前面說過了notfull中存放的是什么線程 if (c + 1 < capacity) notFull.signal(); } } finally { //釋放鎖 putLock.unlock(); } //如果c==0，表示隊列中在添加節點之前就已經有一個節點了，就喚醒條件變量notEmpty中的線程，這些線程就會從隊列中去取數據 if (c == 0) signalNotEmpty(); return c >= 0; } private void signalNotEmpty() { final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { notEmpty.signal(); } finally { takeLock.unlock(); } }
三.put方法
這個方法在隊列尾部插入一個元素，如果隊列有空閑則插入后直接返回，否則就阻塞當前線程直到隊列空閑再插入；而且當前線程在阻塞的時候被其他線程調用了中斷方法，就會拋異常；
public void put(E e) throws InterruptedException { //如果插入的元素為null，直接拋異常 if (e == null) throw new NullPointerException(); int c = -1; //封裝成一個節點 Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; //原子變量記錄隊列中節點數量 final AtomicInteger count = this.count; //這個方法后面帶有Interruptibly，說明當前線程獲取鎖后是可中斷的 putLock.lockInterruptibly(); try { //節點數量到達了最大容量，就將當前線程放到條件變量notFull的隊列中 while (count.get() == capacity) { notFull.await(); } //節點數量沒有到最大，就在鏈表最后添加節點 enqueue(node); //計數器加一，注意如果count等於4，那么c還是等於4，這個方法是原子自增，返回原來的值，注意和incrementAndGet方法的區別 c = count.getAndIncrement(); //這里如果c+1<capacity的話，說明隊列還沒有滿，就喚醒notFull中的線程可以往隊列中添加元素 if (c + 1 < capacity) notFull.signal(); } finally { //釋放鎖 putLock.unlock(); } //如果c==0表示隊列中添加節點之前就已經有了一個節點，喚醒notEmpty中的線程去隊列中取數據 if (c == 0) signalNotEmpty(); } private void signalNotEmpty() { final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { notEmpty.signal(); } finally { takeLock.unlock(); } }
四.poll方法
這個方法是從頭部移除一個元素，如果隊列為空就返回null，這個方法不阻塞；
public E poll() { //記錄隊列中節點數量 final AtomicInteger count = this.count; //如果隊列為空就返回null if (count.get() == 0) return null; E x = null; int c = -1; final ReentrantLock takeLock = this.takeLock; //獲取鎖 takeLock.lock(); try { //這里再判斷一次也是為了防止在獲取鎖之前其他線程調用了poll方法取了節點 //如果隊列不為空，計數器減一 if (count.get() > 0) { //刪除第一個有數據的節點（由於第一個節點是哨兵節點，所以相當於刪除的是第二個節點），方法實現在下面 x = dequeue(); c = count.getAndDecrement(); //如果c>1說明移除頭節點之后，隊列不為空，就喚醒notEmpty中條件隊列中的線程去隊列中取數據 if (c > 1) notEmpty.signal(); } } finally { //釋放鎖 takeLock.unlock(); } //這里這個判斷，注意一下在原子類AtomicInteger中兩個方法，比如初始值為5，那么調用decrementAndGet方法返回的事4， //而調用getAndDecrement方法返回的是5，我們這里額c調用的是后者，所以表示刪除節點之前隊列的數量 //所以這里的意思就是：如果刪除隊列之前的數量等於隊列最大容量，那么刪除之后隊列肯定有空位置，於是就喚醒notFull條件隊列中的線程 //往隊列中添加新的節點 if (c == capacity) signalNotFull(); return x; } //這個方法很簡單稍微提一下，就是將第一個哨兵節點自己引用自己，更好的被gc收集 //將head指向第二個節點，取出該節點的值，然后將該節點內的item置為null，此節點就變成了一個哨兵節點 private E dequeue() { //刪除頭節點 Node<E> h = head; Node<E> first = h.next; h.next = h; // help GC head = first; E x = first.item; first.item = null; return x; } private void signalNotFull() { final ReentrantLock putLock = this.putLock; putLock.lock(); try { notFull.signal(); } finally { putLock.unlock(); } }
五.peek方法
這個方法獲取隊列頭部元素但是不移除該節點；
public E peek() { //隊列為空，返回null if (count.get() == 0) return null; final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { //如果頭節點的下一個節點是空，就返回null，因為此時頭節點指的是哨兵節點啊 Node<E> first = head.next; if (first == null) return null; else //頭節點的下一個節點不為空，那么下一個節點肯定有數據，拿過來就行了 return first.item; } finally { takeLock.unlock(); } }
六.take方法
當前方法跟peek方法基本一樣，只不過這個方法是阻塞的：從隊列頭刪除一個節點，如果隊列為空則阻塞當前線程直到隊列不為空再執行操作，如果在阻塞的時候其他線程修改了中斷標志，那么該線程就拋錯；
//這個方法其實和poll方法基本一樣，沒什么好說的，注意可以拋異常 public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; //注意這里獲取鎖的方式 takeLock.lockInterruptibly(); try { while (count.get() == 0) { notEmpty.await(); } x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } finally { takeLock.unlock(); } if (c == capacity) signalNotFull(); return x; }
七.remove方法
刪除隊列中某個指定的元素，刪除成功返回true，失敗返回false；
public boolean remove(Object o) { //傳入的是null，直接返回false if (o == null) return false; //獲取兩個鎖，方法實現如下，在刪除節點的時候，不能進行入隊和出隊操作，那些線程會被阻塞丟到AQS隊列里 fullyLock(); try { //遍歷找到對應的節點，刪除掉，沒找到返回false for (Node<E> trail = head, p = trail.next; p != null; trail = p, p = p.next) { if (o.equals(p.item)) { //刪除節點的具體方法，實現在下面 unlink(p, trail); return true; } } return false; } finally { //釋放兩個鎖，注意釋放鎖的順序要和獲取鎖的順序相反喲 fullyUnlock(); } } void fullyLock() { putLock.lock(); takeLock.lock(); } void fullyUnlock() { takeLock.unlock(); putLock.unlock(); } //刪除其實很容易，就是跟普通鏈表的刪除一樣，就是把當前要刪除的節點前面的節點指向后面的節點 void unlink(Node<E> p, Node<E> trail) { p.item = null; trail.next = p.next; if (last == p) last = trail; //還是注意getAndDecrement方法返回的是減一之前的值，如果減一之后隊列不是滿的，就喚醒notFull中條件隊列中的線程往隊列中添加節點 if (count.getAndDecrement() == capacity) notFull.signal(); }
八.總結
我們簡單的看了看LinkedBlockingQueue中一些比較重要的方法，比上一篇的ConcurrentLinkedQueue容易好多呀！
其中ConcurrentLinkedQueue是無界非阻塞隊列，底層是用單向鏈表實現，入隊和出隊使用CAS實現；而LinkedBlockingQueue是有界阻塞隊列，底層是用單向鏈表實現，入隊和出隊分別用獨占鎖的方式去處理，所以入隊和出隊是可以同時進行的，而且還為兩個獨占鎖配置了兩個條件隊列，用於存放被阻塞的線層，注意，這里涉及到好幾個隊列，一個是獨占鎖的AQS隊列，一個是條件隊列，一個是存放數據的隊列，不要弄混淆了啊！
用下面這個圖強化記憶：
