Java中SynchronousQueue的底層實現原理剖析
上篇文章談到BlockingQueue的使用場景，并重點分析了ArrayBlockingQueue的實現原理，了解到ArrayBlockingQueue底層是基于數組實現的阻塞隊列。
但是BlockingQueue的實現類中，有一種阻塞隊列比較特殊，就是SynchronousQueue（同步移交隊列），隊列長度為0。
作用就是一個線程往隊列放數據的時候，必須等待另一個線程從隊列中取走數據。同樣，從隊列中取數據的時候，必須等待另一個線程往隊列中放數據。
這樣特殊的隊列，有什么應用場景呢？
1. SynchronousQueue用法
先看一個SynchronousQueue的簡單用例：
/**
* @author 一燈架構
* @apiNote SynchronousQueue示例
**/
public class SynchronousQueueDemo {
public static void main(String[] args) throws InterruptedException {
// 1. 創(chuàng)建SynchronousQueue隊列
BlockingQueue<Integer> synchronousQueue = new SynchronousQueue<>();
// 2. 啟動一個線程，往隊列中放3個元素
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " 入隊列 1");
synchronousQueue.put(1);
Thread.sleep(1);
System.out.println(Thread.currentThread().getName() + " 入隊列 2");
synchronousQueue.put(2);
Thread.sleep(1);
System.out.println(Thread.currentThread().getName() + " 入隊列 3");
synchronousQueue.put(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
// 3. 等待1000毫秒
Thread.sleep(1000L);
// 4. 再啟動一個線程，從隊列中取出3個元素
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " 出隊列 " + synchronousQueue.take());
Thread.sleep(1);
System.out.println(Thread.currentThread().getName() + " 出隊列 " + synchronousQueue.take());
Thread.sleep(1);
System.out.println(Thread.currentThread().getName() + " 出隊列 " + synchronousQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}輸出結果：
Thread-0 入隊列 1
Thread-1 出隊列 1
Thread-0 入隊列 2
Thread-1 出隊列 2
Thread-0 入隊列 3
Thread-1 出隊列 3
從輸出結果中可以看到，第一個線程Thread-0往隊列放入一個元素1后，就被阻塞了。直到第二個線程Thread-1從隊列中取走元素1后，Thread-0才能繼續(xù)放入第二個元素2。
由于SynchronousQueue是BlockingQueue的實現類，所以也實現類BlockingQueue中幾組抽象方法：
為了滿足不同的使用場景，BlockingQueue設計了很多的放數據和取數據的方法。
| 操作 | 拋出異常 | 返回特定值 | 阻塞 | 阻塞一段時間 |
|---|---|---|---|---|
| 放數據 | add | offer | put | offer(e, time, unit) |
| 取數據 | remove | poll | take | poll(time, unit) |
| 查看數據（不刪除） | element() | peek() | 不支持 | 不支持 |
這幾組方法的不同之處就是：
- 當隊列滿了，再往隊列中放數據，add方法拋異常，offer方法返回false，put方法會一直阻塞（直到有其他線程從隊列中取走數據），offer(e, time, unit)方法阻塞指定時間然后返回false。
- 當隊列是空，再從隊列中取數據，remove方法拋異常，poll方法返回null，take方法會一直阻塞（直到有其他線程往隊列中放數據），poll(time, unit)方法阻塞指定時間然后返回null。
- 當隊列是空，再去隊列中查看數據（并不刪除數據），element方法拋異常，peek方法返回null。
工作中使用最多的就是offer、poll阻塞指定時間的方法。
2. SynchronousQueue應用場景
SynchronousQueue的特點：
隊列長度是0，一個線程往隊列放數據，必須等待另一個線程取走數據。同樣，一個線程從隊列中取數據，必須等待另一個線程往隊列中放數據。
這種特殊的實現邏輯有什么應用場景呢？
我的理解就是，如果你希望你的任務需要被快速處理，就可以使用這種隊列。
Java線程池中的newCachedThreadPool（帶緩存的線程池）底層就是使用SynchronousQueue實現的。
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}newCachedThreadPool線程池的核心線程數是0，最大線程數是Integer的最大值，線程存活時間是60秒。
如果你使用newCachedThreadPool線程池，你提交的任務會被更快速的處理，因為你每次提交任務，都會有一個空閑的線程等著處理任務。如果沒有空閑的線程，也會立即創(chuàng)建一個線程處理你的任務。
你想想，這處理效率，杠杠滴！
當然也有弊端，如果你提交了太多的任務，導致創(chuàng)建了大量的線程，這些線程都在競爭CPU時間片，等待CPU調度，處理任務速度也會變慢，所以在使用過程中也要綜合考慮。
3. SynchronousQueue源碼解析
3.1 SynchronousQueue類屬性
public class SynchronousQueue<E> extends AbstractQueue<E> implements BlockingQueue<E> {
// 轉換器，取數據和放數據的核心邏輯都在這個類里面
private transient volatile Transferer<E> transferer;
// 默認的構造方法（使用非公平隊列）
public SynchronousQueue() {
this(false);
}
// 有參構造方法，可以指定是否使用公平隊列
public SynchronousQueue(boolean fair) {
transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}
// 轉換器實現類
abstract static class Transferer<E> {
abstract E transfer(E e, boolean timed, long nanos);
}
// 基于棧實現的非公平隊列
static final class TransferStack<E> extends Transferer<E> {
}
// 基于隊列實現的公平隊列
static final class TransferQueue<E> extends Transferer<E> {
}
}可以看到SynchronousQueue默認的無參構造方法，內部使用的是基于棧實現的非公平隊列，當然也可以調用有參構造方法，傳參是true，使用基于隊列實現的公平隊列。
// 使用非公平隊列（基于棧實現） BlockingQueue<Integer> synchronousQueue = new SynchronousQueue<>(); // 使用公平隊列（基于隊列實現） BlockingQueue<Integer> synchronousQueue = new SynchronousQueue<>(true);
本次就常用的棧實現來剖析SynchronousQueue的底層實現原理。
3.2 棧底層結構
棧結構，是非公平的，遵循先進后出。
使用個case測試一下：
/**
* @author 一燈架構
* @apiNote SynchronousQueue示例
**/
public class SynchronousQueueDemo {
public static void main(String[] args) throws InterruptedException {
// 1. 創(chuàng)建SynchronousQueue隊列
SynchronousQueue<Integer> synchronousQueue = new SynchronousQueue<>();
// 2. 啟動一個線程，往隊列中放1個元素
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " 入隊列 0");
synchronousQueue.put(0);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
// 3. 等待1000毫秒
Thread.sleep(1000L);
// 4. 啟動一個線程，往隊列中放1個元素
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " 入隊列 1");
synchronousQueue.put(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
// 5. 等待1000毫秒
Thread.sleep(1000L);
// 6. 再啟動一個線程，從隊列中取出1個元素
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " 出隊列 " + synchronousQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
// 7. 等待1000毫秒
Thread.sleep(1000L);
// 8. 再啟動一個線程，從隊列中取出1個元素
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " 出隊列 " + synchronousQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}輸出結果：
Thread-0 入隊列 0
Thread-1 入隊列 1
Thread-2 出隊列 1
Thread-3 出隊列 0
從輸出結果中可以看出，符合棧結構先進后出的順序。
3.3 棧節(jié)點源碼
棧中的數據都是由一個個的節(jié)點組成的，先看一下節(jié)點類的源碼：
// 節(jié)點
static final class SNode {
// 節(jié)點值（取數據的時候，該字段為null）
Object item;
// 存取數據的線程
volatile Thread waiter;
// 節(jié)點模式
int mode;
// 匹配到的節(jié)點
volatile SNode match;
// 后繼節(jié)點
volatile SNode next;
}item
節(jié)點值，只在存數據的時候用。取數據的時候，這個值是null。
waiter
存取數據的線程，如果沒有對應的接收線程，這個線程會被阻塞。
mode
節(jié)點模式，共有3種類型：
| 類型值 | 類型描述 | 類型的作用 |
|---|---|---|
| 0 | REQUEST | 表示取數據 |
| 1 | DATA | 表示存數據 |
| 2 | FULFILLING | 表示正在等待執(zhí)行（比如取數據的線程，等待其他線程放數據） |
3.4 put/take流程
放數據和取數據的邏輯，在底層復用的是同一個方法，以put/take方法為例，另外兩個放數據的方法，add和offer方法底層實現是一樣的。
先看一下數據流轉的過程，方便理解源碼。
還是以上面的case為例：
- Thread0先往SynchronousQueue隊列中放入元素0
- Thread1再往SynchronousQueue隊列放入元素1
- Thread2從SynchronousQueue隊列中取出一個元素
第一步：Thread0先往SynchronousQueue隊列中放入元素0
把本次操作組裝成SNode壓入棧頂，item是元素0，waiter是當前線程Thread0，mode是1表示放入數據。
第二步：Thread1再往SynchronousQueue隊列放入元素1
把本次操作組裝成SNode壓入棧頂，item是元素1，waiter是當前線程Thread1，mode是1表示放入數據，next是SNode0。
第三步：Thread2從SynchronousQueue隊列中取出一個元素
這次的操作比較復雜，也是先把本次的操作包裝成SNode壓入棧頂。
item是null（取數據的時候，這個字段沒有值），waiter是null（當前線程Thread2正在操作，所以不用賦值了），mode是2表示正在操作（即將跟后繼節(jié)點進行匹配），next是SNode1。
然后，Thread2開始把棧頂的兩個節(jié)點進行匹配，匹配成功后，就把SNode2賦值給SNode1的match屬性，喚醒SNode1中的Thread1線程，然后彈出SNode2節(jié)點和SNode1節(jié)點。
3.5 put/take源碼實現
先看一下put方法源碼：
// 放數據
public void put(E e) throws InterruptedException {
// 不允許放null元素
if (e == null)
throw new NullPointerException();
// 調用轉換器實現類，放元素
if (transferer.transfer(e, false, 0) == null) {
// 如果放數據失敗，就中斷當前線程，并拋出異常
Thread.interrupted();
throw new InterruptedException();
}
}核心邏輯都在transfer方法中，代碼很長，理清邏輯后，也很容易理解。
// 取數據和放數據操作，共用一個方法
E transfer(E e, boolean timed, long nanos) {
SNode s = null;
// e為空，說明是取數據，否則是放數據
int mode = (e == null) ? REQUEST : DATA;
for (; ; ) {
SNode h = head;
// 1. 如果棧頂節(jié)點為空，或者棧頂節(jié)點類型跟本次操作相同（都是取數據，或者都是放數據）
if (h == null || h.mode == mode) {
// 2. 判斷節(jié)點是否已經超時
if (timed && nanos <= 0) {
// 3. 如果棧頂節(jié)點已經被取消，就刪除棧頂節(jié)點
if (h != null && h.isCancelled())
casHead(h, h.next);
else
return null;
// 4. 把本次操作包裝成SNode，壓入棧頂
} else if (casHead(h, s = snode(s, e, h, mode))) {
// 5. 掛起當前線程，等待被喚醒
SNode m = awaitFulfill(s, timed, nanos);
// 6. 如果這個節(jié)點已經被取消，就刪除這個節(jié)點
if (m == s) {
clean(s);
return null;
}
// 7. 把s.next設置成head
if ((h = head) != null && h.next == s)
casHead(h, s.next);
return (E) ((mode == REQUEST) ? m.item : s.item);
}
// 8. 如果棧頂節(jié)點類型跟本次操作不同，并且不是FULFILLING類型
} else if (!isFulfilling(h.mode)) {
// 9. 再次判斷如果棧頂節(jié)點已經被取消，就刪除棧頂節(jié)點
if (h.isCancelled())
casHead(h, h.next);
// 10. 把本次操作包裝成SNode（類型是FULFILLING），壓入棧頂
else if (casHead(h, s = snode(s, e, h, FULFILLING | mode))) {
// 11. 使用死循環(huán)，直到匹配到對應的節(jié)點
for (; ; ) {
// 12. 遍歷下個節(jié)點
SNode m = s.next;
// 13. 如果節(jié)點是null，表示遍歷到末尾，設置棧頂節(jié)點是null，結束。
if (m == null) {
casHead(s, null);
s = null;
break;
}
SNode mn = m.next;
// 14. 如果棧頂的后繼節(jié)點跟棧頂節(jié)點匹配成功，就刪除這兩個節(jié)點，結束。
if (m.tryMatch(s)) {
casHead(s, mn);
return (E) ((mode == REQUEST) ? m.item : s.item);
} else
// 15. 如果沒有匹配成功，就刪除棧頂的后繼節(jié)點，繼續(xù)匹配
s.casNext(m, mn);
}
}
} else {
// 16. 如果棧頂節(jié)點類型跟本次操作不同，并且是FULFILLING類型，
// 就再執(zhí)行一遍上面第11步for循環(huán)中的邏輯（很少概率出現）
SNode m = h.next;
if (m == null)
casHead(h, null);
else {
SNode mn = m.next;
if (m.tryMatch(h))
casHead(h, mn);
else
h.casNext(m, mn);
}
}
}
}transfer方法邏輯也很簡單，就是判斷本次操作類型是否跟棧頂節(jié)點相同，如果相同，就把本次操作壓入棧頂。否則就跟棧頂節(jié)點匹配，喚醒棧頂節(jié)點線程，彈出棧頂節(jié)點。
transfer方法中調用了awaitFulfill方法，作用是掛起當前線程。
// 等待被喚醒
SNode awaitFulfill(SNode s, boolean timed, long nanos) {
// 1. 計算超時時間
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Thread w = Thread.currentThread();
// 2. 計算自旋次數
int spins = (shouldSpin(s) ?
(timed ? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) {
if (w.isInterrupted())
s.tryCancel();
// 3. 如果已經匹配到其他節(jié)點，直接返回
SNode m = s.match;
if (m != null)
return m;
if (timed) {
// 4. 超時時間遞減
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
s.tryCancel();
continue;
}
}
// 5. 自旋次數減一
if (spins > 0)
spins = shouldSpin(s) ? (spins-1) : 0;
else if (s.waiter == null)
s.waiter = w;
// 6. 開始掛起當前線程
else if (!timed)
LockSupport.park(this);
else if (nanos > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanos);
}
}awaitFulfill方法的邏輯也很簡單，就是掛起當前線程。
take方法底層使用的也是transfer方法：
// 取數據
public E take() throws InterruptedException {
// // 調用轉換器實現類，取數據
E e = transferer.transfer(null, false, 0);
if (e != null)
return e;
// 沒取到，就中斷當前線程
Thread.interrupted();
throw new InterruptedException();
}4. 總結
- SynchronousQueue是一種特殊的阻塞隊列，隊列長度是0，一個線程往隊列放數據，必須等待另一個線程取走數據。同樣，一個線程從隊列中取數據，必須等待另一個線程往隊列中放數據。
- SynchronousQueue底層是基于棧和隊列兩種數據結構實現的。
- Java線程池中的newCachedThreadPool（帶緩存的線程池）底層就是使用SynchronousQueue實現的。
- 如果希望你的任務需要被快速處理，可以使用SynchronousQueue隊列。
到此這篇關于Java中SynchronousQueue的底層實現原理剖析的文章就介紹到這了,更多相關Java SynchronousQueue內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
這篇文章主要介紹了springboot亂碼問題解決方案,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-02-02
這篇文章主要介紹了spring boot 利用注解實現權限驗證的實現代碼，小編覺得挺不錯的，現在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-11-11
這篇文章主要介紹了Java因項目配置不當而引發(fā)的數據泄露解決辦法，本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-09-09
下面小編就為大家分享一篇基于Java 生產者消費者模式(詳細分析)，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-01-01
這篇文章主要介紹了淺談java隨機數的陷阱，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-09-09
Spring Security中successHandler和failureHandler使用方式
這篇文章主要介紹了Spring Security中successHandler和failureHandler使用方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-08-08
這篇文章主要介紹了Java 如何利用緩沖流讀寫文件的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-07-07
這篇文章主要介紹了Java通過數據庫表生成實體類，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習吧2023-02-02
這篇文章主要為大家詳細介紹了java單鏈表使用總結，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-01-01
日常開發(fā)過程中經常會使用json進行數據的傳輸,這就涉及到了對象和json的相互轉化,常用的解決方案有：Jackson（推薦）、谷歌的Gson、阿里的Fastjson,這篇文章主要給大家介紹了關于SpringBoot如何利用jackson格式化時間的相關資料,需要的朋友可以參考下2021-06-06