死磕 java集合之LinkedTransferQueue源碼分析
問題
（1）LinkedTransferQueue是什麼東東？java
（2）LinkedTransferQueue是怎麼實現阻塞隊列的？node
（3）LinkedTransferQueue是怎麼控制併發安全的？git
（4）LinkedTransferQueue與SynchronousQueue有什麼異同？安全
簡介
LinkedTransferQueue是LinkedBlockingQueue、SynchronousQueue（公平模式）、ConcurrentLinkedQueue三者的集合體，它綜合了這三者的方法，而且提供了更加高效的實現方式。數據結構
繼承體系
LinkedTransferQueue實現了TransferQueue接口，而TransferQueue接口是繼承自BlockingQueue的，因此LinkedTransferQueue也是一個阻塞隊列。多線程
TransferQueue接口中定義瞭如下幾個方法：併發
// 嘗試移交元素
boolean tryTransfer(E e);
// 移交元素
void transfer(E e) throws InterruptedException;
// 嘗試移交元素（有超時時間）
boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException;
// 判斷是否有消費者
boolean hasWaitingConsumer();
// 查看消費者的數量
int getWaitingConsumerCount();
複製代碼
主要是定義了三個移交元素的方法，有阻塞的，有不阻塞的，有超時的。app
存儲結構
LinkedTransferQueue使用了一個叫作dual data structure的數據結構，或者叫作dual queue，譯爲雙重數據結構或者雙重隊列。less
雙重隊列是什麼意思呢？dom
放取元素使用同一個隊列，隊列中的節點具備兩種模式，一種是數據節點，一種是非數據節點。
放元素時先跟隊列頭節點對比，若是頭節點是非數據節點，就讓他們匹配，若是頭節點是數據節點，就生成一個數據節點放在隊列尾端（入隊）。
取元素時也是先跟隊列頭節點對比，若是頭節點是數據節點，就讓他們匹配，若是頭節點是非數據節點，就生成一個非數據節點放在隊列尾端（入隊）。
用圖形來表示就是下面這樣：
不論是放元素仍是取元素，都先跟頭節點對比，若是兩者模式不同就匹配它們，若是兩者模式同樣，就入隊。
源碼分析
主要屬性
// 頭節點
transient volatile Node head;
// 尾節點
private transient volatile Node tail;
// 放取元素的幾種方式：
// 當即返回，用於非超時的poll()和tryTransfer()方法中
private static final int NOW = 0; // for untimed poll, tryTransfer
// 異步，不會阻塞，用於放元素時，由於內部使用無界單鏈表存儲元素，不會阻塞放元素的過程
private static final int ASYNC = 1; // for offer, put, add
// 同步，調用的時候若是沒有匹配到會阻塞直到匹配到爲止
private static final int SYNC = 2; // for transfer, take
// 超時，用於有超時的poll()和tryTransfer()方法中
private static final int TIMED = 3; // for timed poll, tryTransfer
複製代碼
主要內部類
static final class Node {
// 是不是數據節點（也就標識了是生產者仍是消費者）
final boolean isData; // false if this is a request node
// 元素的值
volatile Object item; // initially non-null if isData; CASed to match
// 下一個節點
volatile Node next;
// 持有元素的線程
volatile Thread waiter; // null until waiting
}
複製代碼
典型的單鏈表結構，內部除了存儲元素的值和下一個節點的指針外，還包含了是否爲數據節點和持有元素的線程。
內部經過isData區分是生產者仍是消費者。
主要構造方法
public LinkedTransferQueue() {
}
public LinkedTransferQueue(Collection<? extends E> c) {
this();
addAll(c);
}
複製代碼
只有這兩個構造方法，且沒有初始容量，因此是無界的一個阻塞隊列。
入隊
四個方法都是同樣的，使用異步的方式調用xfer()方法，傳入的參數都如出一轍。
public void put(E e) {
// 異步模式，不會阻塞，不會超時
// 由於是放元素，單鏈表存儲，會一直日後加
xfer(e, true, ASYNC, 0);
}
public boolean offer(E e, long timeout, TimeUnit unit) {
xfer(e, true, ASYNC, 0);
return true;
}
public boolean offer(E e) {
xfer(e, true, ASYNC, 0);
return true;
}
public boolean add(E e) {
xfer(e, true, ASYNC, 0);
return true;
}
複製代碼
xfer(E e, boolean haveData, int how, long nanos)的參數分別是：
（1）e表示元素；
（2）haveData表示是不是數據節點，
（3）how表示放取元素的方式，上面提到的四種，NOW、ASYNC、SYNC、TIMED；
（4）nanos表示超時時間；
出隊
出隊的四個方法也是直接或間接的調用xfer()方法，放取元素的方式和超時規則略微不一樣，本質沒有大的區別。
public E remove() {
E x = poll();
if (x != null)
return x;
else
throw new NoSuchElementException();
}
public E take() throws InterruptedException {
// 同步模式，會阻塞直到取到元素
E e = xfer(null, false, SYNC, 0);
if (e != null)
return e;
Thread.interrupted();
throw new InterruptedException();
}
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
// 有超時時間
E e = xfer(null, false, TIMED, unit.toNanos(timeout));
if (e != null || !Thread.interrupted())
return e;
throw new InterruptedException();
}
public E poll() {
// 當即返回，沒取到元素返回null
return xfer(null, false, NOW, 0);
}
複製代碼
取元素就各有各的玩法了，有同步的，有超時的，有當即返回的。
移交元素的方法
public boolean tryTransfer(E e) {
// 當即返回
return xfer(e, true, NOW, 0) == null;
}
public void transfer(E e) throws InterruptedException {
// 同步模式
if (xfer(e, true, SYNC, 0) != null) {
Thread.interrupted(); // failure possible only due to interrupt
throw new InterruptedException();
}
}
public boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException {
// 有超時時間
if (xfer(e, true, TIMED, unit.toNanos(timeout)) == null)
return true;
if (!Thread.interrupted())
return false;
throw new InterruptedException();
}
複製代碼
請注意第二個參數，都是true，也就是這三個方法其實也是放元素的方法。
這裏xfer()方法的幾種模式到底有什麼區別呢？請看下面的分析。
神奇的xfer()方法
private E xfer(E e, boolean haveData, int how, long nanos) {
// 不容許放入空元素
if (haveData && (e == null))
throw new NullPointerException();
Node s = null; // the node to append, if needed
// 外層循環，自旋，失敗就重試
retry:
for (;;) { // restart on append race
// 下面這個for循環用於控制匹配的過程
// 同一時刻隊列中只會存儲一種類型的節點
// 從頭節點開始嘗試匹配，若是頭節點被其它線程先一步匹配了
// 就再嘗試其下一個，直到匹配到爲止，或者到隊列中沒有元素爲止
for (Node h = head, p = h; p != null;) { // find & match first node
// p節點的模式
boolean isData = p.isData;
// p節點的值
Object item = p.item;
// p沒有被匹配到
if (item != p && (item != null) == isData) { // unmatched
// 若是二者模式同樣，則不能匹配，跳出循環後嘗試入隊
if (isData == haveData) // can't match
break;
// 若是二者模式不同，則嘗試匹配
// 把p的值設置爲e（若是是取元素則e是null，若是是放元素則e是元素值）
if (p.casItem(item, e)) { // match
// 匹配成功
// for裏面的邏輯比較複雜，用於控制多線程同時放取元素時出現競爭的狀況的
// 看不懂能夠直接跳過
for (Node q = p; q != h;) {
// 進入到這裏多是頭節點已經被匹配，而後p會變成h的下一個節點
Node n = q.next; // update by 2 unless singleton
// 若是head還沒變，就把它更新成新的節點
// 並把它刪除（forgetNext()會把它的next設爲本身，也就是從單鏈表中刪除了）
// 這時爲何要把head設爲n呢？由於到這裏了，確定head自己已經被匹配掉了
// 而上面的p.casItem()又成功了，說明p也被當前這個元素給匹配掉了
// 因此須要把它們倆都出隊列，讓其它線程能夠從真正的頭開始，不用重複檢查了
if (head == h && casHead(h, n == null ? q : n)) {
h.forgetNext();
break;
} // advance and retry
// 若是新的頭節點爲空，或者其next爲空，或者其next未匹配，就重試
if ((h = head) == null ||
(q = h.next) == null || !q.isMatched())
break; // unless slack < 2
}
// 喚醒p中等待的線程
LockSupport.unpark(p.waiter);
// 並返回匹配到的元素
return LinkedTransferQueue.<E>cast(item);
}
}
// p已經被匹配了或者嘗試匹配的時候失敗了
// 也就是其它線程先一步匹配了p
// 這時候又分兩種狀況，p的next還沒來得及修改，p的next指向了本身
// 若是p的next已經指向了本身，就從新取head重試，不然就取其next重試
Node n = p.next;
p = (p != n) ? n : (h = head); // Use head if p offlist
}
// 到這裏確定是隊列中存儲的節點類型和本身同樣
// 或者隊列中沒有元素了
// 就入隊（無論放元素仍是取元素都得入隊）
// 入隊又分紅四種狀況：
// NOW，當即返回，沒有匹配到當即返回，不作入隊操做
// ASYNC，異步，元素入隊但當前線程不會阻塞（至關於無界LinkedBlockingQueue的元素入隊）
// SYNC，同步，元素入隊後當前線程阻塞，等待被匹配到
// TIMED，有超時，元素入隊後等待一段時間被匹配，時間到了還沒匹配到就返回元素自己
// 若是不是當即返回
if (how != NOW) { // No matches available
// 新建s節點
if (s == null)
s = new Node(e, haveData);
// 嘗試入隊
Node pred = tryAppend(s, haveData);
// 入隊失敗，重試
if (pred == null)
continue retry; // lost race vs opposite mode
// 若是不是異步（同步或者有超時）
// 就等待被匹配
if (how != ASYNC)
return awaitMatch(s, pred, e, (how == TIMED), nanos);
}
return e; // not waiting
}
}
private Node tryAppend(Node s, boolean haveData) {
// 從tail開始遍歷，把s放到鏈表尾端
for (Node t = tail, p = t;;) { // move p to last node and append
Node n, u; // temps for reads of next & tail
// 若是首尾都是null，說明鏈表中尚未元素
if (p == null && (p = head) == null) {
// 就讓首節點指向s
// 注意，這裏插入第一個元素的時候tail指針並無指向s
if (casHead(null, s))
return s; // initialize
}
else if (p.cannotPrecede(haveData))
// 若是p沒法處理，則返回null
// 這裏沒法處理的意思是，p和s節點的類型不同，不容許s入隊
// 好比，其它線程先入隊了一個數據節點，這時候要入隊一個非數據節點，就不容許，
// 隊列中全部的元素都要保證是同一種類型的節點
// 返回null後外面的方法會從新嘗試匹配從新入隊等
return null; // lost race vs opposite mode
else if ((n = p.next) != null) // not last; keep traversing
// 若是p的next不爲空，說明不是最後一個節點
// 則讓p從新指向最後一個節點
p = p != t && t != (u = tail) ? (t = u) : // stale tail
(p != n) ? n : null; // restart if off list
else if (!p.casNext(null, s))
// 若是CAS更新s爲p的next失敗
// 則說明有其它線程先一步更新到p的next了
// 就讓p指向p的next，從新嘗試讓s入隊
p = p.next; // re-read on CAS failure
else {
// 到這裏說明s成功入隊了
// 若是p不等於t，就更新tail指針
// 還記得上面插入第一個元素時tail指針並無指向新元素嗎？
// 這裏就是用來更新tail指針的
if (p != t) { // update if slack now >= 2
while ((tail != t || !casTail(t, s)) &&
(t = tail) != null &&
(s = t.next) != null && // advance and retry
(s = s.next) != null && s != t);
}
// 返回p，即s的前一個元素
return p;
}
}
}
private E awaitMatch(Node s, Node pred, E e, boolean timed, long nanos) {
// 若是是有超時的，計算其超時時間
final long deadline = timed ? System.nanoTime() + nanos : 0L;
// 當前線程
Thread w = Thread.currentThread();
// 自旋次數
int spins = -1; // initialized after first item and cancel checks
// 隨機數，隨機讓一些自旋的線程讓出CPU
ThreadLocalRandom randomYields = null; // bound if needed
for (;;) {
Object item = s.item;
// 若是s元素的值不等於e，說明它被匹配到了
if (item != e) { // matched
// assert item != s;
// 把s的item更新爲s自己
// 並把s中的waiter置爲空
s.forgetContents(); // avoid garbage
// 返回匹配到的元素
return LinkedTransferQueue.<E>cast(item);
}
// 若是當前線程中斷了，或者有超時的到期了
// 就更新s的元素值指向s自己
if ((w.isInterrupted() || (timed && nanos <= 0)) &&
s.casItem(e, s)) { // cancel
// 嘗試解除s與其前一個節點的關係
// 也就是刪除s節點
unsplice(pred, s);
// 返回元素的值自己，說明沒匹配到
return e;
}
// 若是自旋次數小於0，就計算自旋次數
if (spins < 0) { // establish spins at/near front
// spinsFor()計算自旋次數
// 若是前面有節點未被匹配就返回0
// 若是前面有節點且正在匹配中就返回必定的次數，等待
if ((spins = spinsFor(pred, s.isData)) > 0)
// 初始化隨機數
randomYields = ThreadLocalRandom.current();
}
else if (spins > 0) { // spin
// 還有自旋次數就減1
--spins;
// 並隨機讓出CPU
if (randomYields.nextInt(CHAINED_SPINS) == 0)
Thread.yield(); // occasionally yield
}
else if (s.waiter == null) {
// 更新s的waiter爲當前線程
s.waiter = w; // request unpark then recheck
}
else if (timed) {
// 若是有超時，計算超時時間，並阻塞必定時間
nanos = deadline - System.nanoTime();
if (nanos > 0L)
LockSupport.parkNanos(this, nanos);
}
else {
// 不是超時的，直接阻塞，等待被喚醒
// 喚醒後進入下一次循環，走第一個if的邏輯就返回匹配的元素了
LockSupport.park(this);
}
}
}
複製代碼
這三個方法裏的內容特別複雜，很大一部分代碼都是在控制線程安全，各類CAS，咱們這裏簡單描述一下大體的邏輯：
（1）來了一個元素，咱們先查看隊列頭的節點，是否與這個元素的模式同樣；
（2）若是模式不同，就嘗試讓他們匹配，若是頭節點被別的線程先匹配走了，就嘗試與頭節點的下一個節點匹配，如此一直日後，直到匹配到或到鏈表尾爲止；
（3）若是模式同樣，或者到鏈表尾了，就嘗試入隊；
（4）入隊的時候有可能鏈表尾修改了，那就尾指針後移，再從新嘗試入隊，依此往復；
（5）入隊成功了，就自旋或阻塞，阻塞了就等待被其它線程匹配到並喚醒；
（6）喚醒以後進入下一次循環就匹配到元素了，返回匹配到的元素；
（7）是否須要入隊及阻塞有四種狀況：
a）NOW，當即返回，沒有匹配到當即返回，不作入隊操做
對應的方法有：poll()、tryTransfer(e)
b）ASYNC，異步，元素入隊但當前線程不會阻塞（至關於無界LinkedBlockingQueue的元素入隊）
對應的方法有：add(e)、offer(e)、put(e)、offer(e, timeout, unit)
c）SYNC，同步，元素入隊後當前線程阻塞，等待被匹配到
對應的方法有：take()、transfer(e)
d）TIMED，有超時，元素入隊後等待一段時間被匹配，時間到了還沒匹配到就返回元素自己
對應的方法有：poll(timeout, unit)、tryTransfer(e, timeout, unit)
複製代碼
總結
（1）LinkedTransferQueue能夠看做LinkedBlockingQueue、SynchronousQueue（公平模式）、ConcurrentLinkedQueue三者的集合體；
（2）LinkedTransferQueue的實現方式是使用一種叫作雙重隊列的數據結構；
（3）不論是取元素仍是放元素都會入隊；
（4）先嚐試跟頭節點比較，若是兩者模式不同，就匹配它們，組成CP，而後返回對方的值；
（5）若是兩者模式同樣，就入隊，並自旋或阻塞等待被喚醒；
（6）至因而否入隊及阻塞有四種模式，NOW、ASYNC、SYNC、TIMED；
（7）LinkedTransferQueue全程都沒有使用synchronized、重入鎖等比較重的鎖，基本是經過 自旋+CAS 實現；
（8）對於入隊以後，先自旋必定次數後再調用LockSupport.park()或LockSupport.parkNanos阻塞；
彩蛋
LinkedTransferQueue與SynchronousQueue（公平模式）有什麼異同呢？
（1）在java8中二者的實現方式基本一致，都是使用的雙重隊列；
（2）前者徹底實現了後者，但比後者更靈活；
（3）後者無論放元素仍是取元素，若是沒有可匹配的元素，所在的線程都會阻塞；
（4）前者能夠本身控制放元素是否須要阻塞線程，好比使用四個添加元素的方法就不會阻塞線程，只入隊元素，使用transfer()會阻塞線程；
（5）取元素二者基本同樣，都會阻塞等待有新的元素進入被匹配到；
歡迎關注個人公衆號「彤哥讀源碼」，查看更多源碼系列文章, 與彤哥一塊兒暢遊源碼的海洋。
相關文章
- 1. 死磕 java集合之LinkedTransferQueue源碼分析
- 2. 【死磕 Java 集合】— ConcurrentSkipListMap源碼分析
- 3. 死磕 java集合之ArrayBlockingQueue源碼分析
- 4. 死磕 java集合之WeakHashMap源碼分析
- 5. 死磕 java集合之ArrayDeque源碼分析
- 6. 死磕 java集合之CopyOnWriteArrayList源碼分析
- 7. 死磕 java集合之DelayQueue源碼分析
- 8. 死磕 java集合之ArrayList源碼分析
- 9. 死磕 java集合之LinkedList源碼分析
- 10. 死磕 java集合之TreeMap源碼分析（四）-內含彩蛋
- 更多相關文章...
- • Scala Set(集合) - Scala教程
- • C# 集合（Collection） - C#教程
- • 互聯網組織的未來：剖析GitHub員工的任性之源
- • Java Agent入門實戰（二）-Instrumentation源碼概述
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 死磕 java集合之LinkedTransferQueue源碼分析
- 2. 【死磕 Java 集合】— ConcurrentSkipListMap源碼分析
- 3. 死磕 java集合之ArrayBlockingQueue源碼分析
- 4. 死磕 java集合之WeakHashMap源碼分析
- 5. 死磕 java集合之ArrayDeque源碼分析
- 6. 死磕 java集合之CopyOnWriteArrayList源碼分析
- 7. 死磕 java集合之DelayQueue源碼分析
- 8. 死磕 java集合之ArrayList源碼分析
- 9. 死磕 java集合之LinkedList源碼分析
- 10. 死磕 java集合之TreeMap源碼分析（四）-內含彩蛋