并发工具类（四）线程间的交换数据 Exchanger

前言

JDK中为了处理线程之间的同步问题，除了提供锁机制之外，还提供了几个非常有用的并发工具类：CountDownLatch、CyclicBarrier、Semphore、Exchanger、Phaser；
CountDownLatch、CyclicBarrier、Semphore、Phaser 这四个工具类提供一种并发流程的控制手段；而Exchanger工具类则提供了在线程之间交换数据的一种手段。

简介

Exchanger的功能是使2个线程之间交换数据（有不少文章的说法是“传输数据”，应该叫“交换数据”更合适，因为这是两个线程都要向对方传送数据，同时也获取对方的传送过来的数据，是

双向模式

，并不是一个线程向另一个线程传输数据）。它比生产者/消费者模式使用的wait/notify要更加方便。
Exchanger 提供一个同步点，在这个同步点处，两个线程可以交换彼此数据。即一个线程调用了exchange( )方法交换数据，到达了同步点，然后就会一直阻塞等待另一个线程调用exchange( )方法来交换数据。所以，要注意

exchange( )方法是有阻塞的特性。

Exchanger 可能在应用程序（比如

遗传算法和管道设计

）中很有用。

方法摘要

public V exchange(V x) throws InterruptedException
等待另一个线程到达此交换点（除非当前线程被中断），然后将给定的对象传送给该线程，并接收该线程的对象。
public V exchange(V x, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
等待另一个线程到达此交换点（除非当前线程被中断，或者超出了指定的等待时间），然后将给定的对象传送给该线程，同时接收该线程的对

@ Example1 用法示例

以下是重点介绍的一个类，该类使用 Exchanger 在线程间交换缓冲区，因此，在需要时，填充缓冲区的线程获取一个新腾空的缓冲区，并将填满的缓冲区传递给腾空缓冲区的线程

class FillAndEmpty {Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();DataBuffer initialEmptyBuffer = ... a made-up typeDataBuffer 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();}}
复制代码

@ Example2 应用场景示例

Exchanger可以用于

遗传算法，遗传算法里需要选出两个人作为交配对象，这时候会交换两人的数据，并使用交叉规则得出2个交配结果。

Exchanger也可以用于

校对工作。

比如我们需要将纸制银流通过人工的方式录入成电子银行流水，为了避免错误，采用AB岗两人进行录入，录入到Excel之后，系统需要加载这两个Excel，并对这两个Excel数据进行校对，看看是否录入的一致。代码如下：

private static final Exchanger<String> exgr = new Exchanger<String>();
private static ExecutorService threadPool = Executors.newFixedThreadPool(2);public static void main(String[] args) {threadPool.execute(new Runnable() {@Overridepublic void run() {try {String A = "银行流水A";// A录入银行流水数据exgr.exchange(A);//同步点，交换数据} catch (InterruptedException e) {}}});threadPool.execute(new Runnable() {@Overridepublic void run() {try {String B = "银行流水B";// B录入银行流水数据String A = exgr.exchange("B");//同步点，交换数据System.out.println("A和B数据是否一致：" + A.equals(B) + "\nA录入的是："+ A + "\nB录入的是：" + B);} catch (InterruptedException e) {}}});threadPool.shutdown();
}
复制代码

运行结果：

A和B数据是否一致：false
A录入的是：银行流水A
B录入的是：银行流水B
复制代码

文章源地址：https://www.cnblogs.com/jinggod/p/8494384.html