thread_Exchanger数据交换

Posted 2020-08-05 上善若水

 Exchanger 是一个同步辅助类，用于两个并发线程之间在一个同步点进行数据交换。
 允许两个线程在某一个点进行数据交换。
 可以视作双向的同步队列；
 可应用于基因算法、流水线设计等场景

Exchanger提供了 一个同步点 ， 在这个同步点，两个线程可以交换数据 。每个线程通过exchange()方法的入口提供数据给另外的线程，并接收其它线程提供的数据，并返回。 

public class Exchanger1Test {

    // 场景描述：一对一的 生产者和消费者，生产者每次生产5个商品，然后消费者把空的商品容器和生产者交换。
    // 生产者线程一定要先生产数据，再交换数据，消费者线程一定要先交换数据，再消费数据，否则会出现少消费数据的现象
    // 允许原子性的交换两个（多个）对象，但同时只有一对才会成功
    // exchange方法真的帮一对线程交换了数据；
    // exchange方法真的会阻塞调用方线程直至另一方线程参与交易。
    public static void main(String[] args) {

        // Exchanger可以在两个线程之间交换数据，只能是2个线程，他不支持更多的线程之间互换数据。
        // 两个线程必须使用同一个Exchanger对象，且只能是两个线程间的数据交换
        // 当线程A调用Exchange对象的exchange()方法后，他会陷入阻塞状态，直到线程B也调用了exchange()方法，然后以线程安全的方式交换数据，之后线程A和B继续运行
        Exchanger<ArrayList<String>> exchanger = new Exchanger<ArrayList<String>>();
        ExecutorService exec = Executors.newCachedThreadPool();
 
        exec.execute(new Producer(exchanger));
        exec.execute(new Consumer(exchanger));
        exec.shutdown();
        try {
            exec.awaitTermination(30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            // log.error(e, e);
        }    
    }
}

// 生产者
class Producer implements Runnable {
    private ArrayList<String> goods = new ArrayList<String>(); // 商品容器
    private Exchanger<ArrayList<String>> exchanger = new Exchanger<ArrayList<String>>();
    //控制交易双方线程的退出
    private static AtomicBoolean isDone = new AtomicBoolean(true);
    public Producer(Exchanger<ArrayList<String>> exchanger) {
        this.exchanger = exchanger;
    }

    @Override
    public void run() {
        while (!Thread.interrupted() &&  isDone.get()) {//
            
            for (int i = 0; i < 3; i++) { // 生产3次
                System.out.println("------------------------生产者生产第 " + i + "次");
                for (int j = 0; j < 3; j++) { // 每次生产3个商品
                    String e = (long) (Math.random() * 1000) + "";
                    goods.add(e);
                    System.out.println("生产了商品：" + e);
                }
                try {
                    // 生产者线程一定要先生产数据，再交换数据，消费者线程一定要先交换数据，再消费数据
                    // exchanger.exchange(ｖ)的时候，当前线程会被阻塞，直到另一个线程执行该方法，同时完成数据的交换
                    goods = exchanger.exchange(goods); // 交换数据
                    System.out.println("生产者：数据交换完毕：获得交换的商品容器大小：" + goods.size());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

// 消费者
class Consumer implements Runnable {
    private ArrayList<String> goods = new ArrayList<String>(); // 商品容器
    private Exchanger<ArrayList<String>> exchanger = new Exchanger<ArrayList<String>>();
    private static AtomicBoolean isDone = new AtomicBoolean(true);
    public Consumer(Exchanger<ArrayList<String>> exchanger) {
        this.exchanger = exchanger;
    }

    @Override
    public void run() {
        while (!Thread.interrupted() && isDone.get()) {//&& !isDone
            for (int i = 0; i < 3; i++) { // 消费3次
                try {
                    // 生产者线程一定要先生产数据，再交换数据，消费者线程一定要先交换数据，再消费数据
                    // exchanger.exchange(ｖ)的时候，当前线程会被阻塞，直到另一个线程执行该方法，同时完成数据的交换
                    goods = exchanger.exchange(goods); // 交换数据
                    System.out.println("消费者：数据交换完毕：获得交换的商品容器大小：" + goods.size());
                    
                    // 消费商品
                    Iterator<String> it = goods.iterator();
                    if (goods.size() > 0) {
                        System.out.println("*********************消费者消费第 " + i + "次");
                        while (it.hasNext()) {
                            String next = it.next();
                            System.out.println("消费了商品：" + next);
                            it.remove(); // 移除消费了的商品
                        }
                    }
                    
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            
        }
    }
}