并发工具类Phaser、Exchanger使用

import java.util.Date;
import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;

import static java.util.concurrent.ThreadLocalRandom.current;
import static java.lang.Thread.currentThread;

/**
 * 将 Phaser当作 CountDownLatch来用
 * 
 **/
public class PhaserForCountDownLatch {
    public static void main(String[] args) throws InterruptedException {
        // 定义一个 Phaser ， 并未指定“分片数量 parties”，此时在 Phaser 内部分片的数量 parties 默认为 0 ，
        // 后面可以通过 register() 方法来动态增加
        final Phaser phaser = new Phaser();
        // 定义 5 个线程
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                // 调用 Phaser 的 register() 方法使得 phaser 内部的 parties 加一
                phaser.register();
                try {
                    // 采用随机休眠的方式模拟线程的运行时间开销
                    TimeUnit.SECONDS.sleep(current().nextInt(20));

                    // 线程任务结束，执行 arrive()
                    /**
                     * 补充：arrive() 方法类似于 CountDownLatch 的 countdown()
                     * 方法，代表着“当前线程已经到达屏障”，
                     * 但是它不需要等待其他的线程也到达屏障。因此该方法“不是阻塞的方法”，执行之后会立即返回，
                     * 同时该方法会返回一个整数类型的数字，代表着已经到达的 Phase(阶段)编号
                     */
                    phaser.arrive();
                    System.out.println(new Date() + ":" + currentThread() + " completed the work.");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, "T-" + i).start();
        }
        /**
         * 这里让线程休眠的目的： 为了保证在主线程 register() 之前，所有的子线程都能顺利 register ， 否则就会出现
         * phaser 只注册一个 parties ， 并且很快 arrive 的情况。
         */
        TimeUnit.SECONDS.sleep(current().nextInt(10));
        // 主线程也调用注册方法
        phaser.register();

        // 主线程也 arrive() ， 但是它要等待下一个阶段，等待下一个阶段的前提“所有的线程都 arrive ,
        // 也就是 phaser 内部当前 phase 的 unarrived 数量为 0 ”
        phaser.arriveAndAwaitAdvance();
        System.out.println(new Date() + ": all of sub task completed work.");
    }
}

/**
 * 将 Phaser 当作 CyclicBarrier 来使用
 *
 *  phaser.arriveAndAwaitAdvance(): 该方法会等待当前 Phaser 中所有的 part（子线程）都完成了
 *                                  任务才能使线程退出阻塞状态

 **/
public class PhaserForCyclicBarrier
{
   public static void main(String[] args) throws InterruptedException
   {
       // 定义一个分片 parties 为0 的 Phaser
      final Phaser phaser = new Phaser();
      for (int i = 0; i < 5; i++)
      {
         new Thread(()->
         {
            // 子线程调用注册方法
            phaser.register();
            try
            {
               TimeUnit.SECONDS.sleep(current().nextInt(20));
               // 调用 arriveAndAwaitAdvance() 等待所有线程 arrive 然后继续前行
               phaser.arriveAndAwaitAdvance();
               System.out.println(new Date() + ":" + currentThread() + " completed the work.");

            }
            catch (InterruptedException e)
            {
               e.printStackTrace();
            }
         } , "T-"+i).start();
      }
      // 休眠以确保其他子线程顺利调用 register()
      TimeUnit.SECONDS.sleep(10);
      // 主线程调用 register()
      phaser.register();
      phaser.arriveAndAwaitAdvance();
      System.out.println(new Date() + ": all of sub task completed work.");
   }
}

import java.util.concurrent.Exchanger;

public class ExechangerExample {

    public static void main(String[] args) {

        Exchanger exchanger = new Exchanger<>();
        new Thread() {
            @Override
            public void run() {
                String data1 = "data1";
                try {
                    System.out.println(Thread.currentThread().getName() + "交换前的数据:" + data1);
                    String data2 = exchanger.exchange(data1);
                    System.out.println(Thread.currentThread().getName() + "交换后的数据:" + data2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }.start();
        new Thread() {
            @Override
            public void run() {
                String data2 = "data2";
                try {
                    System.out.println(Thread.currentThread().getName() + "交换前的数据:" + data2);
                    String data1 = exchanger.exchange(data2);
                    System.out.println(Thread.currentThread().getName() + "交换后的数据:" + data1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }.start();
    }
}