Java線程隊(duì)列LinkedBlockingQueue的使用
一、定義
LinkedBlockingQueue繼承自AbstractQueue,實(shí)現(xiàn)了BlockingQueue,Serializable接口。
- LinkedBlockingQueue是一個(gè)基于已鏈接節(jié)點(diǎn)的,范圍任意的blocking queue
- 此隊(duì)列按FIFO(先進(jìn)先出)排序元素
- 新元素插入到隊(duì)列的尾部,并且隊(duì)列獲取操作會(huì)獲得位于隊(duì)列頭部的元素
- 鏈接隊(duì)列的吞吐量通常要高于基于數(shù)組的對(duì)列(ArrayBlockingQueue),但是在大多數(shù)并發(fā)應(yīng)用程序中,其可預(yù)知的性能要低
- 可選的容量范圍構(gòu)造方法參數(shù)作為防止隊(duì)列過度擴(kuò)展的一種方法,如果未指定容量,則等于Integer.MAX_VALUE,除非插入節(jié)點(diǎn)會(huì)使隊(duì)列超出容量,否則每次插入后會(huì)動(dòng)態(tài)地創(chuàng)建鏈接節(jié)點(diǎn)
二、比對(duì)分析

阻塞隊(duì)列大致可以分為這幾種:ArrayBlockingQueue,LinkedBlockingQueue,ConcurrentLinkedQueue,DelayQueue,LinkedTransferQueue,SynchronusQueue。
ArrayBlockingQueue--數(shù)組實(shí)現(xiàn)的有界隊(duì)列
會(huì)自動(dòng)阻塞,根據(jù)調(diào)用api不同,有不同特性,當(dāng)隊(duì)列容量不足時(shí),有阻塞能力。
boolean add(E e):在容量不足時(shí),拋出異常。
void put(E e):在容量不足時(shí),阻塞等待。
boolean offer(E e):不阻塞,容量不足時(shí)返回false,當(dāng)前新增數(shù)據(jù)操作放棄。
boolean offer(E e, long timeout, TimeUnit unit):容量不足時(shí),阻塞times時(shí)長(單位為timeunit),如果在阻塞時(shí)長內(nèi),有容量空閑,新增數(shù)據(jù)返回true。如果阻塞時(shí)長范圍內(nèi),無容量空閑,放棄新增數(shù)據(jù),返回false。
LinkedBlockingQueue--鏈?zhǔn)疥?duì)列,隊(duì)列容量不足或?yàn)?時(shí)自動(dòng)阻塞
void put(E e):自動(dòng)阻塞,隊(duì)列容量滿后,自動(dòng)阻塞。
E take():自動(dòng)阻塞,隊(duì)列容量為0后,自動(dòng)阻塞。
ConcurrentLinkedQueue--基礎(chǔ)鏈表同步隊(duì)列
boolean offer(E e):入隊(duì)。
E peek():查看queue中的首數(shù)據(jù)。
E poll():取出queue中的首數(shù)據(jù)。
DelayQueue--延時(shí)隊(duì)列
根據(jù)比較機(jī)制,實(shí)現(xiàn)自定義處理順序的隊(duì)列。常用于定時(shí)任務(wù),如:定時(shí)關(guān)機(jī)。
int compareTo(Delayed o):比較大小,自動(dòng)升序。
比較方法建議和getDelay方法配合完成。如果在DelayQueue是需要按時(shí)完成的計(jì)劃任務(wù),必須配合getDelay方法完成。
long getDelay(TimeUnit unit):獲取計(jì)劃時(shí)長的方法,根據(jù)參數(shù)TimeUnit來決定,如何返回結(jié)果值。
LinkedTransferQueue--轉(zhuǎn)移隊(duì)列
boolean add(E e):隊(duì)列會(huì)保存數(shù)據(jù),不做阻塞等待。
void transfer(E e):是TransferQueue的特有方法。必須有消費(fèi)者(take()方法調(diào)用者)。如果沒有任意線程消費(fèi)數(shù)據(jù),transfer方法阻塞。一般用于處理及時(shí)消息。
SynchronousQueue--同步隊(duì)列,容量為0
是特殊的TransferQueue,必須先有消費(fèi)線程等待,才能使用的隊(duì)列。
boolean add(E e):父類方法,無阻塞,若沒有消費(fèi)線程阻塞等待數(shù)據(jù),則拋出異常。
put(E e):有阻塞,若沒有消費(fèi)線程阻塞等待數(shù)據(jù),則阻塞。
三、源碼解讀
屬性
/**
* 節(jié)點(diǎn)類,用于存儲(chǔ)數(shù)據(jù)
*/
static class Node<E> {
E item;
Node<E> next;
Node(E x) { item = x; }
}
/** 阻塞隊(duì)列的大小,默認(rèn)為Integer.MAX_VALUE */
private final int capacity;
/** 當(dāng)前阻塞隊(duì)列中的元素個(gè)數(shù) */
private final AtomicInteger count = new AtomicInteger();
/**
* 阻塞隊(duì)列的頭結(jié)點(diǎn)
*/
transient Node<E> head;
/**
* 阻塞隊(duì)列的尾節(jié)點(diǎn)
*/
private transient Node<E> last;
/** 獲取并移除元素時(shí)使用的鎖,如take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();
/** notEmpty條件對(duì)象,當(dāng)隊(duì)列沒有數(shù)據(jù)時(shí)用于掛起執(zhí)行刪除的線程 */
private final Condition notEmpty = takeLock.newCondition();
/** 添加元素時(shí)使用的鎖如 put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();
/** notFull條件對(duì)象,當(dāng)隊(duì)列數(shù)據(jù)已滿時(shí)用于掛起執(zhí)行添加的線程 */
private final Condition notFull = putLock.newCondition();從上面的屬性我們知道,每個(gè)添加到LinkedBlockingQueue隊(duì)列中的數(shù)據(jù)都將被封裝成Node節(jié)點(diǎn),添加的鏈表隊(duì)列中,其中head和last分別指向隊(duì)列的頭結(jié)點(diǎn)和尾結(jié)點(diǎn)。與ArrayBlockingQueue不同的是,LinkedBlockingQueue內(nèi)部分別使用了takeLock 和 putLock 對(duì)并發(fā)進(jìn)行控制,也就是說,添加和刪除操作并不是互斥操作,可以同時(shí)進(jìn)行,這樣也就可以大大提高吞吐量。
這里如果不指定隊(duì)列的容量大小,也就是使用默認(rèn)的Integer.MAX_VALUE,如果存在添加速度大于刪除速度時(shí)候,有可能會(huì)內(nèi)存溢出,這點(diǎn)在使用前希望慎重考慮。
另外,LinkedBlockingQueue對(duì)每一個(gè)lock鎖都提供了一個(gè)Condition用來掛起和喚醒其他線程。
構(gòu)造函數(shù)
public LinkedBlockingQueue() {
// 默認(rèn)大小為Integer.MAX_VALUE
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
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();
}
}默認(rèn)的構(gòu)造函數(shù)和最后一個(gè)構(gòu)造函數(shù)創(chuàng)建的隊(duì)列大小都為Integer.MAX_VALUE,只有第二個(gè)構(gòu)造函數(shù)用戶可以指定隊(duì)列的大小。第二個(gè)構(gòu)造函數(shù)最后初始化了last和head節(jié)點(diǎn),讓它們都指向了一個(gè)元素為null的節(jié)點(diǎn)。

最后一個(gè)構(gòu)造函數(shù)使用了putLock來進(jìn)行加鎖,但是這里并不是為了多線程的競(jìng)爭(zhēng)而加鎖,只是為了放入的元素能立即對(duì)其他線程可見。
方法
同樣,LinkedBlockingQueue也有著和ArrayBlockingQueue一樣的方法,我們先來看看入隊(duì)列的方法。
入隊(duì)方法
LinkedBlockingQueue提供了多種入隊(duì)操作的實(shí)現(xiàn)來滿足不同情況下的需求,入隊(duì)操作有如下幾種:
- void put(E e);
- boolean offer(E e);
- boolean offer(E e, long timeout, TimeUnit unit)。
put(E e)
public void put(E e) throws InterruptedException {
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;
// 獲取鎖中斷
putLock.lockInterruptibly();
try {
//判斷隊(duì)列是否已滿,如果已滿阻塞等待
while (count.get() == capacity) {
notFull.await();
}
// 把node放入隊(duì)列中
enqueue(node);
c = count.getAndIncrement();
// 再次判斷隊(duì)列是否有可用空間,如果有喚醒下一個(gè)線程進(jìn)行添加操作
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
// 如果隊(duì)列中有一條數(shù)據(jù),喚醒消費(fèi)線程進(jìn)行消費(fèi)
if (c == 0)
signalNotEmpty();
}小結(jié)put方法來看,它總共做了以下情況的考慮:
- 隊(duì)列已滿,阻塞等待。
- 隊(duì)列未滿,創(chuàng)建一個(gè)node節(jié)點(diǎn)放入隊(duì)列中,如果放完以后隊(duì)列還有剩余空間,繼續(xù)喚醒下一個(gè)添加線程進(jìn)行添加。如果放之前隊(duì)列中沒有元素,放完以后要喚醒消費(fèi)線程進(jìn)行消費(fèi)。
很清晰明了是不是?
我們來看看該方法中用到的幾個(gè)其他方法,先來看看enqueue(Node node)方法:
private void enqueue(Node<E> node) {
last = last.next = node;
}該方法可能有些同學(xué)看不太懂,我們用一張圖來看看往隊(duì)列里依次放入元素A和元素B,畢竟無圖無真相:

接下來我們看看signalNotEmpty,順帶著看signalNotFull方法。
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}為什么要這么寫?因?yàn)閟ignal的時(shí)候要獲取到該signal對(duì)應(yīng)的Condition對(duì)象的鎖才行。
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 {
// 隊(duì)列有可用空間,放入node節(jié)點(diǎn),判斷放入元素后是否還有可用空間,
// 如果有,喚醒下一個(gè)添加線程進(jìn)行添加操作。
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;
}可以看到offer僅僅對(duì)put方法改動(dòng)了一點(diǎn)點(diǎn),當(dāng)隊(duì)列沒有可用元素的時(shí)候,不同于put方法的阻塞等待,offer方法直接方法false。
offer(E e, long timeout, TimeUnit unit)
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
int c = -1;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
// 等待超時(shí)時(shí)間nanos,超時(shí)時(shí)間到了返回false
while (count.get() == capacity) {
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
enqueue(new Node<E>(e));
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return true;
}出隊(duì)方法
入隊(duì)列的方法說完后,我們來說說出隊(duì)列的方法。LinkedBlockingQueue提供了多種出隊(duì)操作的實(shí)現(xiàn)來滿足不同情況下的需求,如下:
- E take();
- E poll();
- E poll(long timeout, TimeUnit unit);
take()
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
// 隊(duì)列為空,阻塞等待
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement();
// 隊(duì)列中還有元素,喚醒下一個(gè)消費(fèi)線程進(jìn)行消費(fèi)
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
// 移除元素之前隊(duì)列是滿的,喚醒生產(chǎn)線程進(jìn)行添加元素
if (c == capacity)
signalNotFull();
return x;
}take方法看起來就是put方法的逆向操作,它總共做了以下情況的考慮:
- 隊(duì)列為空,阻塞等待。
- 隊(duì)列不為空,從隊(duì)首獲取并移除一個(gè)元素,如果消費(fèi)后還有元素在隊(duì)列中,繼續(xù)喚醒下一個(gè)消費(fèi)線程進(jìn)行元素移除。如果放之前隊(duì)列是滿元素的情況,移除完后要喚醒生產(chǎn)線程進(jìn)行添加元素。
我們來看看dequeue方法
可能有些童鞋鏈表算法不是很熟悉,我們可以結(jié)合注釋和圖來看就清晰很多了。

其實(shí)這個(gè)寫法看起來很繞,我們其實(shí)也可以這么寫:
private E dequeue() {
// 獲取到head節(jié)點(diǎn)
Node<E> h = head;
// 獲取到head節(jié)點(diǎn)指向的下一個(gè)節(jié)點(diǎn),也就是節(jié)點(diǎn)A
Node<E> first = h.next;
// 獲取到下下個(gè)節(jié)點(diǎn),也就是節(jié)點(diǎn)B
Node<E> next = first.next;
// head的next指向下下個(gè)節(jié)點(diǎn),也就是圖中的B節(jié)點(diǎn)
h.next = next;
// 得到節(jié)點(diǎn)A的值
E x = first.item;
first.item = null; // help GC
first.next = first; // help GC
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方法去除了take方法中元素為空后阻塞等待這一步驟,這里也就不詳細(xì)說了。同理,poll(long timeout, TimeUnit unit)也和offer(E e, long timeout, TimeUnit unit)一樣,利用了Condition的awaitNanos方法來進(jìn)行阻塞等待直至超時(shí)。這里就不列出來說了。
獲取元素方法
public E peek() {
if (count.get() == 0)
return null;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
Node<E> first = head.next;
if (first == null)
return null;
else
return first.item;
} finally {
takeLock.unlock();
}
}加鎖后,獲取到head節(jié)點(diǎn)的next節(jié)點(diǎn),如果為空返回null,如果不為空,返回next節(jié)點(diǎn)的item值。
刪除元素方法
public boolean remove(Object o) {
if (o == null) return false;
// 兩個(gè)lock全部上鎖
fullyLock();
try {
// 從head開始遍歷元素,直到最后一個(gè)元素
for (Node<E> trail = head, p = trail.next;
p != null;
trail = p, p = p.next) {
// 如果找到相等的元素,調(diào)用unlink方法刪除元素
if (o.equals(p.item)) {
unlink(p, trail);
return true;
}
}
return false;
} finally {
// 兩個(gè)lock全部解鎖
fullyUnlock();
}
}
void fullyLock() {
putLock.lock();
takeLock.lock();
}
void fullyUnlock() {
takeLock.unlock();
putLock.unlock();
}因?yàn)閞emove方法使用兩個(gè)鎖全部上鎖,所以其他操作都需要等待它完成,而該方法需要從head節(jié)點(diǎn)遍歷到尾節(jié)點(diǎn),所以時(shí)間復(fù)雜度為O(n)。我們來看看unlink方法。
void unlink(Node<E> p, Node<E> trail) {
// p的元素置為null
p.item = null;
// p的前一個(gè)節(jié)點(diǎn)的next指向p的next,也就是把p從鏈表中去除了
trail.next = p.next;
// 如果last指向p,刪除p后讓last指向trail
if (last == p)
last = trail;
// 如果刪除之前元素是滿的,刪除之后就有空間了,喚醒生產(chǎn)線程放入元素
if (count.getAndDecrement() == capacity)
notFull.signal();
}問題
看源碼的時(shí)候,我給自己拋出了一個(gè)問題。
- 為什么dequeue里的h.next不指向null,而指向h?
- 為什么unlink里沒有p.next = null或者p.next = p這樣的操作?這個(gè)疑問一直困擾著我,直到我看了迭代器的部分源碼后才豁然開朗,下面放出部分迭代器的源碼:
private Node<E> current;
private Node<E> lastRet;
private E currentElement;
Itr() {
fullyLock();
try {
current = head.next;
if (current != null)
currentElement = current.item;
} finally {
fullyUnlock();
}
}
private Node<E> nextNode(Node<E> p) {
for (;;) {
// 解決了問題1
Node<E> s = p.next;
if (s == p)
return head.next;
if (s == null || s.item != null)
return s;
p = s;
}
}迭代器的遍歷分為兩步,第一步加雙鎖把元素放入臨時(shí)變量中,第二部遍歷臨時(shí)變量的元素。也就是說remove可能和迭代元素同時(shí)進(jìn)行,很有可能remove的時(shí)候,有線程在進(jìn)行迭代操作,而如果unlink中改變了p的next,很有可能在迭代的時(shí)候會(huì)造成錯(cuò)誤,造成不一致問題。這個(gè)解決了問題2。
而問題1其實(shí)在nextNode方法中也能找到,為了正確遍歷,nextNode使用了 s == p的判斷,當(dāng)下一個(gè)元素是自己本身時(shí),返回head的下一個(gè)節(jié)點(diǎn)。
四、總結(jié)
LinkedBlockingQueue是一個(gè)阻塞隊(duì)列,內(nèi)部由兩個(gè)ReentrantLock來實(shí)現(xiàn)出入隊(duì)列的線程安全,由各自的Condition對(duì)象的await和signal來實(shí)現(xiàn)等待和喚醒功能。它和ArrayBlockingQueue的不同點(diǎn)在于:
- 隊(duì)列大小有所不同,ArrayBlockingQueue是有界的初始化必須指定大小,而LinkedBlockingQueue可以是有界的也可以是無界的(Integer.MAX_VALUE),對(duì)于后者而言,當(dāng)添加速度大于移除速度時(shí),在無界的情況下,可能會(huì)造成內(nèi)存溢出等問題。
- 數(shù)據(jù)存儲(chǔ)容器不同,ArrayBlockingQueue采用的是數(shù)組作為數(shù)據(jù)存儲(chǔ)容器,而LinkedBlockingQueue采用的則是以Node節(jié)點(diǎn)作為連接對(duì)象的鏈表。
- 由于ArrayBlockingQueue采用的是數(shù)組的存儲(chǔ)容器,因此在插入或刪除元素時(shí)不會(huì)產(chǎn)生或銷毀任何額外的對(duì)象實(shí)例,而LinkedBlockingQueue則會(huì)生成一個(gè)額外的Node對(duì)象。這可能在長時(shí)間內(nèi)需要高效并發(fā)地處理大批量數(shù)據(jù)的時(shí),對(duì)于GC可能存在較大影響。
- 兩者的實(shí)現(xiàn)隊(duì)列添加或移除的鎖不一樣,ArrayBlockingQueue實(shí)現(xiàn)的隊(duì)列中的鎖是沒有分離的,即添加操作和移除操作采用的同一個(gè)ReenterLock鎖,而LinkedBlockingQueue實(shí)現(xiàn)的隊(duì)列中的鎖是分離的,其添加采用的是putLock,移除采用的則是takeLock,這樣能大大提高隊(duì)列的吞吐量,也意味著在高并發(fā)的情況下生產(chǎn)者和消費(fèi)者可以并行地操作隊(duì)列中的數(shù)據(jù),以此來提高整個(gè)隊(duì)列的并發(fā)性能。
到此這篇關(guān)于Java線程隊(duì)列LinkedBlockingQueue的使用的文章就介紹到這了,更多相關(guān)Java LinkedBlockingQueue內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
java中用float時(shí),數(shù)字后面加f,這樣是為什么你知道嗎
這篇文章主要介紹了java用float時(shí),數(shù)字后面加f,這樣是為什么你知道嗎?具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2021-09-09
SpringBoot優(yōu)化連接數(shù)的方法詳解
SpringBoot開發(fā)最大的好處是簡(jiǎn)化配置,內(nèi)置了Tomcat,下面這篇文章主要給大家介紹了關(guān)于SpringBoot優(yōu)化連接數(shù)的相關(guān)資料,文中通過實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下2023-06-06
Java如何讀寫Properties配置文件(Properties類)
這篇文章主要介紹了Java如何讀寫Properties配置文件(Properties類),具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-05-05
一問詳解SpringBoot配置文件優(yōu)先級(jí)
在SpringBoot項(xiàng)目當(dāng)中,我們要想配置一個(gè)屬性,可以通過這三種方式當(dāng)中的任意一種來配置都可以,那么優(yōu)先級(jí)怎么算,本文主要介紹了一問詳解SpringBoot配置文件優(yōu)先級(jí),需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-04-04
maven項(xiàng)目在實(shí)踐中的構(gòu)建管理之路的方法
這篇文章主要介紹了maven項(xiàng)目在實(shí)踐中的構(gòu)建管理之路的方法,小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過來看看吧2019-05-05
IDEA創(chuàng)建Servlet程序的兩種實(shí)現(xiàn)方法
Servlet是JavaWeb應(yīng)用程序中不可或缺的組件之一,本文主要介紹了IDEA創(chuàng)建Servlet程序的兩種實(shí)現(xiàn)方法,具有一定的參考價(jià)值,感興趣的可以了解一下2023-10-10
Spring AOP實(shí)現(xiàn)復(fù)雜的日志記錄操作(自定義注解)
Spring AOP實(shí)現(xiàn)復(fù)雜的日志記錄操作(自定義注解),具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2021-09-09

