深入剖析Java并发库：Exchanger的工作原理与应用场景

public class Exchanger<V> {
    // 用于等待交换的线程队列，这里简化为一个节点
    private volatile Node<V> slot;
    // ... 其他字段和方法 ...

    public Exchanger() {
        slot = null;
    }

    public V exchange(V x) throws InterruptedException {
        // 创建一个新的节点，包含当前线程和要交换的数据
        Node<V> newNode = new Node<V>(Thread.currentThread(), x);
        // 自旋，直到交换成功
        for (;;) {
            Node<V> oldNode = slot;
            if (oldNode == null) {
                // 如果slot为空，则尝试将新节点设置为slot
                if (compareAndSetSlot(null, newNode)) {
                    // 等待另一个线程来交换数据
                    waitForMatch(newNode);
                    return newNode.item; // 返回交换后的数据
                }
            } else if (oldNode.isWaiting()) {
                // 如果slot中的节点正在等待交换，则尝试进行交换
                if (compareAndSetSlot(oldNode, null)) {
                    // 设置新节点的匹配线程和数据
                    newNode.match = oldNode;
                    // 唤醒等待的线程
                    oldNode.thread.interrupt();
                    return oldNode.item; // 返回交换后的数据
                }
            } else {
                // 如果slot中的节点已经匹配完成，则需要等待直到它变为null或正在等待状态
                // 这里可以加入自旋等待的逻辑，但为了简化，我们省略了这部分代码
            }
            // 如果发生竞争失败或其他情况，这里会有重试的逻辑，同样为了简化而省略
        }
    }

    // 内部节点类，用于存储线程和要交换的数据
    private static final class Node<V> {
        final Thread thread; // 存储线程的引用
        V item; // 存储要交换的数据
        Node<V> match; // 存储匹配的节点，即交换数据的对方节点
        // ... 构造方法和其他方法 ...

        boolean isWaiting() {
            return match == null;
        }
    }

    // 使用CAS操作设置slot的值，这里简化为一个方法调用，实际实现会使用Unsafe类进行操作
    private boolean compareAndSetSlot(Node<V> expect, Node<V> update) {
        // ... CAS操作的实现 ...
        return true; // 这里为了简化而直接返回true，实际实现会有比较和设置的逻辑
    }

    // 等待匹配线程的方法，这里简化为一个方法调用，实际实现会有等待和唤醒的逻辑处理
    private void waitForMatch(Node<V> newNode) throws InterruptedException {
        // ... 等待和唤醒的逻辑处理 ...
    }
}

import java.util.concurrent.Exchanger;

public class ExchangerExample {

    public static void main(String[] args) {
        // 创建一个Exchanger对象
        Exchanger<String> exchanger = new Exchanger<>();

        // 创建一个线程，它将使用"Hello"与另一个线程交换数据
        Thread producer = new Thread(() -> {
            try {
                String producedData = "Hello";
                String consumerData = exchanger.exchange(producedData);
                System.out.println("生产者线程交换后得到的数据: " + consumerData);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "生产者线程");

        // 创建一个线程，它将使用"World"与另一个线程交换数据
        Thread consumer = new Thread(() -> {
            try {
                String consumerData = "World";
                String producedData = exchanger.exchange(consumerData);
                System.out.println("消费者线程交换后得到的数据: " + producedData);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "消费者线程");

        // 启动线程
        producer.start();
        consumer.start();
    }
}

生产者线程交换后得到的数据: World
消费者线程交换后得到的数据: Hello