摘要：JDK 1.5 開始 JUC 包下提供的 Exchanger 類可用於兩個執行緒之間交換資訊。
本文分享自華為雲社群《一行Java程式碼實現兩玩家交換裝備【併發程式設計】》，作者：陳皮的JavaLib 。
1 Exchanger 是什麼
JDK 1.5 開始 JUC 包下提供的 Exchanger 類可用於兩個執行緒之間交換資訊。Exchanger 物件可理解為一個包含2個格子的容器，通過呼叫 exchanger 方法向其中的格子填充資訊，當兩個格子中的均被填充資訊時，自動交換兩個格子中的資訊，然後將交換的資訊返回給呼叫執行緒，從而實現兩個執行緒的資訊交換。
功能看似簡單，但這在某些場景下是很有用處的，例如遊戲中兩個玩家交換裝備；交友軟體男女心儀物件匹配。
下面簡單模擬下兩個玩家交換裝備的場景。
package com.chenpi; import java.util.concurrent.Exchanger; /**
* @Description
* @Author 陳皮
* @Date 2021/7/11
* @Version 1.0
*/
public class ChenPiMain { public static void main(String[] args) throws InterruptedException { Exchanger<String> exchanger = new Exchanger<>(); new Thread(() -> {
String str = null;
try {
str = exchanger.exchange("屠龍刀");
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("交易成功，" + Thread.currentThread().getName() + "獲得" + str);
}, "周芷若").start(); new Thread(() -> {
String str = null;
try {
str = exchanger.exchange("倚天劍");
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("交易成功，" + Thread.currentThread().getName() + "獲得" + str);
}, "張無忌").start();
}
} // 輸出結果如下
交易成功，張無忌獲得屠龍刀
交易成功，周芷若獲得倚天劍
2 Exchanger 詳解
Exchager 類可用於兩個執行緒之間交換資訊，如果一個執行緒呼叫了 Exchanger 物件的 exchange 方法之後，會一直阻塞直到另一個執行緒來和它交換資訊，交換之後的資訊返回給呼叫執行緒，從而實現兩個執行緒的資訊交換。
Exchager 底層也是使用到了自旋和 cas 機制。
注意，如果超過兩個執行緒呼叫同一個 Exchanger 物件 exchange 方法時，結果是不可預計的，只要有2個執行緒滿足條件了，就認為匹配成功並交換資訊。而剩下的未能得到配對的執行緒，則會被阻塞一直等待直到有另一個執行緒能與它匹配與之配對。
package com.chenpi; import java.util.concurrent.Exchanger; /**
* @Description
* @Author 陳皮
* @Date 2021/7/11
* @Version 1.0
*/
public class ChenPiMain { public static void main(String[] args) { Exchanger<String> exchanger = new Exchanger<>(); new Thread(() -> {
String str = null;
try {
str = exchanger.exchange("屠龍刀");
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("交易成功，" + Thread.currentThread().getName() + "獲得" + str);
}, "周芷若").start(); new Thread(() -> {
String str = null;
try {
str = exchanger.exchange("倚天劍");
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("交易成功，" + Thread.currentThread().getName() + "獲得" + str);
}, "張無忌").start(); new Thread(() -> {
String str = null;
try {
str = exchanger.exchange("假的倚天劍");
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("交易成功，" + Thread.currentThread().getName() + "獲得" + str);
}, "成昆").start();
}
} // 輸出結果如下
交易成功，周芷若獲得假的倚天劍
交易成功，成昆獲得屠龍刀
當然，在等待交換資訊的執行緒是可以被中斷的，就比如玩家在等待交易過程中，突然玩家下線了，那就應該中斷執行緒等待。
package com.chenpi; import java.lang.Thread.State;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Exchanger; /**
* @Description
* @Author 陳皮
* @Date 2021/7/11
* @Version 1.0
*/
public class ChenPiMain { public static void main(String[] args) throws InterruptedException { Exchanger<String> exchanger = new Exchanger<>(); List<Thread> threads = new ArrayList<>(3); Thread thread1 = new Thread(() -> {
String str = null;
try {
str = exchanger.exchange("屠龍刀");
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("交易成功，" + Thread.currentThread().getName() + "獲得" + str);
}, "周芷若");
threads.add(thread1); Thread thread2 = new Thread(() -> {
String str = null;
try {
str = exchanger.exchange("倚天劍");
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("交易成功，" + Thread.currentThread().getName() + "獲得" + str);
}, "張無忌");
threads.add(thread2); Thread thread3 = new Thread(() -> {
String str = null;
try {
str = exchanger.exchange("假的屠龍刀");
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("交易成功，" + Thread.currentThread().getName() + "獲得" + str);
}, "成昆");
threads.add(thread3); for (Thread thread : threads) {
thread.start();
} // 等待5秒
Thread.sleep(5000); for (Thread thread : threads) {
System.out.println(thread.getName() + ":" + thread.getState());
// 如果還在阻塞等待則中斷執行緒
if (thread.getState() == State.WAITING) {
thread.interrupt();
}
}
}
} // 輸出結果如下
交易成功，張無忌獲得屠龍刀
交易成功，周芷若獲得倚天劍
周芷若:TERMINATED
張無忌:TERMINATED
成昆:WAITING
交易成功，成昆獲得null
java.lang.InterruptedException
at java.util.concurrent.Exchanger.exchange(Exchanger.java:568)
at com.chenpi.ChenPiMain.lambda$main$2(ChenPiMain.java:47)
at java.lang.Thread.run(Thread.java:748)
上面演示的是執行緒如果等不到另一個執行緒和它交換資訊，則會一直等待下去。其實 Exchanger 還可以設定等待指定時間。比如系統設定玩家交換裝備匹配時間為60秒，如果超出時間則終止交易。
package com.chenpi; import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException; /**
* @Description
* @Author 陳皮
* @Date 2021/7/11
* @Version 1.0
*/
public class ChenPiMain { public static void main(String[] args) { Exchanger<String> exchanger = new Exchanger<>(); new Thread(() -> {
try {
// 超時時間設定為5秒
String str = exchanger.exchange("屠龍刀", 5, TimeUnit.SECONDS);
System.out.println("交易成功，" + Thread.currentThread().getName() + "獲得" + str);
} catch (TimeoutException e) {
System.out.println("交易超時！");
e.printStackTrace();
} catch (InterruptedException e) {
System.out.println("交易異常終止");
e.printStackTrace();
}
}, "周芷若").start();
}
} // 輸出結果如下
交易超時！
java.util.concurrent.TimeoutException
at java.util.concurrent.Exchanger.exchange(Exchanger.java:626)
at com.chenpi.ChenPiMain.lambda$main$0(ChenPiMain.java:22)
at java.lang.Thread.run(Thread.java:748)
3 Exchanger 應用
Exchager 在遺傳演算法和管道設計等應用中是非常有用的。比如兩個執行緒之間交換緩衝區，填充緩衝區的執行緒在需要時從另一個執行緒獲得一個剛清空的緩衝區，並將填充的緩衝區傳遞給清空緩衝區的執行緒。
package com.chenpi; import java.awt.image.DataBuffer;
import java.util.concurrent.Exchanger; /**
* @Description
* @Author 陳皮
* @Date 2021/7/11
* @Version 1.0
*/
public class ChenPiMain { Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();
DataBuffer initialEmptyBuffer = ... a made-up type
DataBuffer initialFullBuffer = ... class FillingLoop implements Runnable { public void run() {
DataBuffer currentBuffer = initialEmptyBuffer;
try {
while (currentBuffer != null) {
addToBuffer(currentBuffer);
if (currentBuffer.isFull()) {
currentBuffer = exchanger.exchange(currentBuffer);
}
}
} catch (InterruptedException ex) { ...handle ...}
}
} class EmptyingLoop implements Runnable { public void run() {
DataBuffer currentBuffer = initialFullBuffer;
try {
while (currentBuffer != null) {
takeFromBuffer(currentBuffer);
if (currentBuffer.isEmpty()) {
currentBuffer = exchanger.exchange(currentBuffer);
}
}
} catch (InterruptedException ex) { ...handle ...}
}
} void start() {
new Thread(new FillingLoop()).start();
new Thread(new EmptyingLoop()).start();
}
}