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;

    @Test
    public 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();
    }

在这里，我们有三个线程：reader、processor和writer。它们共同作为一个单一的管道，在它们之间交换数据。

readerExchanger在reader和processor线程之间共享，而writerExchanger在processor和writer线程之间共享。

请注意，此处的示例仅用于演示。在创建无限循环时务必小心while(true)。另外，为保持代码的可读性，我们省略了一些异常处理。

通过重用缓冲区来交换数据的这种模式允许减少垃圾回收。exchange方法返回相同的队列实例，因此这些对象不会被垃圾回收。与任何阻塞队列不同，Exchanger不会创建任何用于保存和共享数据的节点或对象。

创建这样的管道类似于Disruptor模式，其中一个关键区别是，Disruptor模式支持多个生产者和消费者，而Exchanger可以在一对生产者和消费者之间使用。

4.总结

因此，Java中的Exchanger是什么，它是如何工作的，我们看到了如何使用Exchanger类。此外，我们创建了一个管道，并演示了线程之间无GC的数据交换。