本篇內(nèi)容介紹了“Exchanger的原理與使用方法”的有關(guān)知識(shí)，在實(shí)際案例的操作過程中，不少人都會(huì)遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！
通渭ssl適用于網(wǎng)站、小程序/APP、API接口等需要進(jìn)行數(shù)據(jù)傳輸應(yīng)用場(chǎng)景，ssl證書未來市場(chǎng)廣闊！成為成都創(chuàng)新互聯(lián)的ssl證書銷售渠道，可以享受市場(chǎng)價(jià)格4-6折優(yōu)惠！如果有意向歡迎電話聯(lián)系或者加微信：18982081108（備注：SSL證書合作）期待與您的合作！
前言
在JUC包中，除了一些常用的或者說常見的并發(fā)工具類(ReentrantLock，CountDownLatch，CyclicBarrier，Semaphore)等，還有一個(gè)不常用的線程同步器類 —— Exchanger。
Exchanger是適用在兩個(gè)線程之間數(shù)據(jù)交換的并發(fā)工具類，它的作用是找到一個(gè)同步點(diǎn)，當(dāng)兩個(gè)線程都執(zhí)行到了同步點(diǎn)(exchange方法)之后(有一個(gè)沒有執(zhí)行到就一直等待，也可以設(shè)置等待超時(shí)時(shí)間)，就將自身線程的數(shù)據(jù)與對(duì)方交換。
Exchanger 是什么?
它提供一個(gè)同步點(diǎn)，在這個(gè)同步點(diǎn)兩個(gè)線程可以交換彼此的數(shù)據(jù)。這個(gè)兩個(gè)線程通過exchange方法交換數(shù)據(jù)，如果第一個(gè)線程先執(zhí)行exchange方法，它會(huì)一直等待第二個(gè)線程也執(zhí)行exchange，當(dāng)兩個(gè)線程都到達(dá)同步點(diǎn)時(shí)，這兩個(gè)線程就可以交換數(shù)據(jù)，將本線程生產(chǎn)出來的數(shù)據(jù)傳遞給對(duì)方。因此使用Exchanger的中斷時(shí)成對(duì)的線程使用exchange()方法，當(dāng)有一對(duì)線程到達(dá)了同步點(diǎn)，就會(huì)進(jìn)行交換數(shù)據(jù)，因此該工具類的線程對(duì)象是成對(duì)的。
線程可以在成對(duì)內(nèi)配對(duì)和交換元素的同步點(diǎn)。每個(gè)線程在輸入exchange方法時(shí)提供一些對(duì)象，與合作者線程匹配，并在返回時(shí)接收其合作伙伴的對(duì)象。交換器可以被視為一個(gè)的雙向形式的SynchroniuzedQueue。交換器在諸如遺傳算法和管道設(shè)計(jì)的應(yīng)用中可能是有用的。
一個(gè)用于兩個(gè)工作線程之間交換數(shù)據(jù)的封裝工具類，簡(jiǎn)單說就是一個(gè)線程在完成一定事務(wù)后想與另一個(gè)線程交換數(shù)據(jù)，則第一個(gè)先拿出數(shù)據(jù)的線程會(huì)一直等待第二個(gè)線程，直到第二個(gè)線程拿著數(shù)據(jù)到來時(shí)才能彼此交換對(duì)應(yīng)數(shù)據(jù)。
Exchanger 用法
Exchanger 泛型類型，其中V表示可交換的數(shù)據(jù)類型
V exchanger(V v)：等待另一個(gè)線程到達(dá)此交換點(diǎn)(除非當(dāng)前線程被中斷)，然后將給定的對(duì)象傳送該線程，并接收該線程的對(duì)象。
V exchanger(V v, long timeout, TimeUnit unit)：等待另一個(gè)線程到達(dá)此交換點(diǎn)(除非當(dāng)前線程被中斷或超出類指定的等待時(shí)間)，然后將給定的對(duì)象傳送給該線程，并接收該線程的對(duì)象。
應(yīng)用場(chǎng)景
Exchanger可以用于遺傳算法，遺傳算法里需要選出兩個(gè)人作為交配對(duì)象，這時(shí)候會(huì)交換兩人的數(shù)據(jù)，并使用交叉規(guī)則得出2個(gè)交配結(jié)果。
Exchanger也可以用于校對(duì)工作。比如我們需要將紙制銀流通過人工的方式錄入成電子銀行流水，為了避免錯(cuò)誤，采用AB崗兩人進(jìn)行錄入，錄入到Excel之后，系統(tǒng)需要加載這兩個(gè)Excel，并對(duì)這兩個(gè)Excel數(shù)據(jù)進(jìn)行校對(duì)，看看是否錄入的一致
Exchanger的典型應(yīng)用場(chǎng)景是：一個(gè)任務(wù)在創(chuàng)建對(duì)象，而這些對(duì)象的生產(chǎn)代價(jià)很高，另一個(gè)任務(wù)在消費(fèi)這些對(duì)象。通過這種方式，可以有更多的對(duì)象在被創(chuàng)建的同時(shí)被消費(fèi)。
案例說明
Exchanger 用于兩個(gè)線程間交換數(shù)據(jù)，當(dāng)然實(shí)際參與的線程可以不止兩個(gè)，測(cè)試用例如下：
重慶分公司
重慶分公司
private static void test1() throws InterruptedException { Exchanger exchanger = new Exchanger<>(); CountDownLatch countDownLatch = new CountDownLatch(5); for (int i = 0; i < 5; i++) { new Thread(() -> { try { String origMsg = RandomStringUtils.randomNumeric(6); // 先到達(dá)的線程會(huì)在此等待，直到有一個(gè)線程跟它交換數(shù)據(jù)或者等待超時(shí) 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個(gè)線程因?yàn)闆]有匹配的線程而等待超時(shí)，輸出如下：
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)
上述測(cè)試用例是比較簡(jiǎn)單，可以模擬消息消費(fèi)的場(chǎng)景來觀察Exchanger的行為，測(cè)試用例如下：
private static void test2() throws InterruptedException { Exchanger exchanger = new Exchanger<>(); CountDownLatch countDownLatch = new CountDownLatch(4); CyclicBarrier cyclicBarrier = new CyclicBarrier(4); // 生產(chǎn)者 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(); } } }; // 消費(fèi)者 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(); } 輸出如下，上面生產(chǎn)者和消費(fèi)者線程數(shù)是一樣的，循環(huán)次數(shù)也是一樣的，但是還是出現(xiàn)等待超時(shí)的情形：
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
這種等待超時(shí)是概率出現(xiàn)的，這是為啥?
因?yàn)橄到y(tǒng)調(diào)度的不均衡和Exchanger底層的大量自旋等待導(dǎo)致這4個(gè)線程并不是調(diào)用exchanger成功的次數(shù)并不一致。另外從輸出可以看出，消費(fèi)者線程并沒有像我們想的那樣跟生產(chǎn)者線程一一匹配，生產(chǎn)者線程有時(shí)也充當(dāng)來消費(fèi)者線程，這是為啥?因?yàn)镋xchanger匹配時(shí)完全不關(guān)注這個(gè)線程的角色，兩個(gè)線程之間的匹配完全由調(diào)度決定的，即CPU同時(shí)執(zhí)行來或者緊挨著執(zhí)行來兩個(gè)線程，這兩個(gè)線程就匹配成功來。
源碼分析
Exchanger 類圖
其內(nèi)部主要變量和方法如下：
成員屬性
// ThreadLocal變量，每個(gè)線程都有之間的一個(gè)副本 private final Participant participant; // 高并發(fā)下使用的，保存待匹配的Node實(shí)例 private volatile Node[] arena; // 低并發(fā)下，arena未初始化時(shí)使用的保存待匹配的Node實(shí)例 private volatile Node slot; // 初始值為0，當(dāng)創(chuàng)建arena后被負(fù)責(zé)SEQ，用來記錄arena數(shù)組的可用最大索引， // 會(huì)隨著并發(fā)的增大而增大直到等于最大值FULL， // 會(huì)隨著并行的線程逐一匹配成功而減少恢復(fù)成初始值 private volatile int bound;
還有多個(gè)表示字段偏移量的靜態(tài)屬性，通過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 定義來多個(gè)靜態(tài)變量，如下：
// 初始化arena時(shí)使用， 1 << ASHIFT 是一個(gè)緩存行的大小，避免來不同的Node落入到同一個(gè)高速緩存行 // 這里實(shí)際是把數(shù)組容量擴(kuò)大來8倍，原來索引相鄰的兩個(gè)元素，擴(kuò)容后中間隔來7個(gè)元素，從元素的起始地址上看就隔來8個(gè)元素，中間的7個(gè)都是空的，為來避免原來相鄰的兩個(gè)元素都落入到同一個(gè)緩存行中 // 因?yàn)閍rena是對(duì)象數(shù)組，一個(gè)元素占8字節(jié)，8個(gè)就是64字節(jié) private static final int ASHIFT = 7; // arena 數(shù)組元素的索引最大值即255 private static final int MMASK = 0xff; // arena 數(shù)組的最大長(zhǎng)度即256 private static final int SEQ = MMASK + 1; // 獲取CPU核數(shù) private static final int NCPU = Runtime.getRuntime().availableProcessors(); // 實(shí)際的數(shù)組長(zhǎng)度，因?yàn)槭蔷€程兩兩配對(duì)的，所以最大長(zhǎng)度是核數(shù)除以2 static final int FULL = (NCPU >= (MMASK << 1)) ? MMASK : NCPU >>> 1; // 自旋等待的次數(shù) private static final int SPINS = 1 << 10; // 如果交換的對(duì)象是null，則返回此對(duì)象 private static final Object NULL_ITEM = new Object(); // 如果等待超時(shí)導(dǎo)致交換失敗，則返回此對(duì)象 private static final Object TIMED_OUT = new Object();
內(nèi)部類
Exchanger類中有兩個(gè)內(nèi)部類，一個(gè)Node，一個(gè)Participant。 Participant繼承了ThreadLocal并且重寫了其initialValue方法，返回一個(gè)Node對(duì)象。其定義如下：
@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 { public Node initialValue() { return new Node(); } } 其中Contended注解是為了避免高速緩存行導(dǎo)致的偽共享問題
index用來記錄arena數(shù)組的索引
bound用于記錄上一次的Exchanger bound屬性
collides用于記錄在bound不變的情況下CAS搶占失敗的次數(shù)
hash是自旋等待時(shí)計(jì)算隨機(jī)數(shù)使用的
item表示當(dāng)前線程請(qǐng)求交換的對(duì)象
match是同其它線程交換的結(jié)果，match不為null表示交換成功
parked為跟該Node關(guān)聯(lián)的處于休眠狀態(tài)的線程。
重要方法
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就執(zhí)行后面的方法 (v = slotExchange(item, false, 0L)) == null) && // 如果執(zhí)行slotExchange有結(jié)果就執(zhí)行后面的，否則返回 ((Thread.interrupted() || // 非中斷則執(zhí)行后面的方法 (v = arenaExchange(item, false, 0L)) == null))) throw new InterruptedException(); return (v == NULL_ITEM) ? null : (V)v; }exchange 方法的執(zhí)行步驟：
鴻蒙官方戰(zhàn)略合作共建——HarmonyOS技術(shù)社區(qū)
如果執(zhí)行 soltExchange 有結(jié)果就執(zhí)行后面的 arenaExchange;
如果 slot 被占用，就執(zhí)行 arenaExchange;
返回的數(shù)據(jù) v 是對(duì)方線程的數(shù)據(jù)項(xiàng);
總結(jié)即：如果A線程先調(diào)用，那么A的數(shù)據(jù)項(xiàng)存儲(chǔ)的 item中，則B線程的數(shù)據(jù)項(xiàng)存儲(chǔ)在 math 中;
當(dāng)沒有多線程并發(fā)操作 Exchange 的時(shí)候，使用 slotExchange 就足夠了，slot 是一個(gè) node 對(duì)象;
當(dāng)出現(xiàn)并發(fā)了，一個(gè) slot 就不夠了，就需要使用一個(gè) node 數(shù)組 arena 操作了。
slotExchange()方法
slotExchange 是基于slot屬性來完成交換的，調(diào)用soltExchange方法時(shí)，如果slot屬性為null，當(dāng)前線程會(huì)將slot屬性由null修改成當(dāng)前線程的Node，如果修改失敗則下一次for循環(huán)走solt屬性不為null的邏輯，如果修改成功則自旋等待，自旋一定次數(shù)后通過Unsafe的park方法當(dāng)當(dāng)前線程休眠，可以指定休眠的時(shí)間，如果沒有指定則無限期休眠直到被喚醒;無論是因?yàn)榫€程中斷被喚醒，等待超時(shí)被喚醒還是其它線程unpark喚醒的，都會(huì)檢查當(dāng)前線程的Node的屬性釋放為null，如果不為null說明交互成功，返回該對(duì)象;否則返回null或者TIME_OUT,在返回前會(huì)將item，match等屬性置為null，保存之前自旋時(shí)計(jì)算的hash值，方便下一次調(diào)用slotExchange。
調(diào)用slotExchange方法時(shí)，如果slot屬性不為null，則當(dāng)前線程會(huì)嘗試將其修改null，如果cas修改成功，表示當(dāng)前線程與slot屬性對(duì)應(yīng)的線程匹配成功，會(huì)獲取slot屬性對(duì)應(yīng)Node的item屬性，將當(dāng)前線程交換的對(duì)象保存到slot屬性對(duì)應(yīng)的Node的match屬性，然后喚醒獲取slot屬性對(duì)應(yīng)Node的waiter屬性，即處理休眠狀態(tài)的線程，至此交換完成，同樣的在返回前需要將item，match等屬性置為null，保存之前自旋時(shí)計(jì)算的hash置，方便下一次調(diào)用slotExchange;如果cas修改slot屬性失敗，說明有其它線程也在搶占slot，則初始化arena屬性，下一次for循環(huán)因?yàn)閍rena屬性不為null，直接返回null，從而通過arenaExchange完成交換。
// arena 為null是會(huì)調(diào)用此方法，返回null表示交換失敗 // item是交換的對(duì)象，timed表示是否等待指定的時(shí)間，為false表示無限期等待，ns為等待時(shí)間 private final Object slotExchange(Object item, boolean timed, long ns) { // 獲取當(dāng)前線程關(guān)聯(lián)的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對(duì)應(yīng)的線程與當(dāng)前線程匹配成功 if (U.compareAndSwapObject(this, SLOT, q, null)) { Object v = q.item; // 保存item，即完成交互 q.match = item; // 喚醒q對(duì)應(yīng)的處于休眠狀態(tài)的線程 Thread w = q.parked; if (w != null) U.unpark(w); return v; } // slot修改失敗，其它某個(gè)線程搶占來該slot，多個(gè)線程同時(shí)調(diào)用exchange方法會(huì)觸發(fā)此邏輯 // bound等于0表示未初始化，此處校驗(yàn)避免重復(fù)初始化 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，修改成功則終止循環(huán) if (U.compareAndSwapObject(this, SLOT, null, p)) break; // 修改失敗則繼續(xù)for循環(huán)，將otem恢復(fù)成null p.item = null; } } // 將slot修改為p后會(huì)進(jìn)入此分支 int h = p.hash; // hash初始為0 long end = timed ? System.nanoTime() + ns : 0L; int spins = (NCPU > 1) ? SPINS : 1; Object v; // match保存著同其他線程交換的對(duì)象，如果不為null，說明交換成功了 while ((v = p.match) == null) { // 執(zhí)行自旋等待 if (spins > 0) { h ^= h << 1; h ^= h >>> 3; h ^= h << 10; if (h == 0) h = SPINS | (int)t.getId(); 初始化h // 只有生成的h小于0時(shí)才減少spins else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0) Thread.yield(); } // slot被修改了，已經(jīng)有匹配的線程，重新自旋，讀取屬性，因?yàn)槭窍刃薷膕lot再修改屬性的，兩者因?yàn)镃PU調(diào)度的問題可能有時(shí)間差 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); // 線程被喚醒，繼續(xù)下一次for循環(huán) // 如果是因?yàn)榈却瑫r(shí)而被喚醒，下次for循環(huán)進(jìn)入下沒的else if分支，返回TIMED_OUT p.parked = null; U.putObject(t, BLOCKER, null); } // 將slot修改成p else if (U.compareAndSwapObject(this, SLOT, p, null)) { // timed為flase，無限期等待，因?yàn)橹袛啾粏拘逊祷豱ull // timed為ture，因?yàn)槌瑫r(shí)被喚醒，返回TIMED_OUT，因?yàn)橹袛啾粏拘逊祷豱ull 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; // 保留偽隨機(jī)數(shù),供下次種子數(shù)字 p.hash = h; // 返回 return v; }總結(jié)一下上面執(zhí)行的邏輯：
Exchange 使用了對(duì)象池的技術(shù)，將對(duì)象保存在 ThreadLocal 中，這個(gè)對(duì)象(Node)封裝了數(shù)據(jù)項(xiàng)，線程對(duì)象等關(guān)鍵數(shù)據(jù);
第一個(gè)線程進(jìn)入的時(shí)候，會(huì)將數(shù)據(jù)放到池化對(duì)象中，并賦值給 slot 的 item，并阻塞自己(通常不會(huì)立即阻塞，而是使用 yield 自旋一會(huì)兒)，等待對(duì)方取值;
當(dāng)?shù)诙€(gè)線程進(jìn)入的時(shí)候，會(huì)拿出存儲(chǔ)在 slot item 中的值，然后對(duì) slot 的 match 賦值，并喚醒上次阻塞的線程;
當(dāng)?shù)谝粋€(gè)線程阻塞被喚醒后，說明對(duì)方取到值了，就獲取 slot 的 match 值，并重置 slot 的數(shù)據(jù)和池化對(duì)象的數(shù)據(jù)，并返回自己的數(shù)據(jù);
如果超時(shí)了，就返回 Time_out 對(duì)象;
如果線程中斷了，就返回 null。
在該方法中，會(huì)返回 2 種結(jié)果，一是有效的 item，二是 null 要么是線程競(jìng)爭(zhēng)使用 slot 了，創(chuàng)建了 arena 數(shù)組，要么是線程中斷了。
通過一副圖來看看具體邏輯
arenaExchange() 方法
arenaExchange是基于arena屬性完成交換的，整體邏輯比較復(fù)雜，有以下幾個(gè)要點(diǎn)：
m的初始值就是0，index的初始值也是0，兩個(gè)都是大于等于0且i不大于m，當(dāng)某個(gè)線程多次嘗試搶占index對(duì)應(yīng)數(shù)組元素的Node都失敗的情形下則嘗試將m加1，然后搶占m加1對(duì)應(yīng)的新數(shù)組元素，將其由null修改成當(dāng)前線程關(guān)聯(lián)的Node，然后自旋等待匹配;如果自旋結(jié)束，沒有匹配的線程，則將m加1對(duì)應(yīng)的新數(shù)組元素重新置為null，將m減1，然后再次for循環(huán)搶占其他為null的數(shù)組元素。極端并發(fā)下m會(huì)一直增加直到達(dá)到最大值FULL為止，達(dá)到FULL后只能通過for循環(huán)不斷嘗試與其他線程匹配或者搶占為null的數(shù)組元素，然后隨著并發(fā)減少，m會(huì)一直減少到0。通過這種動(dòng)態(tài)調(diào)整m的方式可以避免過多的線程基于CAS修改同一個(gè)元素導(dǎo)致CAS失敗，提高匹配的效率，這種思想跟LongAdder的實(shí)現(xiàn)是一致的。
只有當(dāng)m等于0的時(shí)候才會(huì)通過Unsafe park方法讓線程休眠，如果不等于0，即此時(shí)存在多個(gè)并行的等待匹配的線程，則主要通過自旋的方式等待其他線程到來，這是因?yàn)榻粨Q動(dòng)作本身是很快的很短暫的，通過自旋等待就可以讓多個(gè)等待的線程快速的完成匹配;只有當(dāng)前只剩下一個(gè)線程的時(shí)候，此時(shí)m肯定等于0，短期內(nèi)沒有匹配的線程，才會(huì)考慮通過park方法阻塞。
// 搶占slot失敗后進(jìn)入此方法，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 // 在創(chuàng)建arena時(shí)，將本來的數(shù)組容量 << ASHIFT,為了避免數(shù)組元素落到了同一個(gè)高速緩存行 // 這里獲取真實(shí)的數(shù)組元素索引時(shí)也需要 << ASHIFR Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE); // 如果q不為null，則將對(duì)應(yīng)的數(shù)組元素置為null，表示當(dāng)前線程和該元素對(duì)應(yīng)的線程匹配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初始化時(shí)時(shí)SEQ，SEQ & MMASK 就是0，即m的初始值就0，m為0時(shí)，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; } } }總結(jié)
Exchange 和 SynchronousQueue 類似，都是通過兩個(gè)線程操作同一個(gè)對(duì)象實(shí)現(xiàn)數(shù)據(jù)交換，只不過就像我們開始說的，SynchronousQueue 使用的是同一個(gè)屬性，通過不同的 isData 來區(qū)分，多線程并發(fā)時(shí)，使用了隊(duì)列進(jìn)行排隊(duì)。
Exchange 使用了一個(gè)對(duì)象里的兩個(gè)屬性，item 和 match，就不需要 isData 屬性了，因?yàn)樵?Exchange 里面，沒有 isData 這個(gè)語(yǔ)義。而多線程并發(fā)時(shí)，使用數(shù)組來控制，每個(gè)線程訪問數(shù)組中不同的槽。
“Exchanger的原理與使用方法”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！
