并发之Exchanger基础(一)

并发工具类Exchanger的API是非常简单的，它的功能就是在两个线程之间传递数据。

在API中式这样介绍的，可以在 对中对元素中进行配对和交换线程的同步点、每个线程将条目上的某个方法呈现给交换方法，与伙伴线程进行匹配，并且在返回时接收其伙伴的对象。 Exchance可以被视为Synchronous的双向形式。Exchanger在管道设计中很有用。

器，它允许在并发任务之间交换数据。具体来说，器类允许在两个线程之间定义同步点。当两个线程都到达同步点时，他们交换数据结构，因此第一个线程的数据结构进入到第二个线程中，第二个线程的数据结构进入到第一个线程中。

源码中的注释有一句:
Overview: The core algorithm is, for an exchange "slot",
* and a participant (caller) with an item:
器算法的核心是通过一个可交换数据的插槽，以及一个可以带有数据项目的参与者源码中的描述如下：
for (;;) {
* if (slot is empty) { // offer
* place item in a Node;
* if (can CAS slot from empty to node) {
* wait for release;
* return matching item in node;
* }
* }
* else if (can CAS slot from node to empty) { // release
* get the item in node;
* set matching item in node;
* release waiting thread;
* }
* // else retry on CAS failure

然后我们来看下，exchange这个方法。
这个方法具有堵塞的特性，也就是这个方法调用后等待其他线程来渠道数据，如果没有其他线程取得数据，则会一直等待。。。

public V exchange(V x, long timeout, TimeUnit unit) 在指定时间内，没有其他线程取出数据，则出现超时异常。

实现代码如下:
线程A如下,这个类主要是实现交换其他线程的数据，等待线程B

public class ThreadA extends Thread{

    private Exchanger<String> exchanger=new Exchanger();

    public ThreadA(Exchanger<String> exchanger){
        this.exchanger=exchanger;
    }

    @Override
    public void run() {
        try {
            System.out.println(Thread.currentThread().getName()+"开始运行，获取线程B的值。。。"+exchanger.exchange("你好，我是线程A"));
            System.out.println("thread A end");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

线程B，跟A的作用类似

public class ThreadB extends Thread{

    private Exchanger<String> exchanger=new Exchanger();

    public ThreadB(Exchanger<String> exchanger){
        this.exchanger=exchanger;
    }

    @Override
    public void run() {
//        super.run();
        try {
            System.out.println(Thread.currentThread().getName()+"开始运行，获取线程A的数据。。。"+exchanger.exchange("我叫线程B"));
            System.out.println("thread B end");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

测试类代码如下:

public class Run {

    public static void main(String[] args) {
        //测试线程A获取线程B没有得到交换数据时，会堵塞
        Exchanger<String> exchanger = new Exchanger<>();
      //  new ThreadA(exchanger).start();  //发现会堵塞
        ThreadA threadA = new ThreadA(exchanger);
        ThreadB threadB = new ThreadB(exchanger);
        threadA.setName("线程A");
        threadB.setName("线程B");
        threadA.start();
        threadB.start();
    }
}

image.png

然后今天是先更新理解性的应用。后续更新一下底层的实现怎么实现的，很有趣。

感谢您阅读我的文章，整理不易，帮我点个赞，谢谢哈。
如果对文章部分还有什么见解或者疑惑，可以私信评论我，欢迎技术讨论。可以加我微信z985085305，获取我整理的一些其他的笔记。

思想的碰撞最能促进技术的进步哦。