JUC-Exchanger总结

1、Exchanger 作用
使两个线程之间进行数据传递。（对是两个之间而不是三个或者更多个线程之间），Exchanger并发辅助类，允许在并发任务之间交换数据。具体来说Exchanger类在两个线程之间定义同步点。当两个线程到达同步点时，它们交换数据结构。需要注意的是Exchanger类只能同步两个线程。内存一致性效果：对于通过Exchanger成功交换对象的每对线程，每个线程中在exchanger()之前的操作 happen-before从另一线程中相应的exchanger()返回的后续操作。
2、常用方法
exchange（） 阻塞当前线程并等待其他线程来取得数据，若没有其他线程来取数据则一直等待。
exchange（） 传递数据
exchange(V v, long timeout, TimeUnit unit) 在指定的时间内没收到消息，则抛出超时的异常。

3、原理

• 内部类Participant继承自ThreadLocal,用来保存线程本地变量Node.
• Node存储用于单槽交换和多槽交换的字段.

单槽位交换(slot exchange)
流程:

• 首先到达的线程:
  • 将slot字段指向自身的Node节点,表示槽位已被占用.
  • 该线程自旋一段时间.若经过一段时间自旋还是没有配对的线程到达,则进入阻塞.(自旋减少上下文切换的开销)
• 后续到达的线程:
  • 此时槽位slot已被占用.则后续的线程将槽位slot清空,取出Node中的item作为交换的数据.
  • 后续的线程把自身的数据存入Node中的match字段中,并唤醒先到达的线程.
• 先到达的线程被唤醒:
  • 检查match是否为空.不为空则退出自旋,将match中的数据返回.

多槽位交换(arena exchange)
触发机制:
在单槽位交换中,若:多个匹配线程竞争修改slot槽位,导致线程CAS修改slot失败,则初始化arena多槽位数组,后续的交换使用多槽位交换.
流程:

• 若槽不为空,则已有线程到达并等待.
  • 获取已到达先携带的数据.
  • 将当前线程携带的数据交换给已到达的线程.
  • 唤醒已到达的线程.
• 若槽位有效且为空.
  • CAS占用槽位成功.
  • 通过spin->yield->block的锁升级方式进行优化的等待其他线程到达.若有线程到达,则交换数据后返回交换后的数据.
  • 若没有等待配对的线程,则阻塞的线程.
• 无效的槽位,需要扩容.
  • 通过CAS方式对数组进行扩容.

注:
数组是连续的内存地址空间.多个slot会被加载到同一个缓存行上。当一个slot改变时,导致该slot所在的缓存行上所有的数据都无效,需要重新从内存加载.

不同版本的差异

• JDK5被设计为容量为1的容器,存放一个等待的线程.当另外一个线程到达时,交换数据后会清空容器.
• JDK6后提供多个slot,增加并发执行的吞吐量.

4、例子

public class ExchangerTester {

    // Exchanger实例.
    private static final Exchanger<String> exchanger = new Exchanger<String>();

    public static void main(String[] args) {
        // 模拟阻塞线程.
        new Thread(() -> {
            try {
                String wares = "红烧肉";
                System.out.println(Thread.currentThread().getName() + "商品方正在等待金钱方，使用货物兑换为金钱.");
                Thread.sleep(2000);
                String money = exchanger.exchange(wares);
                System.out.println(Thread.currentThread().getName() + "商品方使用商品兑换了" + money);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        }).start();

        // 模拟阻塞线程.
        new Thread(() -> {
            try {
                String money = "人民币";
                System.out.println(Thread.currentThread().getName() + "金钱方正在等待商品方，使用金钱购买食物.");
                Thread.sleep(4000);
                String wares = exchanger.exchange(money);
                System.out.println(Thread.currentThread().getName() + "金钱方使用金钱购买了" + wares);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        }).start();
    }

}

输出结果：

Thread-0商品方正在等待金钱方，使用货物兑换为金钱.
Thread-1金钱方正在等待商品方，使用金钱购买食物.
Thread-1金钱方使用金钱购买了红烧肉
Thread-0商品方使用商品兑换了人民币