原始碼分析：Exchanger之資料交換器
阿新 • • 發佈：2021-02-20
## 簡介
[Exchanger](https://jinglingwang.cn)是Java5 開始引入的一個類，它允許兩個執行緒之間交換持有的資料。當[Exchanger](https://jinglingwang.cn)在一個執行緒中呼叫exchange方法之後，會阻塞等待另一個執行緒呼叫同樣的exchange方法，然後以執行緒安全的方式交換資料，之後執行緒繼續執行。
### 官方示例
在JDK的原始碼註釋中，提供了一個簡單的示例demo，稍加修改後就可以執行
```java
public class FillAndEmpty {
Exchanger exchanger = new Exchanger();
Integer initialEmptyBuffer = 1;
Integer initialFullBuffer = 2;
class FillingLoop implements Runnable {
public void run() {
Integer currentBuffer = initialEmptyBuffer;
try {
while (currentBuffer != 2) {
currentBuffer = exchanger.exchange(currentBuffer);
}
System.out.println("FillingLoop:"+currentBuffer);
} catch (InterruptedException ex) {
}
}
}
class EmptyingLoop implements Runnable {
public void run() {
Integer currentBuffer = initialFullBuffer;
try {
while (currentBuffer != 1) {
currentBuffer = exchanger.exchange(currentBuffer);
}
System.out.println("EmptyingLoop:"+currentBuffer);
} catch (InterruptedException ex) {
}
}
}
void start() {
new Thread(new FillingLoop()).start();
new Thread(new EmptyingLoop()).start();
}
public static void main(String[] args){
FillAndEmpty f = new FillAndEmpty();
f.start();
}
}
```
## 原始碼分析
### 內部類
Exchanger 中定義了兩個內部類：Node、Participant
```java
// 使用 @sun.misc.Contended 註解避免出現偽共享
@sun.misc.Contended static final class Node {
int index; // Arena 中的索引
int bound; // Exchanger.bound的最後記錄值
int collides; // 當前 bound 的CAS 失敗數
int hash; // Pseudo-random for spins
Object item; // 執行緒的當前資料項
volatile Object match; // 由釋放執行緒提供的專案
volatile Thread parked; // 當阻塞（parked）時，設定此執行緒，否則為null
}
```
```java
/** 繼承了ThreadLocal，並初始化了Node物件 */
static final class Participant extends ThreadLocal {
public Node initialValue() { return new Node(); }
}
```
### 重要的屬性
```java
/** 每個執行緒的狀態 */
private final Participant participant;
/** 消除陣列；在啟用（在slotExchange中）之前為空。元素訪問使用volatile get和CAS */
private volatile Node[] arena;
/** 在檢測到爭用之前一直使用的插槽，可以理解為先到的執行緒的資料項 */
private volatile Node slot;
/** 每次更新時，將最大有效競技場位置的索引與高位SEQ號進行“或”運算。 */
private volatile int bound;
```
### exchange()方法
等待另一個執行緒到達交換點(除非當前執行緒被中斷)，然後將給定的物件傳遞給它，作為回報接收另一個的物件。
```java
public V exchange(V x) throws InterruptedException {
// 交換後的物件v
Object v;
// item 為交換出去的物件，如果為null則換成NULL_ITEM物件
Object item = (x == null) ? NULL_ITEM : x; // translate null args
// 1.1構造方法沒有初始化arena,所以第一個進來的執行緒看見的arena肯定為null
// 1.2第一個進來的執行緒繼續呼叫slotExchange(item, false, 0L)方法
if ((arena != null || (v = slotExchange(item, false, 0L)) == null) &&
// 2.1 Thread.interrupted()： 檢測執行緒是否有被中斷
// 2.2 arenaExchange(item, false, 0L)：slotExchange方法 返回了null時會進入到這個方法
((Thread.interrupted() || (v = arenaExchange(item, false, 0L)) == null)))
throw new InterruptedException();
return (v == NULL_ITEM) ? null : (V)v;
}
```
arenaExchange()方法總結：
1. 呼叫exchange方法的執行緒等待另一個執行緒到達交換點完成交換資料
2. 如果交換的資料為null，會被轉換成一個`NULL_ITEM` 的Object物件作為轉換的資料項
3. 構造方法未初始化`arena`物件，所以會先呼叫`slotExchange`方法借用slot插槽來交換物件
4. 如果`slotExchange`方法成功返回了另一個交換到的物件，則直接返回交換到的資料項
5. 如果`slotExchange`方法成功返回了null，會繼續呼叫`arenaExchange`方法完成資料交換並返回
### slotExchange()方法
```java
/**
* item:要交換的專案
* timed:是否有設定超時
* ns: 設定的超時時間
* return: 返回另一個執行緒的資料項；如果啟用arena或執行緒在完成之前被中斷，則為null；如果超時，則為TIMED_OUT
*/
private final Object slotExchange(Object item, boolean timed, long ns) {
// 獲取當前執行緒node節點物件
Node p = participant.get();
Thread t = Thread.currentThread(); // 當前執行緒
if (t.isInterrupted()) // preserve interrupt status so caller can recheck
return null;
// 自旋
for (Node q;;) {
if ((q = slot) != null) { // 兩個執行緒先到的執行緒，slot肯定為null,一般後到的執行緒會進入到這個if分支
// 如果在當前執行緒之前已經有執行緒呼叫了exchange方法，slot就肯定不為null,條件成立
if (U.compareAndSwapObject(this, SLOT, q, null)) {// 後來的執行緒會呼叫CAS吧slot再置為null
// q.item 是較早的執行緒的資料項
Object v = q.item;
// item 是當前執行緒的資料項；by: https://jinglingwang.cn
q.match = item;
// 之前阻塞（park）的執行緒
Thread w = q.parked;
if (w != null) //可能另一個執行緒還在自旋，沒有阻塞，所以這裡可能會為null
// 喚醒之前被阻塞的執行緒
U.unpark(w);
// 返回之前的執行緒的資料項
return v;
}
// create arena on contention, but continue until slot null
// 上面CAS修改slot失敗後，會進入到這裡；https://jinglingwang.cn
// SEQ = MMASK + 1 = 256
if (NCPU > 1 && bound == 0 && U.compareAndSwapInt(this, BOUND, 0, SEQ))
// if條件成立，初始化arena陣列
// 我8核的CPU，計算的length是 (4+2) << 7 == 768
arena = new Node[(FULL + 2) << ASHIFT];
}
else if (arena != null)
// 如果上面的if條件成立並且初始化了arena陣列，會進入到arenaExchange方法
return null; // caller must reroute to arenaExchange
else {
p.item = item; // p節點的item設定為當前項item
if (U.compareAndSwapObject(this, SLOT, null, p)) // CAS 修改slot的值，修改成功退出自旋
break;
p.item = null; //CAS 修改失敗沒有退出自旋，重置p節點的item為null
}
}
// 理論上第一個先到的執行緒會進入到下面，會阻塞自己，等待另一個執行緒的資料項到來
// await release
int h = p.hash;
long end = timed ? System.nanoTime() + ns : 0L; // 超時時間
// 根據CPU的核數確定自旋的次數1024 or 1
int spins = (NCPU >