Java Exchanger

Java Exchanger

Exchanger 是 JDK 1.5 起提供的并发工具类，主要用于两个工作线程之间交换数据，兼具特定的同步特性与应用场景。

核心特性

• 对外操作是同步的，确保线程间数据交换的有序性。
• 专门用于成对出现的线程之间交换数据，不支持多线程同时交叉交换。
• 可看作双向的同步队列，线程需等待伙伴线程到达交换点才能完成操作。
• 适用场景包括基因算法、流水线设计等需要线程间数据交互的场景。

核心接口

Exchanger 类的接口设计简洁，包含一个无参构造函数和两个重载的泛型 exchange 方法：

1. public V exchange(V x) throws InterruptedException：无超时时间的交换方法，线程会一直等待伙伴线程，直至交换完成或被中断。
2. public V exchange(V x, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException：带超时时间的交换方法，若超过指定时间伙伴线程仍未到达，会抛出超时异常。

接口工作原理

当一个线程调用 exchange 方法时，会根据伙伴线程的状态执行不同逻辑：

• 若伙伴线程已提前调用 exchange 方法，当前线程会唤醒伙伴线程，两者完成数据交换后各自返回对方的数据。
• 若伙伴线程未到达交换点，当前线程会被挂起，直至满足以下条件之一：
  1. 伙伴线程到达交换点，完成数据交换后正常返回；
  2. 当前线程被其他线程中断，抛出 InterruptedException；
  3. 等待时间超过设定超时阈值（仅带超时参数的方法），抛出 TimeoutException。

实现代码示例

示例 1：基础数据交换

package com.exchanger;import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;/*** @author Jing61*/
public class ExchangerTest {public static void main(String[] args) {ExecutorService service = Executors.newCachedThreadPool();final Exchanger exchanger = new Exchanger();service.execute(new Runnable() {public void run() {try {String data1 = "白粉";System.out.println("线程" + Thread.currentThread().getName() +"正在把数据" + data1 + "换出去");Thread.sleep((long) (Math.random() * 10000));String data2 = (String) exchanger.exchange(data1);System.out.println("线程" + Thread.currentThread().getName() +"换回的数据为" + data2);} catch (Exception e) {}}});service.execute(new Runnable() {public void run() {try {String data1 = "美金";System.out.println("线程" + Thread.currentThread().getName() +"正在把数据" + data1 + "换出去");Thread.sleep((long) (Math.random() * 10000));String data2 = (String) exchanger.exchange(data1);System.out.println("线程" + Thread.currentThread().getName() +"换回的数据为" + data2);} catch (Exception e) {}}});}
}

示例 2：NBA 球员交易模拟

package com.exchanger;import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.*;/*** @author Jing61*/
public class ExchangerDemo {public static void main(String[] args) {var executor = Executors.newCachedThreadPool();final Exchanger exchanger = new Exchanger();executor.execute(new Runnable() {String data1 = "克拉克森，小拉里南斯";@Overridepublic void run() {nbaTrade(data1, exchanger);}});executor.execute(new Runnable() {String data1 = "格里芬";@Overridepublic void run() {nbaTrade(data1, exchanger);}});executor.execute(new Runnable() {String data1 = "史蒂芬.裤裆里";@Overridepublic void run() {nbaTrade(data1, exchanger);}});executor.shutdown();}private static void nbaTrade(String data1, Exchanger exchanger) {try {System.out.println(Thread.currentThread().getName() + "在交易截止之前把 " + data1 + " 交易出去");Thread.sleep((long) (Math.random() * 1000));String data2 = (String) exchanger.exchange(data1);System.out.println(Thread.currentThread().getName() + "交易得到" + data2);} catch (InterruptedException e) {e.printStackTrace();}}
}