并發編程之Exchanger原理與使用

作者：一角錢技術 2020-12-09 08:21:47

Exchanger是適用在兩個線程之間數據交換的并發工具類，它的作用是找到一個同步點，當兩個線程都執行到了同步點(exchange方法)之后(有一個沒有執行到就一直等待，也可以設置等待超時時間)，就將自身線程的數據與對方交換。

前言

在JUC包中，除了一些常用的或者說常見的并發工具類(ReentrantLock，CountDownLatch，CyclicBarrier，Semaphore)等，還有一個不常用的線程同步器類 —— Exchanger。

Exchanger 是什么?

它提供一個同步點，在這個同步點兩個線程可以交換彼此的數據。這個兩個線程通過exchange方法交換數據，如果第一個線程先執行exchange方法，它會一直等待第二個線程也執行exchange，當兩個線程都到達同步點時，這兩個線程就可以交換數據，將本線程生產出來的數據傳遞給對方。因此使用Exchanger的中斷時成對的線程使用exchange()方法，當有一對線程到達了同步點，就會進行交換數據，因此該工具類的線程對象是成對的。

線程可以在成對內配對和交換元素的同步點。每個線程在輸入exchange方法時提供一些對象，與合作者線程匹配，并在返回時接收其合作伙伴的對象。交換器可以被視為一個的雙向形式的SynchroniuzedQueue。交換器在諸如遺傳算法和管道設計的應用中可能是有用的。

一個用于兩個工作線程之間交換數據的封裝工具類，簡單說就是一個線程在完成一定事務后想與另一個線程交換數據，則第一個先拿出數據的線程會一直等待第二個線程，直到第二個線程拿著數據到來時才能彼此交換對應數據。

Exchanger 用法

Exchanger 泛型類型，其中V表示可交換的數據類型
V exchanger(V v)：等待另一個線程到達此交換點(除非當前線程被中斷)，然后將給定的對象傳送該線程，并接收該線程的對象。
V exchanger(V v, long timeout, TimeUnit unit)：等待另一個線程到達此交換點(除非當前線程被中斷或超出類指定的等待時間)，然后將給定的對象傳送給該線程，并接收該線程的對象。

應用場景

Exchanger可以用于遺傳算法，遺傳算法里需要選出兩個人作為交配對象，這時候會交換兩人的數據，并使用交叉規則得出2個交配結果。

Exchanger也可以用于校對工作。比如我們需要將紙制銀流通過人工的方式錄入成電子銀行流水，為了避免錯誤，采用AB崗兩人進行錄入，錄入到Excel之后，系統需要加載這兩個Excel，并對這兩個Excel數據進行校對，看看是否錄入的一致

Exchanger的典型應用場景是：一個任務在創建對象，而這些對象的生產代價很高，另一個任務在消費這些對象。通過這種方式，可以有更多的對象在被創建的同時被消費。

案例說明

Exchanger 用于兩個線程間交換數據，當然實際參與的線程可以不止兩個，測試用例如下：

private static void test1() throws InterruptedException { 
        Exchanger<String> exchanger = new Exchanger<>(); 
        CountDownLatch countDownLatch = new CountDownLatch(5); 
 
        for (int i = 0; i < 5; i++) { 
            new Thread(() ->  { 
 
                try { 
                    String origMsg = RandomStringUtils.randomNumeric(6); 
 
                    // 先到達的線程會在此等待，直到有一個線程跟它交換數據或者等待超時 
                    String exchangeMsg = exchanger.exchange(origMsg,5, TimeUnit.SECONDS); 
 
                    System.out.println(Thread.currentThread().getName() + "\t origMsg:" + origMsg + "\t exchangeMsg:" + exchangeMsg); 
                } catch (InterruptedException e) { 
                    e.printStackTrace(); 
                } catch (TimeoutException e) { 
                    e.printStackTrace(); 
                }finally { 
                    countDownLatch.countDown(); 
                } 
 
            },String.valueOf(i)).start(); 
        } 
 
        countDownLatch.await(); 
    }

第5個線程因為沒有匹配的線程而等待超時，輸出如下：

0  origMsg:524053  exchangeMsg:098544 
3  origMsg:433246  exchangeMsg:956604 
4  origMsg:098544  exchangeMsg:524053 
1  origMsg:956604  exchangeMsg:433246 
java.util.concurrent.TimeoutException 
 at java.util.concurrent.Exchanger.exchange(Exchanger.java:626) 
 at com.nuih.juc.ExchangerDemo.lambda$test1$0(ExchangerDemo.java:37) 
 at java.lang.Thread.run(Thread.java:748)

上述測試用例是比較簡單，可以模擬消息消費的場景來觀察Exchanger的行為，測試用例如下：

private static void test2() throws InterruptedException { 
        Exchanger<String> exchanger = new Exchanger<>(); 
        CountDownLatch countDownLatch = new CountDownLatch(4); 
        CyclicBarrier cyclicBarrier = new CyclicBarrier(4); 
 
        // 生產者 
        Runnable producer = new Runnable() { 
            @Override 
            public void run() { 
                try{ 
                    cyclicBarrier.await(); 
 
                    for (int i = 0; i < 5; i++) { 
                        String msg = RandomStringUtils.randomNumeric(6); 
                        exchanger.exchange(msg,5,TimeUnit.SECONDS); 
                        System.out.println(Thread.currentThread().getName() + "\t producer msg -> " + msg + " ,\t i -> " + i); 
                    } 
 
                }catch (Exception e){ 
                    e.printStackTrace(); 
                }finally { 
                    countDownLatch.countDown(); 
                } 
            } 
        }; 
 
        // 消費者 
        Runnable consumer = new Runnable() { 
            @Override 
            public void run() { 
                try{ 
                    cyclicBarrier.await(); 
                    for (int i = 0; i < 5; i++) { 
                        String msg = exchanger.exchange(null,5,TimeUnit.SECONDS); 
                        System.out.println(Thread.currentThread().getName() + "\t consumer msg -> " + msg + ",\t" + i); 
                    } 
 
                }catch (Exception e){ 
                    e.printStackTrace(); 
                }finally { 
                    countDownLatch.countDown(); 
                } 
            } 
        }; 
 
        for (int i = 0; i < 2; i++){ 
            new Thread(producer).start(); 
            new Thread(consumer).start(); 
        } 
 
        countDownLatch.await(); 
    }

輸出如下，上面生產者和消費者線程數是一樣的，循環次數也是一樣的，但是還是出現等待超時的情形：

Thread-3  consumer msg -> null, 0 
Thread-1  consumer msg -> null, 0 
Thread-1  consumer msg -> null, 1 
Thread-2  producer msg -> 640010 ,  i -> 0 
Thread-2  producer msg -> 733133 ,  i -> 1 
Thread-3  consumer msg -> null, 1 
Thread-3  consumer msg -> 476520, 2 
Thread-1  consumer msg -> 640010, 2 
Thread-1  consumer msg -> null, 3 
Thread-0  producer msg -> 993414 ,  i -> 0 
Thread-0  producer msg -> 292745 ,  i -> 1 
Thread-2  producer msg -> 476520 ,  i -> 2 
Thread-2  producer msg -> 408446 ,  i -> 3 
Thread-3  consumer msg -> null, 3 
Thread-1  consumer msg -> 292745, 4 
Thread-2  producer msg -> 251971 ,  i -> 4 
Thread-0  producer msg -> 078939 ,  i -> 2 
Thread-3  consumer msg -> 251971, 4 
java.util.concurrent.TimeoutException 
 at java.util.concurrent.Exchanger.exchange(Exchanger.java:626) 
 at com.nuih.juc.ExchangerDemo$1.run(ExchangerDemo.java:70) 
 at java.lang.Thread.run(Thread.java:748) 
 
Process finished with exit code 0

這種等待超時是概率出現的，這是為啥?

因為系統調度的不均衡和Exchanger底層的大量自旋等待導致這4個線程并不是調用exchanger成功的次數并不一致。另外從輸出可以看出，消費者線程并沒有像我們想的那樣跟生產者線程一一匹配，生產者線程有時也充當來消費者線程，這是為啥?因為Exchanger匹配時完全不關注這個線程的角色，兩個線程之間的匹配完全由調度決定的，即CPU同時執行來或者緊挨著執行來兩個線程，這兩個線程就匹配成功來。

源碼分析

Exchanger 類圖

其內部主要變量和方法如下：

成員屬性

// ThreadLocal變量，每個線程都有之間的一個副本 
private final Participant participant; 
// 高并發下使用的，保存待匹配的Node實例 
private volatile Node[] arena; 
// 低并發下，arena未初始化時使用的保存待匹配的Node實例 
private volatile Node slot; 
// 初始值為0，當創建arena后被負責SEQ，用來記錄arena數組的可用最大索引， 
// 會隨著并發的增大而增大直到等于最大值FULL， 
// 會隨著并行的線程逐一匹配成功而減少恢復成初始值 
private volatile int bound;

還有多個表示字段偏移量的靜態屬性，通過static代碼塊初始化，如下：

// Unsafe mechanics 
private static final sun.misc.Unsafe U; 
private static final long BOUND; 
private static final long SLOT; 
private static final long MATCH; 
private static final long BLOCKER; 
private static final int ABASE; 
static { 
    int s; 
    try { 
        U = sun.misc.Unsafe.getUnsafe(); 
        Class<?> ek = Exchanger.class; 
        Class<?> nk = Node.class; 
        Class<?> ak = Node[].class; 
        Class<?> tk = Thread.class; 
        BOUND = U.objectFieldOffset 
            (ek.getDeclaredField("bound")); 
        SLOT = U.objectFieldOffset 
            (ek.getDeclaredField("slot")); 
        MATCH = U.objectFieldOffset 
            (nk.getDeclaredField("match")); 
        BLOCKER = U.objectFieldOffset 
            (tk.getDeclaredField("parkBlocker")); 
        s = U.arrayIndexScale(ak); 
        // ABASE absorbs padding in front of element 0 
        ABASE = U.arrayBaseOffset(ak) + (1 << ASHIFT); 
 
    } catch (Exception e) { 
        throw new Error(e); 
    } 
    if ((s & (s-1)) != 0 || s > (1 << ASHIFT)) 
        throw new Error("Unsupported array scale"); 
}

Exchanger 定義來多個靜態變量，如下：

// 初始化arena時使用， 1 << ASHIFT 是一個緩存行的大小，避免來不同的Node落入到同一個高速緩存行 
// 這里實際是把數組容量擴大來8倍，原來索引相鄰的兩個元素，擴容后中間隔來7個元素，從元素的起始地址上看就隔來8個元素，中間的7個都是空的，為來避免原來相鄰的兩個元素都落入到同一個緩存行中 
// 因為arena是對象數組，一個元素占8字節，8個就是64字節 
private static final int ASHIFT = 7; 
// arena 數組元素的索引最大值即255 
private static final int MMASK = 0xff; 
// arena 數組的最大長度即256 
private static final int SEQ = MMASK + 1; 
// 獲取CPU核數 
private static final int NCPU = Runtime.getRuntime().availableProcessors(); 
// 實際的數組長度，因為是線程兩兩配對的，所以最大長度是核數除以2 
static final int FULL = (NCPU >= (MMASK << 1)) ? MMASK : NCPU >>> 1; 
// 自旋等待的次數 
private static final int SPINS = 1 << 10; 
// 如果交換的對象是null，則返回此對象 
private static final Object NULL_ITEM = new Object(); 
// 如果等待超時導致交換失敗，則返回此對象 
private static final Object TIMED_OUT = new Object();

內部類

Exchanger類中有兩個內部類，一個Node，一個Participant。 Participant繼承了ThreadLocal并且重寫了其initialValue方法，返回一個Node對象。其定義如下：

@sun.misc.Contended static final class Node { 
    int index;              // Arena index 
    int bound;              // Last recorded value of Exchanger.bound 
    int collides;           // Number of CAS failures at current bound 
    int hash;               // Pseudo-random for spins 
    Object item;            // This thread's current item 
    volatile Object match;  // Item provided by releasing thread 
    volatile Thread parked; // Set to this thread when parked, else null 
} 
 
/** The corresponding thread local class */ 
static final class Participant extends ThreadLocal<Node> { 
    public Node initialValue() { return new Node(); } 
}

其中Contended注解是為了避免高速緩存行導致的偽共享問題

index用來記錄arena數組的索引
bound用于記錄上一次的Exchanger bound屬性
collides用于記錄在bound不變的情況下CAS搶占失敗的次數
hash是自旋等待時計算隨機數使用的
item表示當前線程請求交換的對象
match是同其它線程交換的結果，match不為null表示交換成功
parked為跟該Node關聯的處于休眠狀態的線程。

重要方法

exchange()方法

@SuppressWarnings("unchecked") 
public V exchange(V x) throws InterruptedException { 
    Object v; 
    Object item = (x == null) ? NULL_ITEM : x; // translate null args 
    if ((arena != null || // 是null就執行后面的方法 
         (v = slotExchange(item, false, 0L)) == null) && 
        // 如果執行slotExchange有結果就執行后面的，否則返回 
        ((Thread.interrupted() || // 非中斷則執行后面的方法 
          (v = arenaExchange(item, false, 0L)) == null))) 
        throw new InterruptedException(); 
    return (v == NULL_ITEM) ? null : (V)v; 
}

exchange 方法的執行步驟：

如果執行 soltExchange 有結果就執行后面的 arenaExchange;
如果 slot 被占用，就執行 arenaExchange;
返回的數據 v 是對方線程的數據項;
總結即：如果A線程先調用，那么A的數據項存儲的 item中，則B線程的數據項存儲在 math 中;
當沒有多線程并發操作 Exchange 的時候，使用 slotExchange 就足夠了，slot 是一個 node 對象;
當出現并發了，一個 slot 就不夠了，就需要使用一個 node 數組 arena 操作了。

slotExchange()方法

slotExchange 是基于slot屬性來完成交換的，調用soltExchange方法時，如果slot屬性為null，當前線程會將slot屬性由null修改成當前線程的Node，如果修改失敗則下一次for循環走solt屬性不為null的邏輯，如果修改成功則自旋等待，自旋一定次數后通過Unsafe的park方法當當前線程休眠，可以指定休眠的時間，如果沒有指定則無限期休眠直到被喚醒;無論是因為線程中斷被喚醒，等待超時被喚醒還是其它線程unpark喚醒的，都會檢查當前線程的Node的屬性釋放為null，如果不為null說明交互成功，返回該對象;否則返回null或者TIME_OUT,在返回前會將item，match等屬性置為null，保存之前自旋時計算的hash值，方便下一次調用slotExchange。

調用slotExchange方法時，如果slot屬性不為null，則當前線程會嘗試將其修改null，如果cas修改成功，表示當前線程與slot屬性對應的線程匹配成功，會獲取slot屬性對應Node的item屬性，將當前線程交換的對象保存到slot屬性對應的Node的match屬性，然后喚醒獲取slot屬性對應Node的waiter屬性，即處理休眠狀態的線程，至此交換完成，同樣的在返回前需要將item，match等屬性置為null，保存之前自旋時計算的hash置，方便下一次調用slotExchange;如果cas修改slot屬性失敗，說明有其它線程也在搶占slot，則初始化arena屬性，下一次for循環因為arena屬性不為null，直接返回null，從而通過arenaExchange完成交換。

// arena 為null是會調用此方法，返回null表示交換失敗 
// item是交換的對象，timed表示是否等待指定的時間，為false表示無限期等待，ns為等待時間 
private final Object slotExchange(Object item, boolean timed, long ns) { 
    // 獲取當前線程關聯的participant Node 
    Node p = participant.get(); 
    Thread t = Thread.currentThread(); 
    // 被中斷，返回null 
    if (t.isInterrupted()) // preserve interrupt status so caller can recheck 
        return null; 
     
    for (Node q;;) { 
        if ((q = slot) != null) { // slot 不為null 
            // 將slot置為null，slot對應的線程與當前線程匹配成功 
            if (U.compareAndSwapObject(this, SLOT, q, null)) { 
                Object v = q.item; 
                // 保存item，即完成交互 
                q.match = item; 
                // 喚醒q對應的處于休眠狀態的線程 
                Thread w = q.parked; 
                if (w != null) 
                    U.unpark(w); 
                return v; 
            } 
            // slot修改失敗，其它某個線程搶占來該slot，多個線程同時調用exchange方法會觸發此邏輯 
            // bound等于0表示未初始化，此處校驗避免重復初始化 
            if (NCPU > 1 && bound == 0 && 
                U.compareAndSwapInt(this, BOUND, 0, SEQ)) 
                arena = new Node[(FULL + 2) << ASHIFT]; 
        } 
        else if (arena != null) 
            return null; // carena不為null，通過arenaExchange交互 
        else { 
            // slot和arena都為null 
            p.item = item; 
            // 修改slot為p，修改成功則終止循環 
            if (U.compareAndSwapObject(this, SLOT, null, p)) 
                break; 
            // 修改失敗則繼續for循環，將otem恢復成null 
            p.item = null; 
        } 
    } 
 
    // 將slot修改為p后會進入此分支 
    int h = p.hash; // hash初始為0 
    long end = timed ? System.nanoTime() + ns : 0L; 
    int spins = (NCPU > 1) ? SPINS : 1; 
    Object v; 
    // match保存著同其他線程交換的對象，如果不為null，說明交換成功了 
    while ((v = p.match) == null) { 
        // 執行自旋等待 
        if (spins > 0) { 
        h ^= h << 1; h ^= h >>> 3; h ^= h << 10; 
        if (h == 0) 
            h = SPINS | (int)t.getId(); 初始化h 
        // 只有生成的h小于0時才減少spins 
        else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0) 
            Thread.yield(); 
        } 
        // slot被修改了，已經有匹配的線程，重新自旋，讀取屬性，因為是先修改slot再修改屬性的，兩者因為CPU調度的問題可能有時間差 
        else if (slot != p) 
            spins = SPINS; 
        // 線程沒有被中斷且arena為null 
        else if (!t.isInterrupted() && arena == null && 
                 (!timed || (ns = end - System.nanoTime()) > 0L)) { 
            U.putObject(t, BLOCKER, this); 
            p.parked = t; 
            if (slot == p) 
                U.park(false, ns); 
            // 線程被喚醒，繼續下一次for循環 
            // 如果是因為等待超時而被喚醒，下次for循環進入下沒的else if分支，返回TIMED_OUT 
            p.parked = null; 
                U.putObject(t, BLOCKER, null); 
        } 
        // 將slot修改成p 
        else if (U.compareAndSwapObject(this, SLOT, p, null)) { 
            // timed為flase，無限期等待，因為中斷被喚醒返回null 
            // timed為ture，因為超時被喚醒，返回TIMED_OUT，因為中斷被喚醒返回null 
            v = timed && ns <= 0L && !t.isInterrupted() ? TIMED_OUT : null; 
            break; 
        } 
    } 
    // 修改match為null，item為null，保存h，下一次exchange是h就不是初始值為0了 
    U.putOrderedObject(p, MATCH, null); 
    // 重置 item 
    p.item = null; 
    // 保留偽隨機數,供下次種子數字 
    p.hash = h; 
    // 返回 
    return v; 
}

總結一下上面執行的邏輯：

Exchange 使用了對象池的技術，將對象保存在 ThreadLocal 中，這個對象(Node)封裝了數據項，線程對象等關鍵數據;
第一個線程進入的時候，會將數據放到池化對象中，并賦值給 slot 的 item，并阻塞自己(通常不會立即阻塞，而是使用 yield 自旋一會兒)，等待對方取值;
當第二個線程進入的時候，會拿出存儲在 slot item 中的值，然后對 slot 的 match 賦值，并喚醒上次阻塞的線程;
當第一個線程阻塞被喚醒后，說明對方取到值了，就獲取 slot 的 match 值，并重置 slot 的數據和池化對象的數據，并返回自己的數據;
如果超時了，就返回 Time_out 對象;
如果線程中斷了，就返回 null。

在該方法中，會返回 2 種結果，一是有效的 item，二是 null 要么是線程競爭使用 slot 了，創建了 arena 數組，要么是線程中斷了。

通過一副圖來看看具體邏輯

arenaExchange() 方法

arenaExchange是基于arena屬性完成交換的，整體邏輯比較復雜，有以下幾個要點：

m的初始值就是0，index的初始值也是0，兩個都是大于等于0且i不大于m，當某個線程多次嘗試搶占index對應數組元素的Node都失敗的情形下則嘗試將m加1，然后搶占m加1對應的新數組元素，將其由null修改成當前線程關聯的Node，然后自旋等待匹配;如果自旋結束，沒有匹配的線程，則將m加1對應的新數組元素重新置為null，將m減1，然后再次for循環搶占其他為null的數組元素。極端并發下m會一直增加直到達到最大值FULL為止，達到FULL后只能通過for循環不斷嘗試與其他線程匹配或者搶占為null的數組元素，然后隨著并發減少，m會一直減少到0。通過這種動態調整m的方式可以避免過多的線程基于CAS修改同一個元素導致CAS失敗，提高匹配的效率，這種思想跟LongAdder的實現是一致的。
只有當m等于0的時候才會通過Unsafe park方法讓線程休眠，如果不等于0，即此時存在多個并行的等待匹配的線程，則主要通過自旋的方式等待其他線程到來，這是因為交換動作本身是很快的很短暫的，通過自旋等待就可以讓多個等待的線程快速的完成匹配;只有當前只剩下一個線程的時候，此時m肯定等于0，短期內沒有匹配的線程，才會考慮通過park方法阻塞。

// 搶占slot失敗后進入此方法，arena不為空     
private final Object arenaExchange(Object item, boolean timed, long ns) { 
    Node[] a = arena; 
    Node p = participant.get(); 
    // index初始為0 
    for (int i = p.index;;) {                      // access slot at i 
        int b, m, c; long j;                       // j is raw array offset 
        // 在創建arena時，將本來的數組容量 << ASHIFT,為了避免數組元素落到了同一個高速緩存行 
        // 這里獲取真實的數組元素索引時也需要 << ASHIFR 
        Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE); 
        // 如果q不為null，則將對應的數組元素置為null，表示當前線程和該元素對應的線程匹配l 
        if (q != null && U.compareAndSwapObject(a, j, q, null)) { 
            Object v = q.item;                     // release 
            q.match = item; // 保存item，交互成功 
            Thread w = q.parked; 
            if (w != null) // 喚醒等待的線程 
                U.unpark(w); 
            return v; 
        } 
        // q為null 或者q不為null，cas搶占q失敗了 
        // bound初始化時時SEQ，SEQ & MMASK 就是0，即m的初始值就0，m為0時，i肯定為0 
        else if (i <= (m = (b = bound) & MMASK) && q == null) { 
            p.item = item;                         // offer 
            if (U.compareAndSwapObject(a, j, null, p)) { 
                long end = (timed && m == 0) ? System.nanoTime() + ns : 0L; 
                Thread t = Thread.currentThread(); // wait 
                for (int h = p.hash, spins = SPINS;;) { 
                    Object v = p.match; 
                    if (v != null) { 
                        U.putOrderedObject(p, MATCH, null); 
                        p.item = null;             // clear for next use 
                        p.hash = h; 
                        return v; 
                    } 
                    else if (spins > 0) { 
                        h ^= h << 1; h ^= h >>> 3; h ^= h << 10; // xorshift 
                        if (h == 0)                // initialize hash 
                            h = SPINS | (int)t.getId(); 
                        else if (h < 0 &&          // approx 50% true 
                                 (--spins & ((SPINS >>> 1) - 1)) == 0) 
                            Thread.yield();        // two yields per wait 
                    } 
                    else if (U.getObjectVolatile(a, j) != p) 
                        spins = SPINS;       // releaser hasn't set match yet 
                    else if (!t.isInterrupted() && m == 0 && 
                             (!timed || 
                              (ns = end - System.nanoTime()) > 0L)) { 
                        U.putObject(t, BLOCKER, this); // emulate LockSupport 
                        p.parked = t;              // minimize window 
                        if (U.getObjectVolatile(a, j) == p) 
                            U.park(false, ns); 
                        p.parked = null; 
                        U.putObject(t, BLOCKER, null); 
                    } 
                    else if (U.getObjectVolatile(a, j) == p && 
                             U.compareAndSwapObject(a, j, p, null)) { 
                        if (m != 0)                // try to shrink 
                            U.compareAndSwapInt(this, BOUND, b, b + SEQ - 1); 
                        p.item = null; 
                        p.hash = h; 
                        i = p.index >>>= 1;        // descend 
                        if (Thread.interrupted()) 
                            return null; 
                        if (timed && m == 0 && ns <= 0L) 
                            return TIMED_OUT; 
                        break;                     // expired; restart 
                    } 
                } 
            } 
            else 
                p.item = null;                     // clear offer 
        } 
        else { 
            if (p.bound != b) {                    // stale; reset 
                p.bound = b; 
                p.collides = 0; 
                i = (i != m || m == 0) ? m : m - 1; 
            } 
            else if ((c = p.collides) < m || m == FULL || 
                     !U.compareAndSwapInt(this, BOUND, b, b + SEQ + 1)) { 
                p.collides = c + 1; 
                i = (i == 0) ? m : i - 1;          // cyclically traverse 
            } 
            else 
                i = m + 1;                         // grow 
            p.index = i; 
        } 
    } 
}

總結

Exchange 和 SynchronousQueue 類似，都是通過兩個線程操作同一個對象實現數據交換，只不過就像我們開始說的，SynchronousQueue 使用的是同一個屬性，通過不同的 isData 來區分，多線程并發時，使用了隊列進行排隊。

Exchange 使用了一個對象里的兩個屬性，item 和 match，就不需要 isData 屬性了，因為在 Exchange 里面，沒有 isData 這個語義。而多線程并發時，使用數組來控制，每個線程訪問數組中不同的槽。

PS：以上代碼提交在 Github ：

https://github.com/Niuh-Study/niuh-juc-final.git

PS：這里有一個技術交流群(扣扣群:1158819530)，方便大家一起交流，持續學習，共同進步，有需要的可以加一下。

責任編輯：姜華來源：今日頭條

編程 Exchanger 工具