java Concurrent包学习笔记（二）：CountDownLatch和CyclicBarrier

时间：2023-01-22 15:19:22

一、CountDownLatch

CountDownLatch一个线程同步的工具，是的一个或者多个线程等待其他线程操作完成之后再执行。

CountDownLatch通过一个给定的数值count来进行初始化，方法await()一直阻塞直到当前的count到达零为止，count的数值通过countDown()方法来减1，count的数值一旦设定就不能再修改，如果需要进行修改，请考虑使用CyclicBarrier。

大体看了一下源代码，是通过同步队列来作为计数器来进行控制的。同步队列是在CountDownLatch内部实现了一个静态内部类，countDown（）通过调用队列来减1。

有两个典型的应用场景：

第一种是一个开始的信号，所有的task任务等待这个信号。类似于百米赛跑中的信号员，所有的运动员都做好准备，等待信号，信号一来，那就开始运行，裁判等到所有的人员到终点后才能知道结果。

第二种是将一个任务分支N部分由M个线程来处理，等待所有的线程M完成任务后做其他的事情，还是刚才的例子，所有运动员跑完之后，才能知道所有人员的排名情况。

public class CountDownLatchTest {

    public static void main(String[] args) throws Exception{

        CountDownLatch s = new CountDownLatch(1);

        CountDownLatch e = new CountDownLatch(6);

        for(int i=0;i<6;i++){

            new Thread(new Worker(s,e)).start();

        }

        System.out.println("i am the judge ,now ,i start the singal");

        s.countDown();

        System.out.println("waiting all task over"+e.getCount());

        e.await();

        System.out.println("all is over");

    }

}

class Worker implements Runnable{

    private final CountDownLatch startSingal ;

    private final CountDownLatch endSingal;

    public Worker(CountDownLatch startSingal, CountDownLatch endSingal) {

        super();

        this.startSingal = startSingal;

        this.endSingal = endSingal;

    }  

    public void run() {

        try {

            System.out.println(Thread.currentThread().getName()+"waiting the start singal...."+startSingal.getCount());

            //等待开始信号信号

            startSingal.await();

            System.out.println(Thread.currentThread().getName()+"start to executer");

            //结束的计数器减一

            endSingal.countDown();

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

    }  

} 
执行结果：

Thread-0waiting the start singal....1

Thread-2waiting the start singal....1

Thread-1waiting the start singal....1

i am the judge ,now ,i start the singal

waiting all task over6

Thread-1start to executer

Thread-3waiting the start singal....1

Thread-4waiting the start singal....1

Thread-3start to executer

Thread-5waiting the start singal....0

Thread-5start to executer

Thread-2start to executer

Thread-0start to executer

Thread-4start to executer

all is over
    public static void main(String[] args) throws Exception{

        CountDownLatch latch = new CountDownLatch(6);

        Executor e = Executors.newFixedThreadPool(6);

        System.out.println("thread number is 6,now start");

        for(int i=0;i<6;i++){

            e.execute(new Worker(latch,i));

        }

        System.out.println("waiting all is over ");

        latch.await();

        System.out.println("all is over");  

    }

}

class Worker implements Runnable{

    private final CountDownLatch number;

    private int temp;

    public Worker(CountDownLatch number, int temp) {

        super();

        this.number = number;

        this.temp = temp;

    }

    public void run() {

        try {

            Thread.sleep(1000);

        } catch (InterruptedException e) {}

        System.out.println(Thread.currentThread().getName()+"runnable  - "+temp);

        number.countDown();

    }  

}

结果

thread number is 6,now start

waiting all is over

pool-1-thread-4runnable - 3

pool-1-thread-5runnable - 4

pool-1-thread-1runnable - 0

pool-1-thread-3runnable - 2

pool-1-thread-2runnable - 1

pool-1-thread-6runnable - 5

all is over

二、CyclicBarrier

CyclicBarrier是一个所有线程必须等待的一个栅栏，直到指定数量的线程都到达栅栏位置，然后所有线程才可以继续做其他事情。

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

CyclicBarrier还提供一个更高级的构造函数CyclicBarrier(int parties, Runnable barrierAction)，用于在线程到达屏障时，优先执行barrierAction这个Runnable对象，方便处理更复杂的业务场景。

使用场景一：所有人等待一个信号才能继续做自己的事，这里这个信号就是“所有人到齐”，只有所有人到齐后才能开始“吃饭”

package cyclicbarrier.demo;

import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

/**

 * @author boshen

 * @date 2018/12/20

 */

public class CyclicBarrierTest1 {

    private CyclicBarrier cb = new CyclicBarrier();

    class StudentThread implements Runnable{

        private String name;

        private int waitSecond;

        StudentThread(String name,int waitSecond){

            this.name = name;

            this.waitSecond = waitSecond;

        }

        public void run(){

            try {

                Thread.sleep(waitSecond);

                System.out.println("学生：" + name + " 开始等待所有人到齐");

                cb.await();

                System.out.println("学生：" + name + " 开始吃饭");

            } catch (InterruptedException e) {

            } catch (BrokenBarrierException e) {

            }

        }

    }

    public static void main(String[] args){

        CyclicBarrierTest1 cb = new CyclicBarrierTest1();

        ExecutorService executorService = Executors.newCachedThreadPool();

        executorService.submit(cb.new StudentThread("张三",));

        executorService.submit(cb.new StudentThread("李四",));

        executorService.submit(cb.new StudentThread("王五",));

        executorService.submit(cb.new StudentThread("马六",));

        executorService.submit(cb.new StudentThread("赵七",));

        executorService.shutdown();

    }

}

执行结果：

学生：李四 开始等待所有人到齐

学生：马六 开始等待所有人到齐

学生：张三 开始等待所有人到齐

学生：王五 开始等待所有人到齐

学生：赵七 开始等待所有人到齐

学生：赵七 开始吃饭

学生：李四 开始吃饭

学生：马六 开始吃饭

学生：张三 开始吃饭

学生：王五 开始吃饭

使用场景二：所有人等待一个信号才能继续做自己的事，这里这个信号就是“所有人到齐后，老师发话可以吃饭了”，所有人才开始“吃饭”

将上面的例子改造一下，使用了CyclicBarrier(int parties, Runnable barrierAction)，即所有人都到齐后先执行barrierAction，然后各线程才继续执行“

package cyclicbarrier.demo;

import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

/**

 * @author boshen

 * @date 2018/12/20

 */

public class CyclicBarrierTest2 {
private CyclicBarrier cb = new CyclicBarrier(5, new Runnable() {

        public void run() {

            System.out.println("老师说： 大家开始吃饭了");

        }

    });
class StudentThread implements Runnable{

        private String name;

        private int waitSecond;

        StudentThread(String name,int waitSecond){

            this.name = name;

            this.waitSecond = waitSecond;

        }

        public void run(){

            try {

                Thread.sleep(waitSecond);

                System.out.println("学生：" + name + " 开始等待所有人到齐");

                cb.await();

                System.out.println("学生：" + name + " 开始吃饭");

            } catch (InterruptedException e) {

            } catch (BrokenBarrierException e) {

            }

        }

    }

    public static void main(String[] args){

        CyclicBarrierTest2 cb = new CyclicBarrierTest2();

        ExecutorService executorService = Executors.newCachedThreadPool();

        executorService.submit(cb.new StudentThread("张三",3));

        executorService.submit(cb.new StudentThread("李四",1));

        executorService.submit(cb.new StudentThread("王五",4));

        executorService.submit(cb.new StudentThread("马六",2));

        executorService.submit(cb.new StudentThread("赵七",5));

        executorService.shutdown();

    }

}

输出：

学生：李四 开始等待所有人到齐

学生：王五 开始等待所有人到齐

学生：张三 开始等待所有人到齐

学生：马六 开始等待所有人到齐

学生：赵七 开始等待所有人到齐
老师说： 大家开始吃饭了

学生：赵七 开始吃饭

学生：李四 开始吃饭

学生：王五 开始吃饭

学生：张三 开始吃饭

学生：马六 开始吃饭

CountDownLatch：一个或者多个线程，等待其他多个线程完成某件事情之后才能执行；

CyclicBarrier：多个线程互相等待，直到到达同一个同步点，再继续一起执行。

对于CountDownLatch来说，重点是“一个线程（多个线程）等待”，而其他的N个线程在完成“某件事情”之后，可以终止，也可以等待。而对于CyclicBarrier，重点是多个线程，在任意一个线程没有完成，所有的线程都必须等待。

CountDownLatch是计数器，线程完成一个记录一个，只不过计数不是递增而是递减，而CyclicBarrier更像是一个阀门，需要所有线程都到达，阀门才能打开，然后继续执行。

相关文章