java-Exchanger详解

1.概述

java.util.concurrent.Exchanger。这在Java中作为两个线程之间交换对象的公共点。

2.Exchanger简介

Exchanger类可用于在两个类型为T的线程之间共享对象。该类仅提供了一个重载的方法exchange(T t)。

当调用exchanger时，它会等待成对的另一个线程也调用它。在这一点上，第二个线程发现第一个线程正在等待其对象。线程交换它们持有的对象并发出交换信号，然后它们可以返回。

让我们看一个例子，以理解两个线程之间使用Exchanger进行消息交换：

 @Test
public void givenThreads_whenMessageExchanged_thenCorrect() {Exchanger<String> exchanger = new Exchanger<>();Runnable taskA = () -> {try {String message = exchanger.exchange("from A");assertEquals("from B", message);} catch (InterruptedException e) {Thread.currentThread().interrupt();throw new RuntimeException(e);}};Runnable taskB = () -> {try {String message = exchanger.exchange("from B");assertEquals("from A", message);} catch (InterruptedException e) {Thread.currentThread().interrupt();throw new RuntimeException(e);}};CompletableFuture.allOf(runAsync(taskA), runAsync(taskB)).join();}

在这里，我们有两个线程使用共同的Exchanger交换彼此之间的消息。让我们看一个例子，在这个例子中，我们从主线程与一个新线程交换对象：

@Test
public void givenThread_WhenExchangedMessage_thenCorrect() throws InterruptedException, ExecutionException {Exchanger<String> exchanger = new Exchanger<>();Runnable runner = () -> {try {String message = exchanger.exchange("from runner");assertEquals("to runner", message);} catch (InterruptedException e) {Thread.currentThread().interrupt();throw new RuntimeException(e);}};CompletableFuture<Void> result = runAsync(runner);String msg = exchanger.exchange("to runner");assertEquals("from runner", msg);result.join();}

请注意，我们需要先启动runner线程，然后在主线程中调用exchange()。

还要注意，如果第二个线程在超时时间内未达到交换点，第一个线程的调用可能会超时。第一个线程应等待多长时间可以使用重载的exchange(T t, long timeout, TimeUnit timeUnit)来控制。

3.无GC数据交换

Exchanger可以用于创建通过一个线程向另一个线程传递数据的管道类型的模式。

   private static final int BUFFER_SIZE = 100;@Testpublic void givenData_whenPassedThrough_thenCorrect() throws InterruptedException, ExecutionException {Exchanger<Queue<String>> readerExchanger = new Exchanger<>();Exchanger<Queue<String>> writerExchanger = new Exchanger<>();int counter = 0;Runnable reader = () -> {Queue<String> readerBuffer = new ConcurrentLinkedQueue<>();while (true) {readerBuffer.add(UUID.randomUUID().toString());if (readerBuffer.size() >= BUFFER_SIZE) {try {readerBuffer = readerExchanger.exchange(readerBuffer);} catch (InterruptedException e) {Thread.currentThread().interrupt();throw new RuntimeException(e);}}}};Runnable processor = () -> {Queue<String> processorBuffer = new ConcurrentLinkedQueue<>();Queue<String> writerBuffer = new ConcurrentLinkedQueue<>();try {processorBuffer = readerExchanger.exchange(processorBuffer);while (true) {writerBuffer.add(processorBuffer.poll());if (processorBuffer.isEmpty()) {try {processorBuffer = readerExchanger.exchange(processorBuffer);writerBuffer = writerExchanger.exchange(writerBuffer);} catch (InterruptedException e) {Thread.currentThread().interrupt();throw new RuntimeException(e);}}}} catch (InterruptedException e) {Thread.currentThread().interrupt();throw new RuntimeException(e);}};Runnable writer = () -> {Queue<String> writerBuffer = new ConcurrentLinkedQueue<>();try {writerBuffer = writerExchanger.exchange(writerBuffer);while (true) {System.out.println(writerBuffer.poll());if (writerBuffer.isEmpty()) {writerBuffer = writerExchanger.exchange(writerBuffer);}}} catch (InterruptedException e) {Thread.currentThread().interrupt();throw new RuntimeException(e);}};CompletableFuture.allOf(runAsync(reader), runAsync(processor), runAsync(writer)).get();}