Java并发的同步辅助类

小鱼儿 2021-09-22 08:16 378阅读 0赞

位于java.util.concurrent的三个同步辅助类  
\- CountDownLatch  
\- CyclicBarrier  
\- Semaphore

--------------------

### CountDownLatch ###

> 直译过来就是倒计数(CountDown)门闩(Latch)。倒计数不用说，门闩的意思顾名思义就是阻止前进。在这里就是指 CountDownLatch.await() 方法在倒计数为0之前会阻塞当前线程。

1.  实例化需要指定一个数量：  
    CountDownLatch latch = new CountDownLatch(5);
2.  某个线程执行latch.await()并阻塞。
3.  其他线程执行latch.countdown(),每执行一次数量 - 1，直到数量为0，执行了await的线程才会被唤醒继续执行。

CountDownLatch 的作用和 Thread.join() 方法类似，可用于一组线程和另外一组线程的协作。例如，主线程在做一项工作之前需要一系列的准备工作，只有这些准备工作都完成，主线程才能继续它的工作。这些准备工作彼此独立，所以可以并发执行以提高速度。在这个场景下就可以使用 CountDownLatch 协调线程之间的调度了。在直接创建线程的年代（Java 5.0 之前），我们可以使用 Thread.join()。在 JUC 出现后，因为线程池中的线程不能直接被引用，所以就必须使用 CountDownLatch 了。

static void test() throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);
        for (int i = 0; i < 3; ++i) 
            //开启其他线程
            new Thread(new Worker(latch), "" + i).start();
            System.out.println("主线程调用await()方法");
            latch.await();
            // 所有线程执行完毕主线程才继续执行
            System.out.println("主线程继续执行"); 
        }
    
    //内部工作类
    static class Worker implements Runnable { 
        private final CountDownLatch latch;
        Worker(CountDownLatch latch) {
        this.latch = latch;
        }
        @Override
        public void run() {
            try {
                //让线程随机睡眠并调用countdown
                Thread.sleep((long) (Math.random() * 5 * 1000));
                System.out.println("线程" + Thread.currentThread().getName() + "调用countDown方法");
                latch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
            }
        }
    
    输出：
    主线程调用await()方法
    线程1调用countDown方法
    线程0调用countDown方法
    线程2调用countDown方法
    主线程继续执行

--------------------

### CyclicBarrier ###

> CyclicBarrier 翻译过来叫循环栅栏、循环障碍什么的（还是有点别扭的。所以还是别翻译了，只可意会不可言传啊）。它主要的方法就是一个：await()。await() 方法没被调用一次，计数便会减少1，并阻塞住当前线程。当计数减至0时，阻塞(栅栏)解除，所有在此 CyclicBarrier 上面阻塞的线程开始运行。在这之后，如果再次调用 await() 方法，计数就又会变成 N-1，新一轮重新开始，这便是 Cyclic(循环) 的含义所在。

需要注意的是：  
\- CyclicBarrier 的使用并不难，但需要主要它所相关的异常。除了常见的异常，CyclicBarrier.await() 方法会抛出一个独有的 BrokenBarrierException。这个异常发生在当某个线程在等待本 CyclicBarrier 时被中断或超时或被重置时，其它同样在这个 CyclicBarrier 上等待的线程便会受到 BrokenBarrierException。意思就是说，同志们，别等了，有个小伙伴已经挂了，咱们如果继续等有可能会一直等下去，所有各回各家吧。  
\- CyclicBarrier.await() 方法带有返回值，用来表示当前线程是第几个到达这个 Barrier 的线程。

1.  和CountDownLatch一样，实例化CyclicBarrier也需要指定一个数量N，而且CyclicBarrier 的构造函数还可以接受一个 Runnable，会在 CyclicBarrier 被释放时执行。
2.  其他程序执行CyclicBarrier.await()方法，每执行一次 -1，并阻塞线程。
3.  直到count为0时，执行第一步指定的runnable对象，并唤醒其他线程继续执行。
4.  下一次调用await()方法时，count数量从头计算，即 N - 1。

static void test() {
        //实例化时指定一个runnable对象
        final CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() {
            @Override
            public void run() {
                System.out.println("barrier的Runnable");
            }
        });
        for (int i = 0; 3 > i; ++i)
            new Thread(new Worker(i, barrier)).start();
    }
    
    //内部工作类
    static class Worker implements Runnable {
        private int myRow;
        private CyclicBarrier barrier;
        private Worker(int row, CyclicBarrier barrier) {
            this.myRow = row;
            this.barrier = barrier;
        }
        @Override
        public void run() {
            System.out.println(myRow + " : before await()");
            try {
                //线程在此处阻塞，直到barrier的count为0才继续执行
                barrier.await();
                System.out.println(myRow + " : after await()");
            } catch (InterruptedException | BrokenBarrierException ex) {
                return;
            }
        }
    }
    
    输出：
    0 : before await()
    1 : before await()
    2 : before await()
    barrier的Runnable
    2 : after await()
    0 : after await()
    1 : after await()

CyclicBarrier 和 CountDownLatch 在用法上的不同  
1. CountDownLatch的count减为0时，继续调用countdown()不会重新计算，count数量依旧为0；  
2. CountDownLatch 适用于一组线程和另一个主线程之间的工作协作，一个主线程等待一组工作线程的任务完毕才继续它的执行是使用 CountDownLatch 的主要场景；  
3. CyclicBarrier 用于一组或几组线程，比如一组线程需要在一个时间点上达成一致，例如同时开始一个工作。  
4. CyclicBarrier 的循环特性和构造函数所接受的 Runnable 参数也是 CountDownLatch 所不具备的。

--------------------

### Semaphore ###

> Semaphore 直译是信号量，可能称它是许可量更容易理解。当然，因为在计算机科学中这个名字由来已久，所以不能乱改。它的功能比较好理解，就是通过构造函数设定一个数量的许可，然后通过 acquire 方法获得许可，release 方法释放许可。

看源码可知，信号量维护了一个许可集permit。

1.  实例化此对象，并指定许可集大小，表示有多少个许可。  
    Semaphore semp = new Semaphore(1);
2.  某个线程调用semp.acquire()获取许可，如果许可集大小为0则线程在此处阻塞，如果大于0，线程继续执行，拿到许可且permit - 1；
3.  拿到许可的线程执行semp.release();释放许可且permit + 1。

public static void main(String[] args) {
        // 线程池
        ExecutorService exec = Executors.newCachedThreadPool();
        // 只能1个线程同时访问
        final Semaphore semp = new Semaphore(1);
        // 模拟5个客户端访问
        for (int index = 0; 5 > index; index++) {
            final int Num = index;
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    try {
                        //获取许可
                        //获取到就继续执行
                        //获取不到阻塞
                        semp.acquire();
                        System.out.println("Accessing: " + Num);
                    //访问完后释放许可，如果屏蔽下面的语句，
                    //则在控制台只能打印1条记录，之后线程一直阻塞
                        semp.release();
                    } catch (InterruptedException e) {}
                }
            };
            exec.execute(run);
        }
        // 退出线程池
        exec.shutdown();
    }