本文作者：王一飛，叩丁狼高級講師。原創(chuàng)文章，轉(zhuǎn)載請注明出處。
概念
LinkedBlockingQueue按照api解釋：一個基于鏈表而實現(xiàn)的有界阻塞隊列。遵循先進先出原則，由隊頭入列，再從隊尾出列。具體操作上跟ArrayBlockingQueue類似，區(qū)別在于底層維護數(shù)據(jù)上，LinkedBlockingQueue底層是一個鏈接，而ArrayBlockingQueue是一個數(shù)組。
內(nèi)部結(jié)構(gòu)
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private final AtomicInteger count = new AtomicInteger(); //隊列元素個數(shù)
private final int capacity; //隊列容器
transient Node<E> head; //隊頭
private transient Node<E> last; //隊尾
//出列入列過程中維護現(xiàn)場安全的各類鎖
private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();
//隊列數(shù)據(jù)節(jié)點
static class Node<E> {
E item;
Node<E> next;
Node(E x) { item = x; }
}
}
基本操作
public class App {
public static void main(String[] args) throws InterruptedException {
LinkedBlockingQueue<String> queue = new LinkedBlockingQueue(5);
//入列
queue.add("a"); //隊列滿后拋異常
queue.put("b");//隊列滿后阻塞
queue.offer("c"); //入列失敗返回false
System.out.println(queue);
queue.put("a");
queue.put("b");
queue.put("c");
queue.put("d");
queue.put("e");
//出列
queue.remove("a"); //刪除指定元素
queue.poll(); //出列，如果隊列為空返回null
queue.take(); //隊列為空，阻塞等待
System.out.println(queue);
}
}
一般推薦使用put入列， take出列
源碼解析
構(gòu)造-LinkedBlockingQueue提供了3個構(gòu)造器，無參， 帶有容量，帶集合數(shù)據(jù)
//無參數(shù)時，默認容量為int的最大值
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
//帶容量參數(shù)【推薦】
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
//初始化隊頭，隊尾
last = head = new Node<E>(null);
}
//帶數(shù)據(jù)集合
public LinkedBlockingQueue(Collection<? extends E> c) {
this(Integer.MAX_VALUE); //容量為int最大值
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();
}
}
LinkedBlockingQueue 3個構(gòu)造器，實際使用中更推薦使用指定容量的隊列。
在繼續(xù)看源碼前，先了解一個原子操作類：AtomicInteger
//int 類型的原子操作，不指定初始為0
//底層維護了一個volatile 修飾變量 value = 0
AtomicInteger atomicInteger = new AtomicInteger();
//獲?。簐alue = 0
System.out.println(atomicInteger.get()); //0
//返回value值0， 然后value 值加一【這里也是原子操作】
System.out.println(atomicInteger.getAndIncrement()); //0
//返回value值1， 然后value 值加一【這里也是原子操作】
System.out.println(atomicInteger.getAndIncrement()); //1
//獲?。簐alue=2
System.out.println(atomicInteger.get()); //2
//int 類型的原子操作，不指定初始為0
//底層維護了一個volatile 修飾變量 value = 0
AtomicInteger atomicInteger = new AtomicInteger();
//獲取：value = 0
System.out.println(atomicInteger.get()); //0
//返回value值0， 然后value 值減一【這里也是原子操作】
System.out.println(atomicInteger.getAndDecrement()); //0
//返回value值1， 然后value 值減一【這里也是原子操作】
System.out.println(atomicInteger.getAndDecrement()); //-1
//返回value值1
System.out.println(atomicInteger.get()); //-2
getAndIncrement ： 返回未操作前value 的值， 然后加1
getAndDecrement ： 返回未操作前value 的值， 然后減1
入列-put：將入列數(shù)據(jù)添加到隊尾，如果隊列滿了，阻塞等待。
public void put(E e) throws InterruptedException {
//入列元素不允許為null
if (e == null) throw new NullPointerException();
//隊列臨時容量緩存，作為執(zhí)行喚醒/阻塞線程操作標記
int c = -1;
Node<E> node = new Node<E>(e);
//入列鎖
final ReentrantLock putLock = this.putLock;
//隊列元素個數(shù)
final AtomicInteger count = this.count;
putLock.lockInterruptibly(); //入列前加鎖，可中斷鎖
try {
//自旋排除硬件加鎖延時問題
//如果隊列已滿，線程阻塞等待
while (count.get() == capacity) {
notFull.await();
}
//數(shù)據(jù)入列
enqueue(node);
//原子操作，入列后再獲取隊列元素個數(shù)，并+1，確保當(dāng)前操作隊列元素個數(shù)最新
c = count.getAndIncrement();
//c + 1 < capacity 表示隊列未滿，仍可添加，喚醒未持鎖而等待的入列線程
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock(); //釋放鎖
}
//c == 0 說明隊列為空，喚醒入列線程入列
if (c == 0)
signalNotEmpty();
}
從上面源碼上看put方法其實做5件事：
1>判斷元素是否null， 為null 拋異常
2>判斷是否滿列，滿列則等待，此處需要留意while這個操作。
理想請求下，if即可，但是有種情況，如果jvm執(zhí)行指令傳到cpu到程序時間片執(zhí)行存在一點的時間延時，while 重復(fù)執(zhí)行，可以減少延時影響。
3>數(shù)據(jù)入列
4>入列后需要喚醒未持鎖而等待的入列線程
5>c==0的判斷， c的值是入列前容量值，如果為0說明入列前，隊列為空，可以存在出列等待線程，所以在c==0的時候，且已經(jīng)入列成功，可以喚醒出列等待線程，讓其順利出列。
出列-take : 出列，從隊頭彈出元素， 如果隊列個數(shù)為0， 阻塞等待。
public E take() throws InterruptedException {
E x;
//隊列臨時容量緩存，作為執(zhí)行喚醒/阻塞線程操作標記
int c = -1;
//隊列元素個數(shù)
final AtomicInteger count = this.count;
//出列鎖
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();//出列前加鎖，可中斷鎖
try {
//自旋排除硬件加鎖延時問題
//如果隊列已空，線程阻塞等待
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue(); //數(shù)據(jù)出列
//原子操作，出列后再獲取隊列元素個數(shù)，并-1，確保當(dāng)前操作隊列元素個數(shù)最新
c = count.getAndDecrement();
//c > 1 表示隊列未空，仍可出列，喚醒未持鎖而等待的出列線程
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();//釋放鎖
}
//c == capacity 說明隊列已滿，喚醒出列線程出列
if (c == capacity)
signalNotFull();
return x;
}
上面源碼看出，take操作跟put原理差不多，執(zhí)行的是反向操作而已。需要注意的是take方法喚醒線程條件c 變量值判斷條件。
1> 出列成功之后， c > 1 表示出列前隊列至少有2個元素，所以出列成功后，喚醒未持鎖而等待的出列線程
2>c == capacity 滿足這個條件， 表示出列前，隊列是滿的，可能存在入列等待線程，出列成功之后，解除等待，喚醒入列等待線程。
刪除-remove : 根據(jù)指定元素刪除隊列中的元素，如果有刪除，如果沒有返回false
public boolean remove(Object o) {
if (o == null) return false; //參數(shù)為null 直接返回
fullyLock(); //為保證入列出列線程安全，加雙鎖
try {
//循環(huán)遍歷列表，刪除指定元素
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 unlink(Node<E> p, Node<E> trail) {
p.item = null;
trail.next = p.next;
if (last == p)
last = trail;
//刪除元素后，隊列元素個數(shù)-1，喚醒入列等待線程
if (count.getAndDecrement() == capacity)
notFull.signal();
}
void fullyUnlock() {
takeLock.unlock();
putLock.unlock();
}
LinkedBlockingQueue 在執(zhí)行刪除操作，需要對takeLock 跟 putLock同時加鎖，其目標確保在刪除期間，其他線程無法操作隊列，進而保證刪除操作的線程安全。
總結(jié)
LinkedBlockingQueue 使用2把鎖確保線程安全，入列時使用putLock，出列時使用takeLock，這種鎖分離操作機制，在一定層面上提高隊列的吞吐量，在高并發(fā)的情況下生產(chǎn)者(入列線程)和消費者(出列線程)可以并行地操作隊列中的數(shù)據(jù)，以此來提高整個隊列的并發(fā)性能。
想獲取更多技術(shù)干貨，請前往叩丁狼官網(wǎng)：http://www.wolfcode.cn/all_article.html
