java多线程之线程通信工具类-阿里云开发者社区

一、概述

在JDK的并发包里提供了几个非常有用的并发工具类。CountDownLatch、CyclicBarrier和 Semaphore工具类提供了一种并发流程控制的手段，Exchanger工具类则提供了在线程间交换数据的一种手段。

它们都在java.util.concurrent包下。先总体概括⼀下都有哪些⼯具类，它们有什么作⽤，然后再分别介绍它们的主要使⽤⽅法和原理

锁作为线程通信的一种方式，本质上锁的目的是保护一种受限资源，保证同一时刻只有一个线程能访问（ReentrantLock），或者只有一个线程能写入（ReadWriteLock）。

还有一种受限资源，它需要保证同一时刻最多有N个线程能访问，比如同一时刻最多创建100个数据库连接，最多允许10个用户下载等。

这种限制数量的锁，如果用Lock数组来实现，就太麻烦了。这种情况就可以使用Semaphore。

Semaphore 信号量维护了一个许可集,每次使用时执行acquire()从Semaphore获取许可，如果没有则会阻塞，每次使用完执行release()释放许可。

Semaphore可以用于做流量控制，特别是公用资源有限的应用场景，比如数据库连接。假如有一个需求，要读取几万个文件的数据，因为都是IO密集型任务，我们可以启动几十个线程并发地读取，但是如果读到内存后，还需要存储到数据库中，而数据库的连接数只有10个，这时我们必须控制只有10个线程同时获取数据库连接保存数据，否则会报错无法获取数据库连接。这个时候，就可以使用Semaphore来做流量控制

案例1：最多允许3个线程同时访问：

public class SemaphoreDemo {
    static class MyThread implements Runnable {
        private int value;
        private Semaphore semaphore;
 
        public MyThread(int value, Semaphore semaphore) {
            this.value = value;
            this.semaphore = semaphore;
        }
 
        @Override
        public void run() {
            try {
                semaphore.acquire(); // 获取permit
                System.out.println("当前线程" + value+"还剩资源数" + semaphore.availablePermits()+"还有" + semaphore.getQueueLength());
                // 睡眠随机时间，打乱释放顺序
                Random random = new Random();
                Thread.sleep(random.nextInt(1000));
                semaphore.release(); // 释放permit
                System.out.println(String.format("线程%d释放了资源", value));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
 
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < 10; i++) {
            new Thread(new MyThread(i, semaphore)).start();
        }
    }
}

案例2：结合线程池访问

public class SemaphoreTest {
    private static final int THREAD_COUNT = 30;
    private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
    private static Semaphore s = new Semaphore(10);
    public static void main(String[] args) {
        for (int i = 0; i< THREAD_COUNT; i++) {
            threadPool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        s.acquire();
                        System.out.println(Thread.currentThread().getName()+"saving data");
                        Thread.sleep(2000L);
                        s.release();
                    } catch (InterruptedException e) {
                    }
                }
            });
        }
        threadPool.shutdown();
    }
}

在代码中，虽然有30个线程在执行，但是只允许10个并发执行。Semaphore的构造方法 Semaphore（int permits）接受一个整型的数字，表示可用的许可证数量。Semaphore（10）表示允许10个线程获取许可证，也就是最大并发数是10。Semaphore的用法也很简单，首先线程使用 Semaphore的acquire()方法获取一个许可证，使用完之后调用release()方法归还许可证。还可以用tryAcquire()方法尝试获取许可证.

Semaphore本质上就是一个信号计数器，用于限制同一时间的最大访问数量。最主要的⽅法是acquire⽅法和release⽅法。acquire()⽅法会申请⼀个permit，⽽ release⽅法会释放⼀个permit。当然，你也可以申请多个acquire(int permits)或者释放多个release(int permits)。每次acquire，permits就会减少⼀个或者多个。如果减少到了0，再有其他线程来 acquire，那就要阻塞这个线程直到有其它线程release permit为⽌。

其他方法 Semaphore还提供一些其他方法，具体如下。

intavailablePermits()：返回此信号量中当前可用的许可证数。
intgetQueueLength()：返回正在等待获取许可证的线程数。
booleanhasQueuedThreads()：是否有线程正在等待获取许可证。
void reducePermits（int reduction）：减少reduction个许可证，是个protected方法。
Collection getQueuedThreads()：返回所有等待获取许可证的线程集合，是个protected方法。

Semaphore实现原理：内部有⼀个继承了AQS的同步器Sync，重写了 tryAcquireShared ⽅法。在这个⽅法⾥，会去尝试获取资源。如果获取失败（想要的资源数量⼩于⽬前已有的资源数量），就会返回⼀个负数（代表尝试获取资源失败）。然后当前线程就会进⼊AQS的等待队列

三、线程间交换数据的Exchanger

Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。这两个线程通过 exchange方法交换数据，如果第一个线程先执行exchange()方法，它会一直等待第二个线程也执行exchange方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。它⽀持泛型，也就是说你可以在两个线程之间传送任何数据。先来⼀个案例看看如何使⽤，⽐如两个线程之间想要传送字符串：

使用场景：两个线程相互等待处理结果并进行数据传递。

    public static void main(String[] args) throws InterruptedException {
        Exchanger<String> exchanger = new Exchanger<>();
        new Thread(() -> {
            try {
                System.out.println("这是线程A，得到了另⼀个线程的数据："
                        + exchanger.exchange("这是来⾃线程A的数据"));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        System.out.println("这个时候线程A是阻塞的，在等待线程B的数据");
        Thread.sleep(1000);
        new Thread(() -> {
            try {
                System.out.println("这是线程B，得到了另⼀个线程的数据："
                        + exchanger.exchange("这是来⾃线程B的数据"));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }

输出

这个时候线程A是阻塞的，在等待线程B的数据
这是线程B，得到了另⼀个线程的数据：这是来⾃线程A的数据
这是线程A，得到了另⼀个线程的数据：这是来⾃线程B的数据

可以看到，当⼀个线程调⽤exchange⽅法后，它是处于阻塞状态的，只有当另⼀个线程也调⽤了exchange⽅法，它才会继续向下执⾏。看源码可以发现它是使⽤ park/unpark来实现等待状态的切换的，但是在使⽤park/unpark⽅法之前，使⽤了 CAS检查，估计是为了提⾼性能。 Exchanger⼀般⽤于两个线程之间更⽅便地在内存中交换数据，因为其⽀持泛型，所以我们可以传输任何的数据，⽐如IO流或者IO缓存。根据JDK⾥⾯的注释的说法，可以总结为⼀下特性：

此类提供对外的操作是同步的；
⽤于成对出现的线程之间交换数据；
可以视作双向的同步队列；
可应⽤于基因算法、流⽔线设计等场景。

Exchanger可以用于遗传算法，遗传算法里需要选出两个人作为交配对象，这时候会交换两人的数据，并使用交叉规则得出2个交配结果。Exchanger也可以用于校对工作，比如我们需要将纸制银行流水通过人工的方式录入成电子银行流水，为了避免错误，采用AB岗两人进行录入，录入到Excel之后，系统需要加载这两个Excel，并对两个Excel数据进行校对，看看是否录入一致

public class ExchangerTest {
    private static final Exchanger<String>exgr = new Exchanger<String>();
    private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
    public static void main(String[] args) {
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    String A = "银行流水A"; // A录入银行流水数据
                    exgr.exchange(A);
                } catch (InterruptedException e) {
                }
            }
        });
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    String B = "银行流水B"; // B录入银行流水数据
                    String A = exgr.exchange("B");
                    System.out.println("A和B数据是否一致：" + A.equals(B) + "，A录入的是："
                            + A + "，B录入是：" + B);
                } catch (InterruptedException e) {
                }
            }
        });
        threadPool.shutdown();
    }
}

Exchanger只能是两个线程交换数据吗？那三个调⽤同⼀个实例的 exchange⽅法会发⽣什么呢？答案是只有前两个线程会交换数据，第三个线程会进⼊阻塞状态。需要注意的是，exchange是可以重复使⽤的。也就是说。两个线程可以使⽤ Exchanger在内存中不断地再交换数据。

如果两个线程有一个没有执行exchange()方法，则会一直等待，如果担心有特殊情况发生，避免一直等待，可以使用exchange（V x，longtimeout，TimeUnit unit）设置最大等待时长

四、等待多线程完成的CountDownLatch

CountDownLatch允许一个或多个线程等待其他线程完成操作。假如有这样一个需求：我们需要解析一个Excel里多个sheet的数据，此时可以考虑使用多线程，每个线程解析一个sheet里的数据，等到所有的sheet都解析完之后，程序需要提示解析完成。在这个需求中，要实现主线程等待所有线程完成sheet的解析操作，最简单的做法是使用 join()方法

public class JoinCountDownLatchTest {
    public static void main(String[] args) throws InterruptedException {
        Thread parser1 = new Thread(new Runnable() {
            @Override
            public void run() {
            }
        });
        Thread parser2 = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("parser2 finish");
            }
        });
        parser1.start();
        parser2.start();
        parser1.join();
        parser2.join();
        System.out.println("all parser finish");
    }
}

join用于让当前执行线程等待join线程执行结束。其实现原理是不停检查join线程是否存活，如果join线程存活则让当前线程永远等待。其中，wait（0）表示永远等待下去

直到join线程中止后，线程的this.notifyAll()方法会被调用，调用notifyAll()方法是在JVM里实现的，所以在JDK里看不到

在JDK 1.5之后的并发包中提供的CountDownLatch也可以实现join的功能，并且比join的功能更多

 
public class CountDownLatchTest {
    static CountDownLatch c = new CountDownLatch(2);
    public static void main(String[] args) throws InterruptedException {
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println(1);
                c.countDown();
                System.out.println(2);
                c.countDown();
            }
        }).start();
        c.await();
        System.out.println("3");
    }
}

CountDownLatch的构造函数接收一个int类型的参数作为计数器，如果你想等待N个点完成，这里就传入N。

当我们调用CountDownLatch的countDown方法时，N就会减1，CountDownLatch的await方法会阻塞当前线程，直到N变成零。由于countDown方法可以用在任何地方，所以这里说的N个点，可以是N个线程，也可以是1个线程里的N个执行步骤。用在多个线程时，只需要把这个 CountDownLatch的引用传递到线程里即可。

如果有某个解析sheet的线程处理得比较慢，我们不可能让主线程一直等待，所以可以使用另外一个带指定时间的await方法——await（long time，TimeUnit unit），这个方法等待特定时间后，就会不再阻塞当前线程。join也有类似的方法

注意计数器必须大于等于0，只是等于0时候，计数器就是零，调用await方法时不会阻塞当前线程。CountDownLatch不可能重新初始化或者修改CountDownLatch对象的内部计数器的值。一个线程调用countDown方法happen-before，另外一个线程调用await方法。

CountDownLatch的⽅法也很简单，如下：

// 构造⽅法：
public CountDownLatch(int count)
public void await() // 等待
public boolean await(long timeout, TimeUnit unit) // 超时等待
public void countDown() // count - 1
public long getCount() // 获取当前还有多少count

案例1：在某些业务场景中，程序执行需要等待某个条件完成后才能继续执行后续的操作。典型的应用为并行计算：当某个处理的运算量很大时，可以将该运算任务拆分成多个子任务，等待所有的子任务都完成之后，父任务再拿到所有子任务的运算结果进行汇总

调用ExecutorService类的shutdown()方法，并不会第一时间内把所有线程全部都销毁掉，而是让当前已有的线程全部执行完，之后，再把线程池销毁掉

代码如下

public class CountDownLatchExample {
    private static final int threadCount = 200;
 
    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newCachedThreadPool();
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        System.out.println(">>>finish");
        exec.shutdown();
    }
 
    private static void test(int threadNum) throws InterruptedException {
        Thread.sleep(100);
        System.out.println(threadNum);
        Thread.sleep(100);
    }
}

支持给定时间等待的示例代码如下

public class CountDownLatchExample {
    private static final int threadCount = 200;
    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newCachedThreadPool();
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await(10, TimeUnit.MICROSECONDS);
        System.out.println(">>> finish");
        exec.shutdown();
    }
    private static void test(int threadNum) throws InterruptedException {
        Thread.sleep(100);
        System.out.println(threadNum);
    }
}

案例2：6 个同学陆续离开教室后值班同学才可以关门

public class CountDownLatchDemo {
    public static void main(String[] args) throws Exception {
        //定义一个数值为 6 的计数器
        CountDownLatch countDownLatch = new CountDownLatch(6);
        //创建 6 个同学
        for (int i = 1; i <= 6; i++) {
            new Thread(() -> {
                try {
                    if (Thread.currentThread().getName().equals("stu 6")) {
                        Thread.sleep(2000);
                    }
                    System.out.println(Thread.currentThread().getName() + "out");//计数器减一,不会阻塞
                    countDownLatch.countDown();
                } catch (Exception
                        e) {
                    e.printStackTrace();
                }
            }, "stu" + i).start();
        }
        //主线程 await 休息
        System.out.println("main thread ......");
        countDownLatch.await();
        //全部离开后自动唤醒主线程
        System.out.println("finish current count" + countDownLatch.getCount());
    }
}

案例3：玩游戏的时候，在游戏真正开始之前，⼀般会等待⼀些前置任务完成，⽐如“加载地图数据”，“加载⼈物模型”，“加载背景⾳乐”等等。只有当所有的东⻄都加载完成后，玩家才能真正进⼊游戏。下⾯我们就来模拟⼀下这个demo

public class CountDownLatchDemo {
    // 定义前置任务线程
    static class PreTaskThread implements Runnable {
        private String task;
        private CountDownLatch countDownLatch;
 
        public PreTaskThread(String task, CountDownLatch countDownLatch) {
            this.task = task;
            this.countDownLatch = countDownLatch;
        }
 
        @Override
        public void run() {
            try {
                Random random = new Random();
                Thread.sleep(random.nextInt(1000));
                System.out.println(task + " - 完成任务");
                countDownLatch.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
 
    public static void main(String[] args) {
        // 假设有三个模块需要加载
        CountDownLatch countDownLatch = new CountDownLatch(3);
        // 主任务
        new Thread(() -> {
            try {
                System.out.println("等待数据加载...");
                System.out.println(String.format("还有%d个前置任务", countDownLatch.getCount()));
                countDownLatch.await();
                System.out.println("数据加载完成正式开始任务！");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        // 前置任务
        new Thread(new PreTaskThread("加载地图数据", countDownLatch)).start();
        new Thread(new PreTaskThread("加载人物模型", countDownLatch)).start();
        new Thread(new PreTaskThread("加载背景音乐", countDownLatch)).start();
    }
}

其实CountDownLatch类的原理挺简单的，内部同样是⼀个基层了AQS的实现类 Sync，且实现起来还很简单，可能是JDK⾥⾯AQS的⼦类中最简单的实现了，有兴趣的读者可以去看看这个内部类的源码。需要注意的是构造器中的计数值（count）实际上就是闭锁需要等待的线程数量。这个值只能被设置⼀次，⽽且CountDownLatch没有提供任何机制去重新设置这个计数值。

1、CyclicBarrirer概述

CyclicBarrirer从名字上来理解是“循环的屏障”的意思。前⾯提到了 CountDownLatch⼀旦计数值 count 被降为0后，就不能再重新设置了，它只能起⼀次“屏障”的作⽤。⽽CyclicBarrier拥有CountDownLatch的所有功能，还可以使⽤ reset() ⽅法重置屏障。

它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行

CyclicBarrier默认的构造方法是CyclicBarrier（int parties），其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞

2、CountDownLatch和CyclicBarrier区别

CountDownLatch和CyclicBarrier都是juc下的并发工具类，二者功能在处理某些事情下看着很相似：都是阻塞线程，但是如果细品和查看源码的话会发现二者之间还是有区别的：

CountDownLatch的作用很简单，就是一个或者一组线程在开始执行操作之前，必须要等到其他线程执行完才可以。我们举一个例子来说明，在考试的时候，老师必须要等到所有人交了试卷才可以走。此时老师就相当于等待线程，而学生就好比是执行的线程。CyclicBarrier同样是等待其他线程都完成了，才可以进行下一步操作，我们再举一个例子，在打王者的时候，在开局前所有人都必须要加载到100%才可以进入，否则所有玩家都相互等待。

都是使用计数器实现，当某个线程调用了CyclicBarrier的await()方法后，该线程就进入了等待状态，而且计数器执行加1操作，当计数器的值达到了设置的初始值，调用await()方法进入等待状态的线程会被唤醒，继续执行各自后续的操作。CyclicBarrier在释放等待线程后可以重用，所以，CyclicBarrier又被称为循环屏障

CountDownLatch主要是阻塞主线程，等待多线程执行完成之后再执行主线程await之后的代码片段，侧重点是主线程等待子线程（多线程）完成之后被唤醒。

CyclicBarrier主要是每个多线程在某一刻阻塞，然后各个多线程之间相互等待，直到最后一个多线程被阻塞，然后冲破栅栏，各自执行自己await()之后的代码段（另一个写法是用两个参数的构造去共完成另一个Runnable任务），侧重点是多线程之间的相互等待。

CountDownLatch是主线程被多线程阻塞，直到多线程执行完成才被唤醒继续执行，所以更关注主线程等待多线程执行完成再继续执行的场景；

CyclicBarrier是多线程各自被阻塞在栅栏前，是多线程之间的相互等待，直到全部的多线程全部执行完成，然后并发的去共同做某件事，比如：赛跑比赛的时候，需要等所有运动员都准备完成之后，才能开始进行比赛。

另外，CountDownLatch是一次性，而CyclicBarrier是可重复利用的（查看源码可以发现当最后一道栅栏被冲破之后，如果还需要用到的话会重新new Generation栅栏对象）

3、CyclicBarrier Barrier被破坏

CyclicBarrier还提供其他有用的方法，比如getNumberWaiting方法可以获得Cyclic-Barrier 阻塞的线程数量，如果在参与者（线程）在等待的过程中，Barrier被破坏，就会抛出 BrokenBarrierException。可以⽤ isBroken() ⽅法检测Barrier是否被破坏。

1. 如果有线程已经处于等待状态，调⽤reset⽅法会导致已经在等待的线程出现 BrokenBarrierException异常。并且由于出现了BrokenBarrierException，将会导致始终⽆法等待。

2. 如果在等待的过程中，线程被中断，也会抛出BrokenBarrierException异常，并且这个异常会传播到其他所有的线程。

3. 如果在执⾏屏障操作过程中发⽣异常，则该异常将传播到当前线程中，其他线程会抛出BrokenBarrierException，屏障被损坏。

4. 如果超出指定的等待时间，当前线程会抛出 TimeoutException 异常，其他线程会抛出BrokenBarrierException异常。

4、CyclicBarrirer案例

案例1：

 
public class CyclicBarrierTest {
    static CyclicBarrier c = new CyclicBarrier(2);
 
    public static void main(String[] args) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    c.await();
                } catch (Exception e) {
                }
                System.out.println(1);
            }
        }).start();
        try {
            c.await();
        } catch (Exception e) {
        }
        System.out.println(2);
    }
}

因为主线程和子线程的调度是由CPU决定的，两个线程都有可能先执行，所以会产生两种输出2 1或者1 2

如果把new CyclicBarrier(2)修改成new CyclicBarrier(3)，则主线程和子线程会永远等待，因为没有第三个线程执行await方法，即没有第三个线程到达屏障，所以之前到达屏障的两个线程都不会继续执行。 CyclicBarrier还提供一个更高级的构造函数CyclicBarrier（int parties，Runnable barrierAction），用于在线程到达屏障时，优先执行barrierAction，方便处理更复杂的业务场景

public class CyclicBarrierTest2{
    static CyclicBarrier c = new CyclicBarrier(2, new A());
 
    public static void main(String[] args) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    c.await();
                } catch (Exception e) {
                }
                System.out.println(1);
            }
        }).start();
        try {
            c.await();
        } catch (Exception e) {
        }
        System.out.println(2);
    }
 
    static class A implements Runnable {
        @Override
        public void run() {
            System.out.println(3);
        }
    }
}

因为CyclicBarrier设置了拦截线程的数量是2，所以必须等代码中的第一个线程和线程A 都执行完之后，才会继续执行主线程，然后输出2

案例2：如果玩⼀个游戏有多个“关卡”，那使⽤ CountDownLatch显然不太合适，那需要为每个关卡都创建⼀个实例。那我们可以使⽤CyclicBarrier来实现每个关卡的数据加载等待功能

public class CyclicBarrierDemo{
    static class PreTaskThread implements Runnable {
        private String task;
        private CyclicBarrier cyclicBarrier;
        public PreTaskThread(String task, CyclicBarrier cyclicBarrier) {
            this.task = task;
            this.cyclicBarrier = cyclicBarrier;
        }
        @Override
        public void run() {
            // 假设总共三个关卡
            for (int i = 1; i < 4; i++) {
                try {
                    Random random = new Random();
                    Thread.sleep(random.nextInt(1000));
                    System.out.println(String.format("关卡%d的任务%s完成", i, task));
                            cyclicBarrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
                cyclicBarrier.reset(); // 重置屏障
            }
        }
    }
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {
            System.out.println("本关卡所有前置任务完成，开始游戏...");
        });
        new Thread(new PreTaskThread("加载地图数据", cyclicBarrier)).start();
        new Thread(new PreTaskThread("加载⼈物模型", cyclicBarrier)).start();
        new Thread(new PreTaskThread("加载背景⾳乐", cyclicBarrier)).start();
    }
}

关卡1的任务加载地图数据完成
关卡1的任务加载背景⾳乐完成
关卡1的任务加载⼈物模型完成
本关卡所有前置任务完成，开始游戏...
关卡2的任务加载地图数据完成
关卡2的任务加载背景⾳乐完成
关卡2的任务加载⼈物模型完成
本关卡所有前置任务完成，开始游戏...
关卡3的任务加载⼈物模型完成
关卡3的任务加载地图数据完成
关卡3的任务加载背景⾳乐完成
本关卡所有前置任务完成，开始游戏...

注意这⾥跟CountDownLatch的代码有⼀些不同。CyclicBarrier没有分为 await() 和 countDown() ，⽽是只有单独的⼀个 await() ⽅法。

⼀旦调⽤await()⽅法的线程数量等于构造⽅法中传⼊的任务总量（这⾥是3），就代表达到屏障了。CyclicBarrier允许我们在达到屏障的时候可以执⾏⼀个任务，可以在构造⽅法传⼊⼀个Runnable类型的对象。上述案例就是在达到屏障时，输出 “本关卡所有前置任务完成，开始游戏...”。

// 构造⽅法
public CyclicBarrier(int parties) {
 this(parties, null);
}
public CyclicBarrier(int parties, Runnable barrierAction) {
 // 具体实现
}

案例3：CyclicBarrier可以用于多线程计算数据，最后合并计算结果的场景。例如，用一个Excel保存了用户所有银行流水，每个Sheet保存一个账户近一年的每笔银行流水，现在需要统计用户的日均银行流水，先用多线程处理每个sheet里的银行流水，都执行完之后，得到每个sheet的日均银行流水，最后，再用barrierAction用这些线程的计算结果，计算出整个Excel的日均银行流水

public class BankWaterService implements Runnable {
    /**
     * 创建4个屏障，处理完之后执行当前类的run方法
     */
    private CyclicBarrier c = new CyclicBarrier(4, this);
    /**
     * 假设只有4个sheet，所以只启动4个线程
     */
    private Executor executor = Executors.newFixedThreadPool(4);
    /**
     * 保存每个sheet计算出的银流结果
     */
    private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new
            ConcurrentHashMap<String, Integer>();
 
    private void count() {
        for (int i = 0; i < 4; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    // 计算当前sheet的银流数据，计算代码省略
                    sheetBankWaterCount
                            .put(Thread.currentThread().getName(), 1);
                    // 银流计算完成，插入一个屏障
                    try {
                        c.await();
                    } catch (InterruptedException |
                            BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }
 
    @Override
    public void run() {
        int result = 0;
        // 汇总每个sheet计算出的结果
        for (Map.Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()) {
            result += sheet.getValue();
        }
        // 将结果输出
        sheetBankWaterCount.put("result", result);
        System.out.println(result);
    }
 
    public static void main(String[] args) {
        BankWaterService bankWaterCount = new BankWaterService();
        bankWaterCount.count();
    }
}

使用线程池创建4个线程，分别计算每个sheet里的数据，每个sheet计算结果是1，再由 BankWaterService线程汇总4个sheet计算出的结果，案例也可以看出在声明CyclicBarrier的时候，还可以指定一个Runnable，当线程达到屏障的时候，可以优先执行Runnable中的方法

 
public class CyclicBarrierExample {
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> {
        log.info("callback is running");
    });
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++){
            final int threadNum = i;
            Thread.sleep(1000);
            executorService.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
    private static void race(int threadNum) throws Exception{
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        cyclicBarrier.await();
        log.info("{} continue", threadNum);
    }
}

CyclicBarrier原理

CyclicBarrier虽说功能与CountDownLatch类似，但是实现原理却完全不同， CyclicBarrier内部使⽤的是Lock + Condition实现的等待/通知模式。详情可以查看这个⽅法的源码：

private int dowait(boolean timed, long nanos)

六、Phaser

Phaser类有点复杂，这⾥只介绍⼀些基本的⽤法和知识点。详情可以查看JDK⽂档，⽂档⾥有这个类⾮常详尽的介绍。前⾯我们介绍了CyclicBarrier，可以发现它在构造⽅法⾥传⼊“任务总量” parties 之后，就不能修改这个值了，并且每次调⽤ await() ⽅法也只能消耗⼀个 parties 计数。但Phaser可以动态地调整任务总量！

名词解释：

party：对应⼀个线程，数量可以通过register或者构造参数传⼊;
arrive：对应⼀个party的状态，初始时是unarrived，当调⽤ arriveAndAwaitAdvance() 或者 arriveAndDeregister() 进⼊arrive状态，可以通过 getUnarrivedParties() 获取当前未到达的数量;
register：注册⼀个party，每⼀阶段必须所有注册的party都到达才能进⼊下⼀阶段;
deRegister：减少⼀个party
phase：阶段，当所有注册的party都arrive之后，将会调⽤Phaser 的 onAdvance() ⽅法来判断是否要进⼊下⼀阶段。

Phaser终⽌的两种途径，Phaser维护的线程执⾏完毕或者 onAdvance() 返回 true 此外Phaser还能维护⼀个树状的层级关系，构造的时候new Phaser(parentPhaser)，对于Task执⾏时间短的场景（竞争激烈），也就是说有⼤量的party, 那可以把每个Phaser的任务量设置较⼩，多个Phaser共同继承⼀个⽗ Phaser。

还是游戏的案例。假设我们游戏有三个关卡，但只有第⼀个关卡有新⼿教程，需要加载新⼿教程模块。但后⾯的第⼆个关卡和第三个关卡都不需要。我们可以⽤ Phaser来做这个需求

public class PhaserDemo{
    static class PreTaskThread implements Runnable {
        private String task;
        private Phaser phaser;
 
        public PreTaskThread(String task, Phaser phaser) {
            this.task = task;
            this.phaser = phaser;
        }
 
        @Override
        public void run() {
            for (int i = 1; i < 4; i++) {
                try {
                    // 第⼆次关卡起不加载NPC，跳过
                    if (i >= 2 && "加载新⼿教程".equals(task)) {
                        continue;
                    }
                    Random random = new Random();
                    Thread.sleep(random.nextInt(1000));
                    System.out.println(String.format("关卡%d，需要加载%d个模块，当前 ",i, phaser.getRegisteredParties(), task));
                    // 从第⼆个关卡起，不加载NPC
                    if (i == 1 && "加载新⼿教程".equals(task)) {
                        System.out.println("下次关卡移除加载【新⼿教程】模块");
                        phaser.arriveAndDeregister(); // 移除⼀个模块
                    } else {
                        phaser.arriveAndAwaitAdvance();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
 
    public static void main(String[] args) {
        Phaser phaser = new Phaser(4) {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println(String.format("第%d次关卡准备完成", phase + 1));
                return phase == 3 || registeredParties == 0;
            }
        };
        new Thread(new PreTaskThread("加载地图数据", phaser)).start();
        new Thread(new PreTaskThread("加载⼈物模型", phaser)).start();
        new Thread(new PreTaskThread("加载背景⾳乐", phaser)).start();
        new Thread(new PreTaskThread("加载新⼿教程", phaser)).start();
    }
}

关卡1，需要加载4个模块，当前模块【加载背景⾳乐】
关卡1，需要加载4个模块，当前模块【加载新⼿教程】
下次关卡移除加载【新⼿教程】模块
关卡1，需要加载3个模块，当前模块【加载地图数据】
关卡1，需要加载3个模块，当前模块【加载⼈物模型】
第1次关卡准备完成
关卡2，需要加载3个模块，当前模块【加载地图数据】
关卡2，需要加载3个模块，当前模块【加载背景⾳乐】
关卡2，需要加载3个模块，当前模块【加载⼈物模型】
第2次关卡准备完成
关卡3，需要加载3个模块，当前模块【加载⼈物模型】
关卡3，需要加载3个模块，当前模块【加载地图数据】
关卡3，需要加载3个模块，当前模块【加载背景⾳乐】
第3次关卡准备完成

这⾥要注意关卡1的输出，在“加载新⼿教程”线程中调⽤了 arriveAndDeregister() 减少⼀个party之后，后⾯的线程使⽤ getRegisteredParties() 得到的是已经被修改后的parties了。但是当前这个阶段 (phase)，仍然是需要4个parties都arrive才触发屏障的。从下⼀个阶段开始，才需要3个parties都arrive就触发屏障。另外Phaser类⽤来控制某个阶段的线程数量很有⽤，但它并在意这个阶段具体有哪些线程arrive，只要达到它当前阶段的parties值，就触发屏障。所以我这⾥的案例虽然制定了特定的线程（加载新⼿教程）来更直观地表述Phaser的功能，但是其实 Phaser是没有分辨具体是哪个线程的功能的，它在意的只是数量，这⼀点需要注意。

Phaser类的原理

Phaser类的原理相⽐起来要复杂得多。它内部使⽤了两个基于Fork-Join框架的原⼦类辅助：

private final AtomicReference<QNode> evenQ;
private final AtomicReference<QNode> oddQ;
static final class QNode implements ForkJoinPool.ManagedBlocker {
 // 实现代码
}

总的来说，CountDownLatch，CyclicBarrier，Phaser是⼀个⽐⼀个强⼤，但也⼀个⽐⼀个复杂。根据⾃⼰的业务需求合理选择即可

java多线程之线程通信工具类

一、概述

三、线程间交换数据的Exchanger

四、等待多线程完成的CountDownLatch

1、CyclicBarrirer概述

2、CountDownLatch和CyclicBarrier区别

3、CyclicBarrier Barrier被破坏

4、CyclicBarrirer案例

六、Phaser

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

java多线程之线程通信工具类

一、概述

三、线程间交换数据的Exchanger

四、等待多线程完成的CountDownLatch

1、CyclicBarrirer概述

2、CountDownLatch和CyclicBarrier区别

3、CyclicBarrier Barrier被破坏

4、CyclicBarrirer案例

六、Phaser

热门文章

最新文章

相关课程

相关电子书

相关实验场景