Java阻塞隊列中的異類，SynchronousQueue底層實現原理剖析

来源:https://www.cnblogs.com/yidengjiagou/archive/2022/11/23/16919232.html
-Advertisement-
Play Games

上篇文章談到BlockingQueue的使用場景，並重點分析了ArrayBlockingQueue的實現原理，瞭解到ArrayBlockingQueue底層是基於數組實現的阻塞隊列。 但是BlockingQueue的實現類中，有一種阻塞隊列比較特殊，就是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. 創建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() 不支持 不支持

這幾組方法的不同之處就是：

  1. 當隊列滿了，再往隊列中放數據，add方法拋異常，offer方法返回false，put方法會一直阻塞（直到有其他線程從隊列中取走數據），offer(e, time, unit)方法阻塞指定時間然後返回false。
  2. 當隊列是空，再從隊列中取數據，remove方法拋異常，poll方法返回null，take方法會一直阻塞（直到有其他線程往隊列中放數據），poll(time, unit)方法阻塞指定時間然後返回null。
  3. 當隊列是空，再去隊列中查看數據（並不刪除數據），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 棧底層結構

棧結構，是非公平的，遵循先進後出。

image

使用個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為例：

  1. Thread0先往SynchronousQueue隊列中放入元素0
  2. Thread1再往SynchronousQueue隊列放入元素1
  3. Thread2從SynchronousQueue隊列中取出一個元素

第一步：Thread0先往SynchronousQueue隊列中放入元素0

把本次操作組裝成SNode壓入棧頂，item是元素0，waiter是當前線程Thread0，mode是1表示放入數據。

image

第二步：Thread1再往SynchronousQueue隊列放入元素1

把本次操作組裝成SNode壓入棧頂，item是元素1，waiter是當前線程Thread1，mode是1表示放入數據，next是SNode0。

image

第三步：Thread2從SynchronousQueue隊列中取出一個元素

這次的操作比較複雜，也是先把本次的操作包裝成SNode壓入棧頂。

item是null（取數據的時候，這個欄位沒有值），waiter是null（當前線程Thread2正在操作，所以不用賦值了），mode是2表示正在操作（即將跟後繼節點進行匹配），next是SNode1。

image

然後，Thread2開始把棧頂的兩個節點進行匹配，匹配成功後，就把SNode2賦值給SNode1的match屬性，喚醒SNode1中的Thread1線程，然後彈出SNode2節點和SNode1節點。

image

image

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. 總結

  1. SynchronousQueue是一種特殊的阻塞隊列，隊列長度是0，一個線程往隊列放數據，必須等待另一個線程取走數據。同樣，一個線程從隊列中取數據，必須等待另一個線程往隊列中放數據。
  2. SynchronousQueue底層是基於棧和隊列兩種數據結構實現的。
  3. Java線程池中的newCachedThreadPool（帶緩存的線程池）底層就是使用SynchronousQueue實現的。
  4. 如果希望你的任務需要被快速處理，可以使用SynchronousQueue隊列。

我是「一燈架構」，如果本文對你有幫助，歡迎各位小伙伴點贊、評論和關註，感謝各位老鐵，我們下期見

image


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • python中的列表和元組
    # 1.列表的格式 # [數據1,數據2,數據3,···] # 列表 可變數據類型 # 列表可以存儲多個數據，數據之間的逗號以英文逗號分隔 # 列表可以存儲不同類型數據，但一般存儲同一數據類型，這樣便於操作 # list_name = [] # 定義了一個空的列表 # 定義了一個有數據的列表 # 可 ...
  • Java 中九種 Map 的遍歷方式，你一般用的是哪種呢？
    日常工作中 Map 絕對是我們 Java 程式員高頻使用的一種數據結構，那 Map 都有哪些遍歷方式呢？這篇文章阿粉就帶大家看一下，看看你經常使用的是哪一種。 通過 entrySet 來遍歷 1、通過 for 和 map.entrySet() 來遍歷 第一種方式是採用 for 和 Map.Entry ...
  • easylogging++的那些事(二)巨集定義
    由於博主有很多個python環境，如msys64的python，anaconda3的python和官網下載的python, 當我在vscode運行python，需要安裝對應的包時，用pip安裝，如下 安裝成功了，但是還是沒有找到 原因非常簡單，就是我vscode使用的python環境不是上面那個py ...
  • <七>深入理解new和delete的原理
    new ,delete 運算符 int *p =new int; delete p; 看一下彙編代碼 可以看到new 和delete 運算符其實也是 operator運算符重載函數的調用 malloc和new malloc 按位元組開闢記憶體 new在開闢記憶體的時候需要指定類型 new int[10] ...
  • Java學習四
    一.小結 1.迴圈語句有三類：while迴圈，do-while迴圈和for迴圈 2.迴圈中需要重覆執行的語句所構成的整體稱為迴圈體 3.迴圈體執行一次稱為迴圈的一次迭代 4.無限迴圈是指迴圈語句被無限次執行 5.在設計迴圈時，既需要考慮迴圈控制構體，還需要考慮迴圈體 6.while迴圈首先檢查迴圈繼 ...
  • day22-web開發會話技術04
    WEB開發會話技術04 14.Session生命周期 14.1生命周期說明 public void setMaxInactiveInterval(int interval)：設置session的超時時間(以秒為單位)，超過指定的時長，session就會被銷毀。 值為正數的時候，設置session的超 ...
  • 《Go 語言併發之道》讀書筆記（五）
    遞歸與Stream流轉換 今天寫一個很久以前一直不太會的，今天花了大量的時間進行研究處理，現將代碼解析於此 list轉為類中一個屬性為key,類實例為value的Map Map<String, List<OrgTreeVo>> orgMap = orgList.stream().filter(h - ...
  • Windows OpenGL 圖像褐色
    目錄 一.OpenGL 圖像褐色 1.原始圖片 2.效果演示 二.OpenGL 圖像褐色源碼下載 三.猜你喜歡 零基礎 OpenGL ES 學習路線推薦 : OpenGL ES 學習目錄 >> OpenGL ES 基礎 零基礎 OpenGL ES 學習路線推薦 : OpenGL ES 學習目錄 >> ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類