大家好，我是小黑，一個在互聯網茍且偷生的農民工，
佇列
學過資料結構的同學應該都知道，佇列是資料結構中一種特殊的線性表結構，和平時使用的List，Set這些資料結構相比有點特殊，它的特殊之處在于它只允許在佇列的頭部（Head）進行洗掉操作，在尾部（Tail）進行插入操作，這種方式的佇列我們稱之為先進先出佇列（FIFO），
在JDK1.5中推出了佇列這一資料結構的具體實作，介面Queue是對于佇列的定義，并有一些列具有特殊功能的佇列實作，
在Queue介面中定義了佇列的如下方法：
其中add(E)并非Queue介面新定義，而是從Collection介面繼承而來的，
阻塞佇列
BlockingQueue介面也是在JDK1.5中推出，存放在java.util.concurrent包中，繼承自Queue，所以在BlockingQueue中有Queue的所有方法，
從名字就可以看出BlockingQueue是一種阻塞佇列，它支持在檢索元素時如果佇列為空可以一直阻塞等待直到有元素可以獲取，同樣在添加元素時如果佇列已滿會阻塞等待佇列中有空閑的存盤空間，
BlockingQueue的方法可以歸納為四類：
- 在操作時如不能立即滿足，會直接拋出例外
- 在操作時如不能立即滿足，則回傳特殊的值，如插入、移除方法會回傳false，檢查方法會回傳null
- 在操作時如不能立即滿足，則會阻塞等待，直到操作成功
- 在操作時如不能立即滿足，則會阻塞等待給定的時間長度，時間到達后如果還不能滿足則回傳null
這四類方法總結如下，
因為在BlockingQueue的一些方法中，會通過null表示某種操作的失敗，所以不允許在BlockingQueue中存放null值元素，會在操作時拋出NullPointerExection例外，
BlockingQueue因為是一個容器嘛，所以它也有容量的限制，在具體實作類中有可以設定容量的實作類，也有不可以設定容量的實作類，不能設定容量的實作類容量默認為Integer.MAX_VALUE，
BlockingQueue是定義在java.util.concurrent包中，那么它在并發情況下到底是不是執行緒安全的呢？
在JDK提供的BlockingQueue的具體實作類中，上面表格中的方法實作都是執行緒安全的，在內部都使用了鎖或者其他形式的并發控制保證操作的原子性，
但是有一點要注意，就是一些批量處理的方法例如addAll、containsAll、retainAll和removeAll這些方法并不一定是執行緒安全的，使用時注意，
說完BlockingQueue介面我們接下來看看它都有哪些具體的實作呢？以及在它們內部是如何做到執行緒安全和阻塞的呢？
ArrayBlockingQueue
ArrayBlockingQueue是一個底層由陣列支持額有界阻塞佇列，
重要屬性
先來看看ArrayBlockingQueue中都有哪些屬性，
// 存放元素的陣列
final Object[] items;
// 用來記錄取元素的下標，用于下一次在take，poll,remove，peek方法中使用
int takeIndex;
// 用來記錄添加元素的下標，用于下一次put,offer,add等方法使用
int putIndex;
// 記錄佇列中元素數量
int count;
// 用于控制并發訪問時保證執行緒安全的鎖
final ReentrantLock lock;
// 用于佇列空時阻塞和喚醒等待執行緒的條件
private final Condition notEmpty;
// 用于佇列滿時阻塞和喚醒等待執行緒的條件
private final Condition notFull;
我們通過這些佇列中的屬性基本可以知道ArrayBlockingQueue中都有哪些重要資訊，可以看出ArrayBlockingQueue就是使用Object[]來存放元素的，
那么應該如何創建一個ArrayBlockingQueue呢？
構造方法
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
默認的構造方法需要傳入一個int型別的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();
}
從這個方法我們看出傳入的false表示會在內部用于創建一個ReentrantLock物件，我們都知道ReentrantLock支持公平和非公平的實作，我們猜想一下，這里的這個fair值是不是表示該阻塞佇列對于阻塞排隊的執行緒支持公平和非公平的策略呢？這里先賣個關子，在后面的方法中我們具體說，
除了這兩種創建的方式，ArrayBlockingQueue還支持傳入一個Collection集合，
public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {
// 先創建一個ArrayBlockingQueue實體
this(capacity, fair);
final ReentrantLock lock = this.lock;
lock.lock();
try {
int i = 0;
try {
// 回圈將collection中的元素放入queue中
for (E e : c) {
checkNotNull(e);
items[i++] = e;
}
} catch (ArrayIndexOutOfBoundsException ex) {
// 如果collection的元素個數超出queue的容量大小，會拋出例外
throw new IllegalArgumentException();
}
count = i;
putIndex = (i == capacity) ? 0 : i;
} finally {
lock.unlock();
}
}
添加元素
先來看看添加一個新元素到ArrayBlockingQueue是如何實作的，怎樣保證執行緒安全的，
add(e)
public boolean add(E e) {
// 呼叫父類中的add(e)方法
return super.add(e);
}
public boolean add(E e) {
// 這里會直接呼叫offer(e)方法，如果offer方法回傳false,則直接拋出例外
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
add方法的實作邏輯本質上是對offer方法套了一層殼，如果offer方法回傳false時，拋出例外，所以我們直接看offer方法的實作就好，
offer(e)
public boolean offer(E 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();
}
}
可以看到offer方法的邏輯還是比較簡單的，先檢查入參不能為空，然后加鎖保證入隊操作的原子性，在獲取鎖成功后入隊，如果佇列已滿則直接回傳false，所以offer方法并不會阻塞，
put(e)
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
// 可被中斷方式獲取鎖
lock.lockInterruptibly();
try {
while (count == items.length)
// 佇列滿時會阻塞
notFull.await();
// 入隊
enqueue(e);
} finally {
lock.unlock();
}
}
put方法和offer方法唯一的區別，就是會在佇列滿的時候使用Condition條件物件notFull阻塞等待，
private void enqueue(E x) {
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;
// 入隊成功，喚醒等待的移除元素操作執行緒
notEmpty.signal();
}
在enqueue方法中才會完成對佇列中的陣列元素的賦值動作，完成之后喚醒阻塞等待的移除元素操作執行緒，
offer(e,time,unit)
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) {
// 時間小于等于0時，回傳false
if (nanos <= 0)
return false;
// 阻塞等待指定時間
nanos = notFull.awaitNanos(nanos);
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}
offer(e,time,unit)方法與offer(e)方法相比，主要時多了一個等待時間，會在時間到達時如果沒有空間添加元素回傳false，
移除元素
ArrayBlockingQueue中移除元素的方法主要有remove()，poll()，take()，poll(time,unit)四個，這幾個方法的實作邏輯都比較簡單，這里不在單獨貼代碼 ，我們來看一下阻塞方法take()的實作即可，
take()
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
// 加鎖
lock.lockInterruptibly();
try {
while (count == 0)
// 如果元素數量==0，表示佇列中為空，則阻塞等待
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
dequeue()
private E dequeue() {
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
// 取出元素之后，喚醒其他等待執行緒，
notFull.signal();
return x;
}
LinkedBlockingQueue
LinkedBlockingQueue是一個基于鏈表結構的阻塞佇列，可以在創建時指定邊界大小，也可以不指定，在不指定邊界時容量為
Integer.MAX_VALUE，
重要屬性
我們先來看看在LinkedBlockingQueue中都有哪些重要的屬性，
// 內部類Node節點，用來存放鏈表中的元素
static class Node<E> {
// 節點元素
E item;
// 當前節點的下一個節點，如果為空表示沒有下一個節點
Node<E> next;
Node(E x) { item = x; }
}
// 佇列的容量
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 ReentrantLock putLock = new ReentrantLock();
// 控制消費者的條件
private final Condition notEmpty = takeLock.newCondition();
// 控制生產者的條件
private final Condition notFull = putLock.newCondition();
在LinkedBlockingQueue中使用Node來存放元素，和指向下一個節點的鏈表指標，
構造方法
在LinkedBlockingQueue的構造方法中，會創建一個創建一個不存放元素的Node物件賦值給head和last，
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
// 創建一個不存放元素的Node物件賦值給head和last
last = head = new Node<E>(null);
}
public LinkedBlockingQueue(Collection<? extends E> c) {
this(Integer.MAX_VALUE);
final ReentrantLock putLock = this.putLock;
putLock.lock();
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(e)
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
if (count.get() == capacity)
return false;
int c = -1;
Node<E> node = new Node<E>(e);
// 使用putLock加鎖
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
if (count.get() < capacity) {
// 入隊
enqueue(node);
// 數量+1
c = count.getAndIncrement();
if (c + 1 < capacity)
// 喚醒一個生產者執行緒
notFull.signal();
}
} finally {
putLock.unlock();
}
if (c == 0)
// 喚醒消費者執行緒
signalNotEmpty();
// 入隊失敗情況會回傳false
return c >= 0;
}
對于鏈表結構的LinkedBlockingQueue來說，入隊操作要簡單很多，只需要將node節點掛在最后一個節點last的next，然后將自己賦值給last，
private void enqueue(Node<E> node) {
last = last.next = node;
}
put(e)
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
int c = -1;
Node<E> node = new Node<E>(e);
// 使用putLock加鎖
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) {
// 如果佇列容量已使用完則阻塞
notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}
對比結果也和我們最開始的方法匯總表格一樣，offer(e)方法會在入隊時如果佇列已滿直接回傳false，而put(e)會一直阻塞等待，知道入隊成功，
add(e)方法和offer(e,time,unit)方法實作邏輯上沒有特殊之處，這里不再放原始碼，
移除元素
poll()
public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0)
return null;
E x = null;
int c = -1;
// 使用takeLock加鎖
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
if (count.get() > 0) {
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
// 還有元素時喚醒一個生產者執行緒
notEmpty.signal();
}
} finally {
takeLock.unlock();
}
if (c == capacity)
// 喚醒生產者執行緒
signalNotFull();
return x;
}
poll()方法會在元素出隊時如果沒有元素則直接回傳null，
// 出隊方法
private E dequeue() {
Node<E> h = head;
Node<E> first = h.next;
h.next = h;
head = first;
E x = first.item;
first.item = null;
return x;
}
take()
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
// 使用takeLock加鎖
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;
}
同樣，take方法會在沒有元素時一直等待，
對比
我們來對比一下ArrayBlockingQueue和LinkedBlockingQueue都有哪些區別，
- ArrayBlockingQueue基于陣列實作，LinkedBlockingQueue基于鏈表實作
- ArrayBlockingQueue在添加和移除元素的操作中共用一把鎖，LinkedBlockingQueue使用
takeLock和putLock兩把鎖 - ArrayBlockingQueue在添加和移除元素時直接使用元素的型別處理，LinkedBlockingQueue需要轉成Node物件
- ArrayBlockingQueue創建時必須指定容量，LinkedBlockingQueue可以不指定，默認容量為
Integer.MAX_VALUE
由于LinkedBlockingQueue使用兩把鎖將入隊操作和出隊操作分離，這會大大提高佇列的吞吐量，在高并發情況下生產者和消費者可以并行處理，提高并發性能，
但是LinkedBlockingQueue默認是無界佇列，要小心記憶體溢位風險，所以最好在創建時指定容量大小，
BlockingQueue介面的實作類除了本期介紹的這兩種，還有PriorityBlockingQueue，SynchronousQueue，LinkedBlockingDeque等，每一個都有它獨特的特性和使用場景，后面我們再單獨深入決議，
好的，本期內容就到這里，我們下期見，關注我的公眾號【小黑說Java】,更多干貨內容，
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/298844.html
標籤：其他