ConcurrentUtils 常用类

Posted 2020-10-23 xiaobinggan

CyclicBarrier，CountDownLatch，Semaphore的一些用法

CyclicBarrier

含义

栅栏允许两个或者多个线程在某个集合点同步。当一个线程到达集合点时，它将调用await()方法等待其它的线程。线程调用await()方法后，CyclicBarrier将阻塞这个线程并将它置入休眠状态等待其它线程的到来。等最后一个线程调用await()方法时，CyclicBarrier将唤醒所有等待的线程然后这些线程将继续执行。CyclicBarrier可以传入另一个Runnable对象作为初始化参数。当所有的线程都到达集合点后，CyclicBarrier类将Runnable对象作为线程执行。

方法

await()：使线程置入休眠直到最后一个线程的到来之后唤醒所有休眠的线程

CyclicBarrier的使用场景：每个线程代表一个运动员，只有当所有的运动员都准备好后，才一起出发，只要有一个没有准备好，大家等待；

public class UseCyclicBarrier {

    // 静态内部类 实现 Runnable 接口
    static class Runner implements Runnable {
        private CyclicBarrier barrier;
        private String name;

        public Runner(CyclicBarrier barrier, String name) {
            this.barrier = barrier;
            this.name = name;
        }

        @Override
        public void run() {
            try {
                Thread.sleep(1000 * (new Random()).nextInt(5));
                System.out.println(name + " 准备OK.");
                barrier.await(); //
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println(name + " go!!!");
        }
    }

    public static void main(String[] args) {
        // 创建一个CyclicBarrier，将在3个线程处于等待状态时，启动
        CyclicBarrier barrier = new CyclicBarrier(3);
        // 创建一个有3个线程的线程池
        ExecutorService executor = Executors.newFixedThreadPool(3);

        executor.submit(new Thread(new Runner(barrier, "zhangsan")));
        executor.submit(new Thread(new Runner(barrier, "lisi")));
        executor.submit(new Thread(new Runner(barrier, "wangwu")));

        executor.shutdown();

    }
}

View Code

 执行结果

zhangsan 准备OK.
lisi 准备OK.
wangwu 准备OK.
wangwu go!!!
lisi go!!!
zhangsan go!!!

View Code

 CountDownLatch

含义

CountDownLatch可以理解为一个计数器在初始化时设置初始值，当一个线程需要等待某些操作先完成时，需要调用await()方法。这个方法让线程进入休眠状态直到等待的所有线程都执行完成。每调用一次countDown()方法内部计数器减1,直到计数器为0时唤醒。线程同步点比较特殊，为内部计数器值为0时开始。

方法

核心方法两个：countDown()和await()

countDown():使CountDownLatch维护的内部计数器减1,每个被等待的线程完成的时候调用

await():线程在执行到CountDownLatch的时候会将此线程置于休眠

使用场景：会议室里等与会人员到齐了会议才能开始。

public class VideoConference implements Runnable {
    private final CountDownLatch controller;

    public VideoConference(int number) {
        // 构造函数中 创建 countDoenLatch 这个对象
        controller = new CountDownLatch(number);
    }

    public void arrive(String name) {
        System.out.println(name+" has arrived");

        controller.countDown();// 调用countDown()方法，使内部计数器减1
        System.out.println(controller.getCount()+"VideoConference: Waiting for participants"
                );
    }

    @Override
    public void run() {
        try {
            controller.await();// 等待，直到CoutDownLatch计数器为0
            System.out.println("人已到齐 ，我们开始开会！");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static class Participant implements Runnable {
        private VideoConference conference;

        private String name;

        public Participant(VideoConference conference, String name) {
            this.conference = conference;
            this.name = name;
        }

        @Override
        public void run() {
            Long duration = (long) (Math.random() * 10);
            try {
                TimeUnit.SECONDS.sleep(duration);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            conference.arrive(name);// 每到一个人员，CountDownLatch计数器就减少1
        }
    }

    public static void main(String[] args) {
        VideoConference conference = new VideoConference(5);
        Thread threadConference = new Thread(conference);
        threadConference.start();// 开启await()方法，在内部计数器为0之前线程处于等待状态
        for (int i = 0; i < 5; i++) {
            Participant p = new Participant(conference, "Participant " + i);
            Thread t = new Thread(p);
            t.start();
        }
    }
}

View Code

需要注意的地方

CountDownLatch比较容易记忆的是他的功能，是一个线程计数器。等计数器为0时那些先前因调用await()方法休眠的线程被唤醒。

CountDownLatch能够控制的线程是哪些？是那些调用了CountDownLatch的await()方法的线程

具体使用方式，容易忘记：先运行await()方法的线程，例子中是视频会议的线程。然后是执行与会者 线程，这里的处理是每到一位（每创建一个线程并运行run()方法时就使计数器减1）就让计数器减1，等计数器减为0时唤醒因调用await()方法进入休眠的线程。这里的这些与会者就是要等待的线程。

Semaphore翻译成字面意思为 信号量，Semaphore可以控同时访问的线程个数，通过 acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可。

  
  
public class UseSemaphore {  
  
    public static void main(String[] args) {  
        // 线程池  
        ExecutorService exec = Executors.newCachedThreadPool();  
        // 只能5个线程同时访问  
        final Semaphore semp = new Semaphore(3);  
        // 模拟20个客户端访问  
        for (int index = 0; index < 10; index++) {  
            final int NO = index;  
            Runnable run = new Runnable() {  
                public void run() {  
                    try {  
                        // 获取许可  
                        semp.acquire();  
                        System.out.println("Accessing: " + NO);  
                        //模拟实际业务逻辑
                        Thread.sleep((long) (Math.random() * 10000));  
                        // 访问完后，释放  
                        semp.release();  
                    } catch (InterruptedException e) {  
                    }  
                }  
            };  
            exec.execute(run);  
        } 
        
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 退出线程池  
        exec.shutdown();  
    }  
}  

View Code