简约而不简单的CountDownLatch

Posted 2021-07-15 java叶新东老师

CountDownLatch是什么？

    CountDownLatch是JDK1.5之后提供的一个同步工具，在并发包下面，它可以让一个或多个线程等待，一直等到其他线程中执行完成一组操作。实现功能和java多线程中的join() 方法很像；想要详细了解join方法可以看我另一篇文章： java多线程join()方法的作用和实现原理

 CountDownLatch有哪些常用方法

CountDownLatch在调用构造方法初始化时，需要指定用给定一个整数作为计数器；这个计数器用来阻塞await方法；

• countDown方法 ： 计数器会被减1，调用此方法不会阻塞
• await方法           ： 如果计数器大于0时，线程会被阻塞，一直到计数器被countDown方法减到0时，线程才会继续执行。计数器是无法重置的，当计数器被减到0时，调用await方法都会直接返回。从而解除阻塞状态执行await方法后面的代码；

应用场景

    说明白了用法，那么这个同步工具能用在哪些场景呢？假设我们有以下的场景：

比如有三个人小红、小李、小王， 三个人相约一起去酒店吃饭，菜已经点好了， 三个人从不同的地方出发，只有三个人都到了酒店之后才会开始上菜；那么这三个人就分别代表三个线程，这三个线程执行完之后才会执行 “上菜” 的代码逻辑；这种场景下就可以使用CountDownLatch实现

上代码  ConutDownLatchDemo.java

package com.Lock;

import java.util.concurrent.CountDownLatch;

/**
 *  使用CountDownLatch同步器实现多线程等待
 *  比如有三个人小红、小李、小王， 三个人相约一起去酒店吃饭，菜已经点好了， 三个人从不同的地方出发，只有三个人都到了酒店之后才会开始上菜；那么这三个人就分别代表三个线程，这三个线程执行完之后才会执行 “上菜” 的代码逻辑，
 *
 */
public class ConutDownLatchDemo implements  Runnable{


    CountDownLatch latch ;

    public ConutDownLatchDemo (CountDownLatch latch){
        this.latch = latch;
    }


    @Override
    public void run() {

        System.out.println(Thread.currentThread().getName()+"开始出发了");
        try {
            Thread.sleep(1000);  // 到酒店的过程中用延时1秒来代替
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName()+"到酒店了");
        latch.countDown();// 计数器 - 1
    }
}

class Hotel implements  Runnable{

    // 同步工具
    CountDownLatch latch ;

    public Hotel (CountDownLatch latch){
        this.latch = latch;
    }
    @Override
    public void run() {

        System.out.println(Thread.currentThread().getName() +"正在等待大家的到来.....");
        try {
            // 如果计数器的值 > 0 ，则执行当前方法时进入阻塞，如果小于0，则放行
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("人齐了，"+Thread.currentThread().getName() +"服务员开始上菜");
    }
}

class Main1{
    public static void main(String[] args) throws InterruptedException {

        CountDownLatch latch = new CountDownLatch(3);
        new Thread(new ConutDownLatchDemo(latch), "小红").start();
        new Thread(new ConutDownLatchDemo(latch), "小王").start();
        new Thread(new ConutDownLatchDemo(latch), "小李").start();

        // 酒店线程
        new Thread(new Hotel(latch), "酒店").start();
    }
}

通过打印结果可以看到，服务员上菜之前，酒店的线程一直阻塞着，只有当三个人都到了酒店之后才会开始上菜；

其中一个线程阻塞了或者挂了怎么办？

三个人一起到饭店吃饭，如果有一个人在路上出了点事情，迟迟不能到饭店，或者说临时被老板叫去加班， 不来了，其他的人也不能一直等啊，这时候怎么办？

解决办法很简单，await() 有一个重载的方法 

await(long timeout, TimeUnit unit)

比如我们设置为5秒钟，

latch.await(5, TimeUnit.SECONDS);

那么超过5秒后，不管人来没来齐，服务员都会照样上菜；

CountDownLatch的实现原理

CountDownLatch有一个内部类叫做Sync，它继承了AbstractQueuedSynchronizer类，其中维护了一个整数state，并且保证了修改state的可见性和原子性。

创建CountDownLatch实例时，也会创建一个Sync的实例，同时把计数器的值传给Sync实例，具体是这样的：

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

在 countDown方法中，只调用了Sync实例的releaseShared方法，具体是这样的：

public void countDown() {
    sync.releaseShared(1);
}

其中的releaseShared方法，先对计数器进行减1操作，如果减1后的计数器为0，唤醒被await方法阻塞的所有线程，具体是这样的：

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) { //对计数器进行减一操作
        doReleaseShared();//如果计数器为0，唤醒被await方法阻塞的所有线程
        return true;
    }
    return false;
}

其中的tryReleaseShared方法，先获取当前计数器的值，如果计数器为0时，就直接返回；如果不为0时，使用CAS方法对计数器进行减1操作，具体是这样的：

protected boolean tryReleaseShared(int releases) {
    for (;;) {//死循环，如果CAS操作失败就会不断继续尝试。
        int c = getState();//获取当前计数器的值。
        if (c == 0)// 计数器为0时，就直接返回。
            return false;
        int nextc = c-1;
        if (compareAndSetState(c, nextc))// 使用CAS方法对计数器进行减1操作
            return nextc == 0;//如果操作成功，返回计数器是否为0
    }
}

在await方法中，只调用了Sync实例的acquireSharedInterruptibly方法，具体是这样的：

public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

其中acquireSharedInterruptibly方法，判断计数器是否为0，如果不为0则阻塞当前线程，具体是这样的：

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)//判断计数器是否为0
        doAcquireSharedInterruptibly(arg);//如果不为0则阻塞当前线程
}

 其中tryAcquireShared方法，是AbstractQueuedSynchronizer中的一个模板方法，其具体实现在Sync类中，其主要是判断计数器是否为零，如果为零则返回1，如果不为零则返回-1，具体是这样的：

protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}