1. SynchronousQueue 功能簡介
SynchronousQueue 是 BlockingQueue 家族中的一個成員, 不同于其他的成員, 它具有以下特性:
1. 整個 queue 沒有容量, 表現(xiàn)為, 你每次進(jìn)行put值進(jìn)去時, 必須等待相應(yīng)的 consumer 拿走數(shù)據(jù)后才可以再次 put 數(shù)據(jù)
2. queue 對應(yīng) peek, contains, clear, isEmpty ... 等方法其實是無效的
3. 整個 queue 分為 公平(TransferQueue FIFO)與非公平模式(TransferStack LIFO 默認(rèn))
4. 若使用 TransferQueue, 則隊列中永遠(yuǎn)會存在一個 dummy node
2. SynchronousQueue 構(gòu)造函數(shù)
/**
* Creates a {@code SynchronousQueue} with nonfair access policy
*/
public SynchronousQueue() { this(false); }
/**
* Creates a {@code KSynchronousQueue} with the specified fairness policy
* @param fair
*/
public SynchronousQueue(boolean fair){
// 通過 fair 值來決定內(nèi)部用 使用 queue 還是 stack 存儲線程節(jié)點
transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}
我們可以看到默認(rèn)使用的 TransferStack 作為內(nèi)部節(jié)點容器, 我們可以通過 fair 來決定公平與否
3. 公平模式 TransferQueue
/**
* 這是一個非常典型的 queue , 它有如下的特點
* 1. 整個隊列有 head, tail 兩個節(jié)點
* 2. 隊列初始化時會有個 dummy 節(jié)點
* 3. 這個隊列的頭節(jié)點是個 dummy 節(jié)點/ 或 哨兵節(jié)點, 所以操作的總是隊列中的第二個節(jié)點(AQS的設(shè)計中也是這也)
*/
/** 頭節(jié)點 */
transient volatile QNode head;
/** 尾節(jié)點 */
transient volatile QNode tail;
/**
* Reference to a cancelled node that might not yet have been
* unlinked from queue because it was last inserted node
* when it was cancelled
*/
/**
* 對應(yīng) 中斷或超時的 前繼節(jié)點,這個節(jié)點存在的意義是標(biāo)記, 它的下個節(jié)點要刪除
* 何時使用:
* 當(dāng)你要刪除 節(jié)點 node, 若節(jié)點 node 是隊列的末尾, 則開始用這個節(jié)點,
* 為什么呢洪乍？
* 大家知道 刪除一個節(jié)點 直接 A.CASNext(B, B.next) 就可以,但是當(dāng) 節(jié)點 B 是整個隊列中的末尾元素時,
* 一個線程刪除節(jié)點B, 一個線程在節(jié)點B之后插入節(jié)點 這樣操作容易致使插入的節(jié)點丟失, 這個cleanMe很像
* ConcurrentSkipListMap 中的 刪除添加的 marker 節(jié)點, 他們都是起著相同的作用
*/
transient volatile QNode cleanMe;
TransferQueue(){
/**
* 構(gòu)造一個 dummy node, 而整個 queue 中永遠(yuǎn)會存在這樣一個 dummy node
* dummy node 的存在使得 代碼中不存在復(fù)雜的 if 條件判斷
*/
QNode h = new QNode(null, false);
head = h;
tail = h;
}
/**
* 推進(jìn) head 節(jié)點,將 老節(jié)點的 oldNode.next = this, help gc,
* 這種和 ConcurrentLinkedQueue 中一樣
*/
void advanceHead(QNode h, QNode nh){
if(h == head && unsafe.compareAndSwapObject(this, headOffset, h, nh)){
h.next = h; // forget old next help gc
}
}
/** 更新新的 tail 節(jié)點 */
void advanceTail(QNode t, QNode nt){
if(tail == t){
unsafe.compareAndSwapObject(this, tailOffset, t, nt);
}
}
/** CAS 設(shè)置 cleamMe 節(jié)點 */
boolean casCleanMe(QNode cmp, QNode val){
return cleanMe == cmp && unsafe.compareAndSwapObject(this, cleanMeOffset, cmp, val);
}
從代碼中我們知道, TransferQueue 是個 dual queue, 初始化時默認(rèn)會個一個 dummy node;
而最特別的是 cleanMeNode, cleanMeNode是一個標(biāo)記節(jié)點, cleanMeNode.next 節(jié)點是因中斷或超時需要刪除的節(jié)點绑谣，是在清除 隊列最尾端節(jié)點時, 不直接刪除這個節(jié)點, 而是間刪除節(jié)點的前繼節(jié)點標(biāo)示為 cleanMe 節(jié)點, 為下次刪除做準(zhǔn)備, 功能和 ConcurrentSkipListMap 中的 marker 節(jié)點差不多, 都是防止在同一地點插入節(jié)點的同時因刪除節(jié)點而造成節(jié)點的丟失, 不明白的可以看 ConcurrentSkipListMap.
3. 公平模式 TransferQueue transfer方法
這個方法的主邏輯:
1. 若隊列為空 / 隊列中的尾節(jié)點和自己的 類型相同, 則添加 node
到隊列中, 直到 timeout/interrupt/其他線程和這個線程匹配
timeout/interrupt awaitFulfill方法返回的是 node 本身
匹配成功的話, 要么返回 null (producer返回的), 或正真的傳遞值 (consumer 返回的)
2. 隊列不為空, 且隊列的 head.next 節(jié)點是當(dāng)前節(jié)點匹配的節(jié)點,
進(jìn)行數(shù)據(jù)的傳遞匹配, 并且通過 advanceHead 方法幫助 先前 block 的節(jié)點 dequeue
直接看代碼 transfer
/**
* Puts or takes an item
* 主方法
*
* @param e if non-null, the item to be handed to a consumer;
* if null, requests that transfer return an item
* offered by producer.
* @param timed if this operation should timeout
* @param nanos the timeout, in nanosecond
* @return
*/
@Override
E transfer(E e, boolean timed, long nanos) {
/**
* Basic algorithm is to loop trying to take either of
* two actions:
*
* 1. If queue apparently empty or holding same-mode nodes,
* try to add node to queue of waiters, wait to be
* fulfilled (or cancelled) and return matching item.
*
* 2. If queue apparently contains waiting items, and this
* call is of complementary mode, try to fulfill by CAS'ing
* item field of waiting node and dequeuing it, and then
* returning matching item.
*
* In each case, along the way, check for gurading against
* seeing uninitialized head or tail value. This never
* happens in current SynchronousQueue, but could if
* callers held non-volatile/final ref to the
* transferer. The check is here anyway because it places
* null checks at top of loop, which is usually faster
* than having them implicity interspersed
*
* 這個 producer / consumer 的主方法, 主要分為兩種情況
*
* 1. 若隊列為空 / 隊列中的尾節(jié)點和自己的 類型相同, 則添加 node
* 到隊列中, 直到 timeout/interrupt/其他線程和這個線程匹配
* timeout/interrupt awaitFulfill方法返回的是 node 本身
* 匹配成功的話, 要么返回 null (producer返回的), 或正真的傳遞值 (consumer 返回的)
*
* 2. 隊列不為空, 且隊列的 head.next 節(jié)點是當(dāng)前節(jié)點匹配的節(jié)點,
* 進(jìn)行數(shù)據(jù)的傳遞匹配, 并且通過 advanceHead 方法幫助 先前 block 的節(jié)點 dequeue
*/
QNode s = null; // constrcuted/reused as needed
boolean isData = (e != null); // 1.判斷 e != null 用于區(qū)分 producer 與 consumer
for(;;){
QNode t = tail;
QNode h = head;
if(t == null || h == null){ // 2. 數(shù)據(jù)未初始化, continue 重來
continue; // spin
}
if(h == t || t.isData == isData){ // 3. 隊列為空, 或隊列尾節(jié)點和自己相同 (注意這里是和尾節(jié)點比價, 下面進(jìn)行匹配時是和 head.next 進(jìn)行比較)
QNode tn = t.next;
if(t != tail){ // 4. tail 改變了, 重新再來
continue;
}
if(tn != null){ // 5. 其他線程添加了 tail.next, 所以幫助推進(jìn) tail
advanceTail(t, tn);
continue;
}
if(timed && nanos <= 0){ // 6. 調(diào)用的方法的 wait 類型的, 并且 超時了, 直接返回 null, 直接見 SynchronousQueue.poll() 方法,說明此 poll 的調(diào)用只有當(dāng)前隊列中正好有一個與之匹配的線程在等待被【匹配才有返回值
return null;
}
if(s == null){
s = new QNode(e, isData); // 7. 構(gòu)建節(jié)點 QNode
}
if(!t.casNext(null, s)){ // 8. 將 新建的節(jié)點加入到 隊列中
continue;
}
advanceTail(t, s); // 9. 幫助推進(jìn) tail 節(jié)點
Object x = awaitFulfill(s, e, timed, nanos); // 10. 調(diào)用awaitFulfill, 若節(jié)點是 head.next, 則進(jìn)行一些自旋, 若不是的話, 直接 block, 知道有其他線程 與之匹配, 或它自己進(jìn)行線程的中斷
if(x == s){ // 11. 若 (x == s)節(jié)點s 對應(yīng)額線程 wait 超時 或線程中斷, 不然的話 x == null (s 是 producer) 或 是正真的傳遞值(s 是 consumer)
clean(t, s); // 12. 對接點 s 進(jìn)行清除, 若 s 不是鏈表的最后一個節(jié)點, 則直接 CAS 進(jìn)行 節(jié)點的刪除, 若 s 是鏈表的最后一個節(jié)點, 則 要么清除以前的 cleamMe 節(jié)點(cleamMe != null), 然后將 s.prev 設(shè)置為 cleanMe 節(jié)點, 下次進(jìn)行刪除 或直接將 s.prev 設(shè)置為cleanMe
return null;
}
if(!s.isOffList()){ // 13. 節(jié)點 s 沒有 offlist
advanceHead(t, s); // 14. 推進(jìn)head 節(jié)點, 下次就調(diào)用 s.next 節(jié)點進(jìn)行匹配(這里調(diào)用的是 advanceHead, 因為代碼能執(zhí)行到這邊說明s已經(jīng)是 head.next 節(jié)點了)
if(x != null){ // and forget fields
s.item = s;
}
s.waiter = null; // 15. 釋放線程 ref
}
return (x != null) ? (E)x :e;
}else{ // 16. 進(jìn)行線程的匹配操作, 匹配操作是從 head.next 開始匹配 (注意 隊列剛開始構(gòu)建時 有個 dummy node, 而且 head 節(jié)點永遠(yuǎn)是個 dummy node 這個和 AQS 中一樣的)
QNode m = h.next; // 17. 獲取 head.next 準(zhǔn)備開始匹配
if(t != tail || m == null || h != head){
continue; // 18. 不一致讀取, 有其他線程改變了隊列的結(jié)構(gòu)inconsistent read
}
/** producer 和 consumer 匹配操作
* 1. 獲取 m的 item (注意這里的m是head的next節(jié)點
* 2. 判斷 isData 與x的模式是否匹配, 只有produce與consumer才能配成一對
* 3. x == m 判斷是否 節(jié)點m 是否已經(jīng)進(jìn)行取消了, 具體看(QNOde#tryCancel)
* 4. m.casItem 將producer與consumer的數(shù)據(jù)進(jìn)行交換 (這里存在并發(fā)時可能cas操作失敗的情況)
* 5. 若 cas操作成功則將h節(jié)點dequeue
*
* 疑惑: 為什么將h進(jìn)行 dequeue, 而不是 m節(jié)點
* 答案: 因為每次進(jìn)行配對時, 都是將 h 是個 dummy node, 正真的數(shù)據(jù)節(jié)點 是 head.next
*/
Object x = m.item;
if(isData == (x != null) || // 19. 兩者的模式是否匹配 (因為并發(fā)環(huán)境下 有可能其他的線程強(qiáng)走了匹配的節(jié)點)
x == m || // 20. m 節(jié)點 線程中斷或者 wait 超時了
!m.casItem(x, e) // 21. 進(jìn)行 CAS 操作 更改等待線程的 item 值(等待的有可能是 concumer / producer)
){
advanceHead(h, m); // 22.推進(jìn) head 節(jié)點 重試 (尤其 21 操作失敗)
continue;
}
advanceHead(h, m); // 23. producer consumer 交換數(shù)據(jù)成功, 推進(jìn) head 節(jié)點
LockSupport.unpark(m.waiter); // 24. 換線等待中的 m 節(jié)點, 而在 awaitFulfill 方法中 因為 item 改變了, 所以 x != e 成立, 返回
return (x != null) ? (E)x : e; // 25. 操作到這里若是 producer, 則 x != null, 返回 x, 若是consumer， 則 x == null,.返回 producer(其實就是 節(jié)點m) 的 e
}
}
}
OK, 我們梳理一下一般性的流程:
1. 一開始整個queue為空, 線程直接封裝成QNode, 通過 awaitFulfill 方法進(jìn)入自旋等待狀態(tài), 除非超時或線程中斷, 不然一直等待, 直到有線程與之匹配
2. 下個再來的線程若isData與尾節(jié)點一樣, 則進(jìn)行第一步, 不然進(jìn)行數(shù)據(jù)轉(zhuǎn)移(步驟 21), 然后 unpark 等待的線程
3. 等待的線程被喚醒, 從awaitFulfill方法返回, 最后將結(jié)果返回
4. 公平模式 TransferQueue awaitFulfill
/**
* Spins/blocks until node s is fulfilled
*
* 主邏輯: 若節(jié)點是 head.next 則進(jìn)行 spins 一會, 若不是, 則調(diào)用 LockSupport.park / parkNanos(), 直到其他的線程對其進(jìn)行喚醒
*
* @param s the waiting node
* @param e the comparsion value for checking match
* @param timed true if timed wait
* @param nanos timeout value
* @return matched item, or s of cancelled
*/
Object awaitFulfill(QNode s, E e, boolean timed, long nanos){
final long deadline = timed ? System.nanoTime() + nanos : 0L;// 1. 計算 deadline 時間 (只有 timed 為true 時才有用)
Thread w = Thread.currentThread(); // 2. 獲取當(dāng)前的線程
int spins = ((head.next == s) ? // 3. 若當(dāng)前節(jié)點是 head.next 時才進(jìn)行 spin, 不然的話不是浪費 CPU 嗎, 對挖
(timed ? maxTimeSpins : maxUntimedSpins) : 0);
for(;;){ // loop 直到 成功
if(w.isInterrupted()){ // 4. 若線程中斷, 直接將 item = this, 在 transfer 中會對返回值進(jìn)行判斷 (transfer中的 步驟 11)
s.tryCancel(e);
}
Object x = s.item;
if(x != e){ // 5. 在進(jìn)行線程阻塞->喚醒, 線程中斷, 等待超時, 這時 x != e,直接return 回去
return x;
}
if(timed){
nanos = deadline - System.nanoTime();
if(nanos <= 0L){ // 6. 等待超時, 改變 node 的item值, 進(jìn)行 continue, 下一步就到 awaitFulfill的第 5 步 -> return
s.tryCancel(e);
continue;
}
}
if(spins > 0){ // 7. spin 一次一次減少
--spins;
}
else if(s.waiter == null){
s.waiter = w;
}
else if(!timed){ // 8. 進(jìn)行沒有超時的 park
LockSupport.park(this);
}
else if(nanos > spinForTimeoutThreshold){ // 9. 自旋次數(shù)過了, 直接 + timeout 方式 park
LockSupport.parkNanos(this, nanos);
}
}
}
梳理邏輯:
1. 計算timeout時間(若 time = true)
2. 判斷 當(dāng)前節(jié)點是否是 head.next 節(jié)點(queue中有個dummy node 的存在, AQS 中也是這樣), 若是的話就進(jìn)行 spin 的賦值, 其他的節(jié)點沒有這個需要, 浪費資源
3. 接下來就是自旋, 超過次數(shù)就進(jìn)行阻塞, 直到有其他線程喚醒, 或線程中斷(這里線程中斷返回的是 Node 自己)
5. 公平模式 TransferQueue clean
/**
* Gets rid of cancelled node s with original predecessor pred.
* 對 中斷的 或 等待超時的 節(jié)點進(jìn)行清除操作
*/
void clean(QNode pred, QNode s) {
s.waiter = null; // forget thread // 1. 清除掉 thread 引用
/*
* At any given time, exactly one node on list cannot be
* deleted -- the last inserted node. To accommodate this,
* if we cannot delete s, we save its predecessor as
* "cleanMe", deleting the previously saved version
* first. At least one of node s or the node previously
* saved can always be deleted, so this always terminates.
*
* 在程序運(yùn)行中的任何時刻, 最后插入的節(jié)點不能被刪除(這里的刪除指 通過 cas 直接刪除, 因為這樣直接刪除會有多刪除其他節(jié)點的風(fēng)險)
* 當(dāng) 節(jié)點 s 是最后一個節(jié)點時, 將 s.pred 保存為 cleamMe 節(jié)點, 下次再進(jìn)行清除操作
*/
while (pred.next == s) { // Return early if already unlinked // 2. 判斷 pred.next == s, 下面的 步驟2 可能導(dǎo)致 pred.next = next
QNode h = head;
QNode hn = h.next; // Absorb cancelled first node as head
if (hn != null && hn.isCancelled()) { // 3. hn 中斷或者超時, 則推進(jìn) head 指針, 若這時 h 是 pred 則 loop 中的條件 "pred.next == s" 不滿足, 退出 loop
advanceHead(h, hn);
continue;
}
QNode t = tail; // Ensure consistent read for tail
if (t == h) // 4. 隊列為空, 說明其他的線程進(jìn)行操作, 刪除了 節(jié)點(注意這里永遠(yuǎn)會有個 dummy node)
return;
QNode tn = t.next;
if (t != tail) // 5. 其他的線程改變了 tail, continue 重新來
continue;
if (tn != null) {
advanceTail(t, tn); // 6. 幫助推進(jìn) tail
continue;
}
if (s != t) { // If not tail, try to unsplice // 7. 節(jié)點 s 不是尾節(jié)點, 則 直接 CAS 刪除節(jié)點(在隊列中間進(jìn)行這種刪除是沒有風(fēng)險的)
QNode sn = s.next;
if (sn == s || pred.casNext(s, sn))
return;
}
QNode dp = cleanMe; // 8. s 是隊列的尾節(jié)點, 則 cleanMe 出場
if (dp != null) { // Try unlinking previous cancelled node
QNode d = dp.next; // 9. cleanMe 不為 null, 進(jìn)行刪除刪一次的 s節(jié)點, 也就是這里的節(jié)點d
QNode dn;
if (d == null || // d is gone or // 10. 這里有幾個特殊情況 1. 原來的s節(jié)點()也就是這里的節(jié)點d已經(jīng)刪除; 2. 原來的節(jié)點 cleanMe 已經(jīng)通過 advanceHead 進(jìn)行刪除; 3 原來的節(jié)點 s已經(jīng)刪除 (所以 !d.siCancelled), 存在這三種情況, 直接將 cleanMe 清除
d == dp || // d is off list or
!d.isCancelled() || // d not cancelled or
(d != t && // d not tail and // 11. d 不是tail節(jié)點, 且dn沒有offlist, 直接通過 cas 刪除 上次的節(jié)點 s (也就是這里的節(jié)點d); 其實就是根據(jù) cleanMe 來清除隊列中間的節(jié)點
(dn = d.next) != null && // has successor
dn != d && // that is on list
dp.casNext(d, dn))) // d unspliced
casCleanMe(dp, null); // 12. 清除 cleanMe 節(jié)點, 這里的 dp == pred 若成立, 說明清除節(jié)點s搏予， 成功, 直接 return, 不然的話要再次 loop, 接著到 步驟 13, 設(shè)置這次的 cleanMe 然后再返回
if (dp == pred)
return; // s is already saved node
} else if (casCleanMe(null, pred)) // 原來的 cleanMe 是 null, 則將 pred 標(biāo)記為 cleamMe 為下次 清除 s 節(jié)點做標(biāo)識
return; // Postpone cleaning s
}
}
clean 方法是 整個代碼分析過程中的難點：
1. 難在并發(fā)的情況比較多
2. cleanMe 節(jié)點存在的意義
調(diào)用這個方法都是由 節(jié)點線程中斷或等待超時時調(diào)用的, 清除時分兩種情況討論:
1. 刪除的節(jié)點不是queue尾節(jié)點, 這時 直接 pred.casNext(s, s.next) 方式來進(jìn)行刪除(和ConcurrentLikedQueue中差不多)
2. 刪除的節(jié)點是隊尾節(jié)點
1) 此時 cleanMe == null, 則 前繼節(jié)點pred標(biāo)記為 cleanMe, 為下次刪除做準(zhǔn)備
2) 此時 cleanMe != null, 先刪除上次需要刪除的節(jié)點, 然后將 cleanMe至null, 讓后再將 pred 賦值給 cleanMe
這時我們想起了 ConcurrentSkipListMap 中的 marker 節(jié)點, 對, marker 和 cleanMe 都是起著防止并發(fā)環(huán)境中多刪除節(jié)點的功能
