Java并发编程——CountDownLatch

作者: 小波同学 | 来源:发表于2021-12-11 22:53 被阅读0次

一、闭锁 CountDownLatch

一个同步工具类，允许一个或者多个线程一直等待，直到其他线程的操作都执行完成之后再继续往下执行。

使用场景：在一些应用场合中，需要等待某个条件达到要求后才能做后面的事情；同时当线程都完成后也会触发事件，以便进行后面的操作。这个时候就可以使用CountDownLatch。

CountDownLatch最重要的方法是countDown()和await()，前者主要是计数减一，后者是等待计数到0，如果没有到达0，就继续阻塞等待。

countdownlatch

如上图，左边三只小熊，可以当成三个线程，每一只撞到栏杆，计数器就减1，这相当于执行了countDown方法；

右边有两只暴走小熊在等待计数器变为0，可以当成两个线程，执行了await方法；

最终左边三只暴走小熊抵达了栏杆处，计数器变为0，唤醒了右边的暴走小熊，暴走小熊就开始动起来了。

二、执行原理

CountDownLatch是基于AQS共享模式的使用。

如下图，我们通过给CountDownLatch构造函数传入state的值。

countDown方法本质是释放共享锁，核心实现逻辑是：state>0 && state-1，如果state>0，则state减一，否则执行失败；

await方法本质是获取共享锁，核心实现是：getState()==0，如果state==0，则表示获取成功，否则线程阻塞进入等待队列；

当state减到0的时候，会唤醒等待队列中的所有线程，尝试继续获取共享锁，这个时候正常是所有线程都能获取成功的。

三、CountDownLatch的用法

3.1 CountDownLatch典型用法1

某一线程在开始运行前等待n个线程执行完毕。将CountDownLatch的计数器初始化为new CountDownLatch(n)，每当一个任务线程执行完毕，就将计数器减1 countdownLatch.countDown()，当计数器的值变为0时，在CountDownLatch上await()的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。

/**
 * @Description: 工厂中，质检，5个工人检查，所有人都认为通过，才通过
 */
public class CountDownLatchDemo {
    
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(5);

        ExecutorService executorService = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 5; i++) {
            final int no = i+1;
            executorService.submit(() -> {
                try {
                    Thread.sleep(new Random().nextInt(10000));
                    System.out.println("NO." + no + "完成了检查");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    countDownLatch.countDown();
                }
            });
        }

        System.out.println("等待5个人检查完......");

        countDownLatch.await();
        System.out.println("所有人都完成了工作，等待进入下一环节");
    }
}

3.2 CountDownLatch典型用法2

实现多个线程开始执行任务的最大并行性。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的CountDownLatch(1)，将其计算器初始化为1，多个线程在开始执行任务前首先countdownlatch.await()，当主线程调用countDown()时，计数器变为0，多个线程同时被唤醒。

/**
 * @Description: 模拟100米跑步，5名选手都准备好了，只等裁判员一声令下，所有人同时开跑
 */
public class CountDownLatchDemo2 {

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);

        ExecutorService executorService = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 5; i++) {
            final int no = i+1;
            executorService.submit(() -> {
                System.out.println("No." + no + "，准备完毕，等待发令枪");
                try {
                    countDownLatch.await();
                    System.out.println("No." + no + "，开始跑步");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        //裁判员检查发令枪......
        Thread.sleep(5000);
        System.out.println("发令枪响，比赛开始......");
        countDownLatch.countDown();
    }
}

3.3 CountDownLatch两种用法结合使用

/**
 * @Description: 模拟100米跑步，5名选手都准备好了，只等裁判员一声令下，所有人同时开跑,
 * 所有人都到终点后，比赛结束
 */
public class CountDownLatchDemo3 {

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch begin = new CountDownLatch(1);
        CountDownLatch end = new CountDownLatch(5);

        ExecutorService executorService = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 5; i++) {
            final int no = i+1;
            executorService.submit(() -> {
                System.out.println("No." + no + "，准备完毕，等待发令枪");
                try {
                    begin.await();
                    System.out.println("No." + no + "，开始跑步");
                    Thread.sleep(new Random().nextInt(10000));
                    System.out.println("No." + no + "，跑到终点了");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    end.countDown();
                }
            });
        }

        //裁判员检查发令枪......
        Thread.sleep(5000);
        System.out.println("发令枪响，比赛开始......");
        begin.countDown();

        //等待5个线程都执行完毕之后
        end.await();
        System.out.println("所有人到达终点，比赛结束");
    }
}

3.4 CountDownLatch注意点

扩展用法：多个线程等待多个线程完成执行后，再同时执行。
CountDownLatch是不能够重用的，如果需要重新计数，可以考虑使用CyclicBarrier或者创建新的CountDownLatch实例。

四、源码

4.1 构造方法：创建一个Sync对象，而Sync继承AQS

public class CountDownLatch {
    
    private final Sync sync;
    
    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        //设置计数器实际上是将值赋给了AQS状态变量state
        this.sync = new Sync(count);
    }
}

4.2 Sync 是CountDownLatch的内部私有类，组合到CountDownLatch里

public class CountDownLatch {
    
    private static final class Sync extends AbstractQueuedSynchronizer {
        //设置计数器实际上是将值赋给了AQS状态变量state
        Sync(int count) {
            setState(count);
        }

        //获取状态变量state的值
        int getCount() {
            return getState();
        }

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

        protected boolean tryReleaseShared(int releases) {
            //循环进行CAS，直到当前线程完成CAS减去1操作
            for (;;) {
                int c = getState();
                //当前状态值为0则直接返回
                if (c == 0)
                    return false;
                int nextc = c-1;
                //使用CAS让计数器值减去1
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

    private final Sync sync;
}

在AQS中state是一个private volatile int类型的变量。CountDownLatch使用state来计数，CountDownLatch的getCount最终调用的是AQS的getState()，返回state进行计数。

4.3 await()方法：调用AQS的acquireSharedInterruptibly方法

public class CountDownLatch {
    
    private final Sync sync;
    
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }   
}

acquireSharedInterruptibly方法：获取共享锁

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
    //获取共享锁
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        //判断线程是否为中断状态，如果是抛出interruptedException 
        if (Thread.interrupted())
            throw new InterruptedException();
        //尝试获取共享锁，尝试成功就返回，否则调用doAcquireSharedInterruptibly方法    
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }
}

tryAcquireShared方法：尝试获取共享锁

public class CountDownLatch {
    
    private final Sync sync;

    private static final class Sync extends AbstractQueuedSynchronizer {
        //尝试获取共享锁，重写AQS里面的方法
        protected int tryAcquireShared(int acquires) {
            //锁状态 == 0，表示所没有被任何线程所获取，
            //即是可获取的状态，否则锁是不可获取的状态
            return (getState() == 0) ? 1 : -1;
        }
    }
}

doAcquireSharedInterruptibly方法：会使得当前线程一直等待，直到当前线程获取到锁（或被中断）才返回

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
    private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        //创建“当前线程”的Node节点，且node中记录的锁是“共享锁”类型，并将节点添加到CLH队列末尾。
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                //获取前继节点，如果前继节点是等待锁队列的表头，则尝试获取共享锁
                // 判断新增的节点的前一个节点是否头节点
                final Node p = node.predecessor();
                if (p == head) {
                    // 是头节点，那么在此尝试获取共享锁
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        // 获取成功，把当前节点变为新的head节点，
                        //并且检查后续节点是否可以在共享模式下等待，
                        //并且允许继续传播，则调用doReleaseShared继续唤醒下一个节点尝试获取锁
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                //前继节点不是头节点，当前线程一直等待，直到获取到锁
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
}

shouldParkAfterFailedAcquire方法：判断当前线程获取锁失败之后是否需要挂起

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
 /*说明：4.shouldParkAfterFailedAcquire 返回当前线程是否应该阻塞
    (01) 关于waitStatus请参考下表(中扩号内为waitStatus的值)

    CANCELLED[1]  -- 当前线程已被取消
    SIGNAL[-1]    -- “当前线程的后继线程需要被unpark(唤醒)”。
                        一般发生情况是：当前线程的后继线程处于阻塞状态，
                        而当前线程被release或cancel掉，因此需要唤醒当前线程的后继线程。
    CONDITION[-2] -- 当前线程(处在Condition休眠状态)在等待Condition唤醒
    PROPAGATE[-3] -- (共享锁)其它线程获取到“共享锁”
    [0]           -- 当前线程不属于上面的任何一种状态。
    
    (02) shouldParkAfterFailedAcquire()通过以下规则，判断“当前线程”是否需要被阻塞。

    规则1：如果前继节点状态为SIGNAL，表明当前节点需要被unpark(唤醒)，此时则返回true。
    规则2：如果前继节点状态为CANCELLED(ws>0)，说明前继节点已经被取消，则通过先前回溯找到一个有效(非CANCELLED状态)的节点，并返回false。
    规则3：如果前继节点状态为非SIGNAL、非CANCELLED，则设置前继的状态为SIGNAL，并返回false。
    */  
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        // 前驱节点的状态
        int ws = pred.waitStatus;
        // 如果前驱节点是SIGNAL状态，则意味着当前线程需要unpark唤醒，此时返回true
        if (ws == Node.SIGNAL)
            return true;
            
        // 如果前继节点是取消的状态即前驱节点状态为CANCELLED 
        if (ws > 0) {
            // 从队尾向前寻找第一个状态不为CANCELLED的节点
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // 将前驱节点的状态设置为SIGNAL
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }   
}

4.4 countDown()源码

public class CountDownLatch {
    
    private final Sync sync;

    public void countDown() {
        //该方法其实调用AQS中的releaseShared(1)释放共享锁方法。
        sync.releaseShared(1);
    }
}

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
    //目的是让当前线程释放它所持有的共享锁，它首先会通过tryReleaseShared()去尝试释放共享锁。
    //尝试成功，则直接返回；尝试失败，则通过doReleaseShared()去释放共享锁。
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }
}

tryReleaseShared()在CountDownLatch.java中被重写，释放共享锁，将锁计数器-1

public class CountDownLatch {
    
    private final Sync sync;

    private static final class Sync extends AbstractQueuedSynchronizer {
    
        protected boolean tryReleaseShared(int releases) {
            for (;;) {
                // 获取“锁计数器”的状态
                int c = getState();
                if (c == 0)
                    return false;
                // “锁计数器”-1 
                int nextc = c-1;
                // 通过CAS函数进行赋值。
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }
}

参考：
https://www.itzhai.com/articles/graphical-several-fun-concurrent-helper-classes.html

https://zhuanlan.zhihu.com/p/95835099

https://www.cnblogs.com/200911/p/6059719.html

Java并发编程——CountDownLatch

一、闭锁 CountDownLatch

二、执行原理

三、CountDownLatch的用法

3.1 CountDownLatch典型用法1

3.2 CountDownLatch典型用法2

3.3 CountDownLatch两种用法结合使用

3.4 CountDownLatch注意点

四、源码

4.1 构造方法：创建一个Sync对象，而Sync继承AQS

4.2 Sync 是CountDownLatch的内部私有类，组合到CountDownLatch里

4.3 await()方法：调用AQS的acquireSharedInterruptibly方法

4.4 countDown()源码

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

并发

并发

给自己