Java 高併發核心組件源碼解析：SynchronousQueue 無鎖傳輸隊列與線程互助機制

SynchronousQueue 定位：容量為 0 的「傳遞通道」

不同于 ArrayBlockingQueue 等有緩衝隊列，SynchronousQueue **不存儲元素**。put(e) 必須等待 take()，take() 必須等待 put(e)，雙方像接力賽交棒——**數據直接從生產者線程傳遞給消費者線程**。這特性使其成為線程池 ThreadPoolExecutor 默認工作隊列（配合 SynchronousQueue 實現「來任務創建線程，無任務回收線程」）。

核心抽象：Transferer 轉移器

內部通過 Transferer 接口統一 put/take 語義：

interface Transferer {
    Object transfer(Object e, boolean timed, long nanos);
    // e != null → 生產者傳遞數據
    // e == null → 消費者請求數據
    // 返回非 null → 成功（消費者返回數據，生產者返回 e）
    // 返回 null → 超時/中斷失敗
}

JDK 7+ 默認使用 非公平模式 TransferQueue（基於 CAS 的無鎖鏈表），JDK 6 為公平模式 TransferStack（鎖+棧）。本文聚焦非公平模式。

等待隊列節點：SNode

static final class SNode {
    volatile SNode next;        // 隊列下一節點
    volatile SNode match;       // 匹配節點（配對的生產者/消費者）
    volatile Thread waiter;     // 等待線程
    Object item;                // 數據（生產者放入，消費者取出）
    int mode;                   // 模式：REQUEST(消費者) / DATA(生產者) / FULFILLING(正在匹配)
    // ... CAS 字段更新方法
}

transfer() 三階段狀態機

頭節點 head 指向隊列首部。每次調用 transfer 進入大循環，根據頭節點狀態分三種情況：

階段 1：入隊等待（頭為空 或 模式相同）

SNode h = head;
if (h == null || h.mode == mode) { // 隊列空、或同類操作（如連續 take）
    if (timed && nanos <= 0) return null; // 非阻塞模式直接失敗
    SNode s = new SNode(item, mode);      // 創建當前操作節點
    if (casHead(h, s)) {                  // CAS 入隊頭
        SNode m = awaitFulfill(s, timed, nanos); // 自旋/阻塞等待匹配
        if (m == s) { clean(s); return null; } // 被取消
        // 幫助匹配節點出隊（友情幫助）
        if ((h = head) != null && h.next == s) casHead(h, s.next);
        return (mode == REQUEST) ? m.item : s.item;
    }
}

• awaitFulfill：自旋嘗試匹配 → 失敗則 LockSupport.park 掛起 → 被 unpark 喚醒 → 檢查 match 字段是否被設置。
• 線程被喚醒後，**主動幫助完成頭節點出隊**（CAS 更新 head），減少其他線程競爭。

階段 2：發起匹配（頭節點模式互補）

} else if (!isFulfilling(h.mode)) { // 頭節點非匹配中
    if (h.isCancelled()) { casHead(h, h.next); continue; } // 清理取消節點
    // 將自己標記為 FULFILLING，嘗試匹配頭節點
    if (casHead(h, s = new SNode(item, FULFILLING | mode))) {
        for (;;) {
            SNode m = s.next; // 待匹配節點（原頭節點）
            if (m == null) { casHead(s, null); s = null; break; } // 隊列空了
            SNode mn = m.next;
            if (m.tryMatch(s)) { // CAS 設置 m.match = s，喚醒 m.waiter
                casHead(s, mn); // 雙雙出隊
                return (mode == REQUEST) ? m.item : s.item;
            } else {
                s.casNext(m, mn); // 匹配失敗（被其他線程搶走），幫助刪除 m
            }
        }
    }
}

• 當前線程發現頭節點是「互補操作」（如自己 put、頭是 take），**主動升級為履行者 (FULFILLING)**，嘗試完成匹配。
• tryMatch 通過 CAS 設置 m.match = s，成功則 LockSupport.unpark(m.waiter) 喚醒對方。
• 若匹配失敗（別的線程已完成），**幫助刪除已完成節點** s.casNext(m, mn)，繼續循環嘗試下一個等待者。

階段 3：協助完成（頭節點已是 FULFILLING）

} else { // 發現頭節點正在履行（FULFILLING）
    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); // 幫助刪除失敗節點
    }
    // 自己不返回，重新進入大循環處理自己的操作
}

• 線程發現頭節點已被其他線程標記為 FULFILLING，**選擇協助完成**而非競爭。
• 協助成功後，**不處理自己的請求**，重新進入循環（continue 隱含），以新狀態重新評估。

「線程互助」設計哲學：從競爭到協作

這種「幫助鄰居」機制源自 **Michael & Scott 無鎖隊列算法**，同樣應用於 ConcurrentLinkedQueue、LinkedTransferQueue。核心思想：**將「完成匹配」作為共享任務，任何線程均可貢獻 CAS 操作推進全局狀態**，消除單點瓶頸。

實戰啟示

1. 線程池選型：new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<>()) 實現「無界線程池，按需創建」，適合短時異步任務。
2. 避免誤用：SynchronousQueue 不適合「生產快、消費慢」場景（會大量創建線程導致 OOM），需配合 ThreadPoolExecutor.DiscardPolicy 或有界隊列。
3. 源碼閱讀技巧：關注 mode 狀態流轉、match 雙向綁定、casHead/casNext 原子更新鏈表結構、LockSupport.park/unpark 停喚機制。