您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 綜合教學

Java 多線程核心組件深度解析：Exchanger 交換器原理、實戰與數據分片應用模式

編輯：Java 綜合教學

前言

Exchanger<V> 是 JDK 1.5 引入的同步工具類，提供成對線程間數據原子性交換的同步點。它可視為雙向的 SynchronousQueue，適用於生產者-消費者雙緩衝、遺傳算法種群交換、管道流數據交接等場景。本文從源碼原理、實戰範例、應用模式三維度深度剖析。

一、核心原理：CAS + Park/Unpark 實現雙向同步

關鍵結構 (JDK 8/11/17 核心邏輯一致)

public class Exchanger<V> {
    // 參與者節點，佇列中等待配對
    @SuppressWarnings("serial")
    static final class Node extends AbstractQueuedSynchronizer {
        // ... 省略 AQS 繼承細節
    }
    
    // 線程局部緩存：複用節點對象，減少 GC
    private final ThreadLocal<Node> localNode = ThreadLocal.withInitial(Node::new);
    
    // 交換入口
    public V exchange(V x) throws InterruptedException {
        return doExchange(x, null, 0); // 無超時
    }
    
    public V exchange(V x, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
        return doExchange(x, unit.toNanos(timeout), 0);
    }
}

交換流程簡化版

線程 A 調用 exchange(dataA)，創建/獲取 Node，嘗試 CAS 佔據 slot (單槽位)。
若槽位為空：A 佔據並 LockSupport.park() 掛起，等待夥伴。
若槽位已被 B 佔據：A 取出 B 的數據 dataB，將自己的 dataA 存入，LockSupport.unpark(B) 喚醒 B，返回 dataB。
B 被喚醒後，從節點獲取 dataA，返回完成交換。
超時/中斷處理：清理槽位，拋出異常。

關鍵特性

成對性： 必須兩兩配對，單數線程會永久阻塞/超時。
原子性： 數據交換在 CAS 保護下瞬間完成，無中間狀態。
可重入性： 同一線程連續調用會創建新節點，不死鎖。
無公平性保證： 先到先配對，非 FIFO。

二、實戰範例：雙線程數據交換

import java.util.concurrent.*;
import java.util.Random;

public class ExchangerDemo {
    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        Exchanger<String> exchanger = new Exchanger<>();
        
        // 線程 A：生產 "love"
        pool.submit(() -> {
            String send = "love";
            System.out.println(Thread.currentThread().getName() + " 發送: " + send);
            sleepRandom();
            try {
                String received = exchanger.exchange(send);
                System.out.println(Thread.currentThread().getName() + " 接收: " + received);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        // 線程 B：生產 "hate"
        pool.submit(() -> {
            String send = "hate";
            System.out.println(Thread.currentThread().getName() + " 發送: " + send);
            sleepRandom();
            try {
                String received = exchanger.exchange(send);
                System.out.println(Thread.currentThread().getName() + " 接收: " + received);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        pool.shutdown();
    }
    
    private static void sleepRandom() {
        try { Thread.sleep(new Random().nextInt(1000)); } 
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }
}

/* 典型輸出：
pool-1-thread-1 發送: love
pool-1-thread-2 發送: hate
pool-1-thread-1 接收: hate
pool-1-thread-2 接收: love
*/

三、進階應用模式：雙緩衝數據分片交換

場景：高吞吐日誌收集、網絡封包處理、流式計算。生產者填充 Buffer A，消費者處理 Buffer B，滿/空時通過 Exchanger 原子交換引用，零拷貝、無鎖競爭。

public class DualBufferProcessor<T> {
    private final Exchanger<Buffer<T>> exchanger = new Exchanger<>();
    private final Buffer<T> bufferA = new Buffer<>(1024);
    private final Buffer<T> bufferB = new Buffer<>(1024);
    
    // 生產者線程
    public void startProducer(Producer<T> producer) {
        new Thread(() -> {
            Buffer<T> current = bufferA;
            while (!Thread.interrupted()) {
                current.clear();
                producer.fill(current); // 阻塞式填充
                if (current.isFull()) {
                    try {
                        current = exchanger.exchange(current); // 交換獲得空 Buffer
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }, "Producer").start();
    }
    
    // 消費者線程
    public void startConsumer(Consumer<T> consumer) {
        new Thread(() -> {
            Buffer<T> current = bufferB;
            while (!Thread.interrupted()) {
                try {
                    current = exchanger.exchange(current); // 交換獲得滿 Buffer
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
                consumer.process(current); // 處理數據
                current.clear(); // 清空準備下輪
            }
        }, "Consumer").start();
    }
    
    // 簡易緩衝區
    static class Buffer<T> {
        final T[] items;
        int size = 0;
        @SuppressWarnings("unchecked")
        Buffer(int cap) { items = (T[]) new Object[cap]; }
        boolean isFull() { return size == items.length; }
        void clear() { size = 0; }
        void add(T t) { items[size++] = t; }
        T get(int i) { return items[i]; }
        int size() { return size; }
    }
    
    @FunctionalInterface
    interface Producer<T> { void fill(Buffer<T> buf); }
    @FunctionalInterface
    interface Consumer<T> { void process(Buffer<T> buf); }
}

四、對比：Exchanger vs 其他同步工具

工具	核心語義	適用場景	線程數	數據流向
Exchanger	雙向交換	雙緩衡、配對計算、基因交叉	成對 (2n)	雙向
SynchronousQueue	單向移交	直接遞交、線程池拒絕策略	多生產多消費	單向
CyclicBarrier	屏障聚合	多線程階段同步、MapReduce	N (固定)	無數據交換
Phaser	分階段屏障	動態註冊、多階段流水線	動態	無數據交換
CompletableFuture	異步編排	任務鏈、組合、異常處理	任意	單向結果傳遞

五、避坑指南與最佳實踐

確保成對調用： 奇數線程會導致最後一個永久阻塞，生產環境必須加超時 exchange(x, 5, TimeUnit.SECONDS)。
交換對象不可變/線程安全： 交換後雙方持有同一對象引用，若需隔離請交換副本或不可變對象。
避免在交換點做耗時操作： 交換應極快完成，業務處理移至交換前後。
大對象交換考慮堆外內存： 結合 ByteBuffer.allocateDirect 或 Netty ByteBuf 避免 GC 壓力。
診斷工具： jstack 查看線程狀態為 WAITING (parking) 於 Exchanger.doExchange 即為等待配對中。