上篇文章談到BlockingQueue的使用場景，并重點分析了ArrayBlockingQueue的實作原理，了解到ArrayBlockingQueue底層是基于陣列實作的阻塞佇列，
但是BlockingQueue的實作類中，有一種阻塞佇列比較特殊，就是SynchronousQueue（同步移交佇列），佇列長度為0，
作用就是一個執行緒往佇列放資料的時候，必須等待另一個執行緒從佇列中取走資料，同樣，從佇列中取資料的時候，必須等待另一個執行緒往佇列中放資料，
這樣特殊的佇列，有什么應用場景呢？
1. SynchronousQueue用法
先看一個SynchronousQueue的簡單用例：
/**
* @author 一燈架構
* @apiNote SynchronousQueue示例
**/
public class SynchronousQueueDemo {
public static void main(String[] args) throws InterruptedException {
// 1. 創建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才能繼續放入第二個元素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執行緒池，你提交的任務會被更快速的處理，因為你每次提交任務，都會有一個空閑的執行緒等著處理任務，如果沒有空閑的執行緒，也會立即創建一個執行緒處理你的任務，
你想想，這處理效率，杠杠滴！
當然也有弊端，如果你提交了太多的任務，導致創建了大量的執行緒，這些執行緒都在競爭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. 創建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 堆疊節點原始碼
堆疊中的資料都是由一個個的節點組成的，先看一下節點類的原始碼：
// 節點
static final class SNode {
// 節點值（取資料的時候，該欄位為null）
Object item;
// 存取資料的執行緒
volatile Thread waiter;
// 節點模式
int mode;
// 匹配到的節點
volatile SNode match;
// 后繼節點
volatile SNode next;
}
-
item
節點值，只在存資料的時候用，取資料的時候，這個值是null，
-
waiter
存取資料的執行緒，如果沒有對應的接收執行緒，這個執行緒會被阻塞，
-
mode
節點模式，共有3種型別：
型別值 型別描述 型別的作用 0 REQUEST 表示取資料 1 DATA 表示存資料 2 FULFILLING 表示正在等待執行（比如取資料的執行緒，等待其他執行緒放資料）
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表示正在操作（即將跟后繼節點進行匹配），next是SNode1，
然后，Thread2開始把堆疊頂的兩個節點進行匹配，匹配成功后，就把SNode2賦值給SNode1的match屬性，喚醒SNode1中的Thread1執行緒，然后彈出SNode2節點和SNode1節點，
3.5 put/take原始碼實作
看完 了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. 如果堆疊頂節點為空，或者堆疊頂節點型別跟本次操作相同（都是取資料，或者都是放資料）
if (h == null || h.mode == mode) {
// 2. 判斷節點是否已經超時
if (timed && nanos <= 0) {
// 3. 如果堆疊頂節點已經被取消，就洗掉堆疊頂節點
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. 如果這個節點已經被取消，就洗掉這個節點
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. 如果堆疊頂節點型別跟本次操作不同，并且不是FULFILLING型別
} else if (!isFulfilling(h.mode)) {
// 9. 再次判斷如果堆疊頂節點已經被取消，就洗掉堆疊頂節點
if (h.isCancelled())
casHead(h, h.next);
// 10. 把本次操作包裝成SNode（型別是FULFILLING），壓入堆疊頂
else if (casHead(h, s = snode(s, e, h, FULFILLING | mode))) {
// 11. 使用死回圈，直到匹配到對應的節點
for (; ; ) {
// 12. 遍歷下個節點
SNode m = s.next;
// 13. 如果節點是null，表示遍歷到末尾，設定堆疊頂節點是null，結束，
if (m == null) {
casHead(s, null);
s = null;
break;
}
SNode mn = m.next;
// 14. 如果堆疊頂的后繼節點跟堆疊頂節點匹配成功，就洗掉這兩個節點，結束，
if (m.tryMatch(s)) {
casHead(s, mn);
return (E) ((mode == REQUEST) ? m.item : s.item);
} else
// 15. 如果沒有匹配成功，就洗掉堆疊頂的后繼節點，繼續匹配
s.casNext(m, mn);
}
}
} else {
// 16. 如果堆疊頂節點型別跟本次操作不同，并且是FULFILLING型別，
// 就再執行一遍上面第11步for回圈中的邏輯（很少概率出現）
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方法邏輯也很簡單，就是判斷本次操作型別是否跟堆疊頂節點相同，如果相同，就把本次操作壓入堆疊頂，否則就跟堆疊頂節點匹配，喚醒堆疊頂節點執行緒，彈出堆疊頂節點，
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. 如果已經匹配到其他節點，直接回傳
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佇列，
我是「一燈架構」，如果本文對你有幫助，歡迎各位小伙伴點贊、評論和關注，感謝各位老鐵，我們下期見
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/538148.html
標籤：其他