Java并发工具包二---CountDownLatch、CyclicBarrier、Executors

1、CountDownLatch
- 1.1、什么是CountDownLatch？
- 1.2、CountDownLatch的使用场景
- 1.3、CountDownLatch的api
- 1.4、CountDownLatch如何工作？
- 1.5、CountDownLatch中api反着用
2、CyclicBarrier
- 2.1、CyclicBarrier是什么？
- 2.2、CyclicBarrier的api
- 2.3、代码示例
3、Exchanger
- 3.1、Exchanger是什么？

1、CountDownLatch 1.1、什么是CountDownLatch？

CountDownLatch这个类能够使一个线程等待其他线程完成各自的工作后再执行。就比如说在服务启动时，主框架的一些代码逻辑希望在其它服务启动完成之后再执行。

1.2、CountDownLatch的使用场景

1、Zookeeper分布式锁。
2、Jmeter模拟高并发。

1.3、CountDownLatch的api

// 定义一个 CountDownLatch ,使得需要执行两个线程才能结束
CountDownLatch countDownLatch = new CountDownLatch(2);
// 第一个api，每执行一次，线程数减一
countDownLatch.countDown();
// 第二个api，会等到countDownLatch中的state（AQS中节点的state属性）值为0时才执行后面的逻辑
countDownLatch.await();

1.4、CountDownLatch如何工作？

CountDownLatch是通过一个计数器来实现的，计数器的初始值为线程的数量。每当一个线程完成了自己的任务后，计数器的值就会减1。当计数器值到达0时，它表示所有的线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。
代码示例如下：

public static void main(String[] args) throws InterruptedException {
    long now = System.currentTimeMillis();
    CountDownLatch countDownLatch = new CountDownLatch(2);

    new Thread(new SeeDoctorTask(countDownLatch)).start();
    new Thread(new QueueTask(countDownLatch)).start();

	// 主程序会卡在这里，直到上面的两个线程都执行完成
	// 然后再从这个地方出发，执行主程序
    countDownLatch.await();
    System.out.println("over，回家 cost:"+(System.currentTimeMillis()-now));
}

1.5、CountDownLatch中api反着用

一般情况下，CountDownLatch的两个api一般是先用countDown()，再用await()，而这两个api其实反着用也有另一个作用，代码示例如下：

public class CountDownLaunchRunner {

    static int sub = 0;
    static Object object = new Object();

    public static void main(String[] args) throws InterruptedException {
        long now = System.currentTimeMillis();
        CountDownLatch countDownLatch = new CountDownLatch(1);
        for(int i=0;i<10;i++){
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        countDownLatch.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    synchronized (object){
                        for(int j=0;j<1000;j++){
                            sub++;
                        }
                    }

                }
            });
        }
        // 这里等待10个线程全部启动成功
        Thread.sleep(3000);
        countDownLatch.countDown();
        System.out.println("全部准备完毕，出发：");
    }

}

相当于一批线程全部准备完毕，一起执行，类似于CyclicBarrier。

2、CyclicBarrier 2.1、CyclicBarrier是什么？

栅栏屏障，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。
CyclicBarrier默认的构造方法是CyclicBarrier（int parties），其参数表示屏障拦截的线程数量，每个线程调用await方法告CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。

2.2、CyclicBarrier的api

// 阻塞点，所有线程到达这一屏障之后，才可以开门出发
cyclicBarrier.await();

2.3、代码示例

从代码中我们就可以看出，当所有线程都准备好之后，才可以一块出发，可以使用多次，而前面的CountDownLatch反着用api时只能使用一次。

public class CyclicBarrierRunner implements Runnable {
    private CyclicBarrier cyclicBarrier;
    private int index ;

    public CyclicBarrierRunner(CyclicBarrier cyclicBarrier, int index) {
        this.cyclicBarrier = cyclicBarrier;
        this.index = index;
    }

    public void run() {
        try {
            System.out.println("index: " + index);
            cyclicBarrier.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(11, new Runnable() {
            public void run() {
                System.out.println("所有特工到达屏障，准备开始执行秘密任务");
            }
        });
        for (int i = 0; i < 10; i++) {
            new Thread(new CyclicBarrierRunner(cyclicBarrier, i)).start();
        }
        cyclicBarrier.await();
        System.out.println("全部到达屏障....1");

        Thread.sleep(5000);

        for (int i = 0; i < 10; i++) {
            new Thread(new CyclicBarrierRunner(cyclicBarrier, i)).start();
        }
        cyclicBarrier.await();
        System.out.println("全部到达屏障....1");
    }

}

3、Exchanger 3.1、Exchanger是什么？

当一个线程运行到exchange()方法时会阻塞，另一个线程运行到exchange()时，二者交换数据，然后执行后面的程序。
代码示例如下：

public class ExchangerRunner {

    public static void main(String []args) {

        final Exchanger exchanger = new Exchanger();
        for(int i = 0 ; i < 10 ; i++) {
            final Integer num = i;
            new Thread() {
                public void run() {
                    System.out.println("我是线程：Thread_" + this.getName() + "我的数据是：" + num);
                    try {
                        Integer exchangeNum = exchanger.exchange(num);
                        Thread.sleep(1000);
                        System.out.println("我是线程：Thread_" + this.getName() + "我原先的数据为：" + num + " , 交换后的数据为：" + exchangeNum);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }.start();
        }
    }

}

Java并发工具包二---CountDownLatch、CyclicBarrier、Executors

Java相关栏目本月热门文章