概述
LinkedBlockingQueue也是一個阻塞隊列，相比於ArrayBlockingQueue，他的底層是使用鏈表實現的，而且是一個可有界可無界的隊列，在生產和消費的時候使用了兩把鎖，提高並發，是一個高效的阻塞隊列，下面就分析一下這個隊列的源碼。
屬性
//鏈表節點定義
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;
/** Lock held by take, poll, etc */
//消費者的鎖
private final ReentrantLock takeLock = new ReentrantLock();
/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();
/** Lock held by put, offer, etc */
//生產者的鎖
private final ReentrantLock putLock = new ReentrantLock();
/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();
構造方法
//默認構造方法，無界
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);
}
消費者常用方法
take()方法
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; }
進入dequeue()方法
//通過這個方法可以看出，鏈表的首節點的值是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;
}
poll()方法
public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0)
return null;
E x = null;
int c = -1;
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(long timeout, TimeUnit unit)
這個方法和上面的區別就是加入了時延，在規定的時間沒有消費成功，就返回失敗。
生產者常用方法
add()方法
public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
直接調用父類AbstractQueue的方法
offer(E 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);
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
//雙重判斷
if (count.get() < capacity) {
//加入鏈表
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
//喚醒生產者線程，繼續插入
notFull.signal();
}
} finally {
putLock.unlock();
}
if (c == 0)
//說明里面有一個元素，喚醒消費者
signalNotEmpty();
return c >= 0;
}
進入enqueue()方法
private void enqueue(Node<E> node) {
// assert putLock.isHeldByCurrentThread();
// assert last.next == null;
last = last.next = node;
}
直接放到鏈表的尾部
offer(E e, long timeout, TimeUnit unit)
和poll(E e,long timeout,TimeUnit unit)相反。
put(E e)方法
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
// Note: convention in all put/take/etc is to preset local var
// holding count negative to indicate failure unless set.
int c = -1;
Node<E> node = new Node<E>(e);
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();
}
總結
總體來說比較簡單，下面就列一下LindedBlockingQueue的特點：
- 生產者和消費者采用不同的鎖控制，提高並發效率
- 底層采用鏈表存儲，構造方法中可以傳入隊列的容量，默認為無界
- 鏈表的首節點是一個空節點