转载

J.U.C 之CountDownLatch

CountDownLatch 与 CyclicBarrier 有点相似。

用给定的计数初始化 CountDownLatch。由于调用了 #countDown() 方法，所以在当前计数到达零之前，#await() 方法会一直受阻塞。之后，会释放所有等待的线程，#await() 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier 。

J.U.C 之CountDownLatch

CountDownLatch 是通过一个计数器来实现的，当我们在 new 一个 CountDownLatch 对象的时候，需要带入该计数器值，该值就表示了线程的数量。

每当一个线程完成自己的任务后，计数器的值就会减 1 。 当计数器的值变为0时，就表示所有的线程均已经完成了任务，然后就可以恢复等待的线程继续执行了。 虽然，CountDownLatch 与 CyclicBarrier 有那么点相似，但是他们还是存在一些区别的：

  1. CountDownLatch 的作用是允许 1 或 N 个线程等待其他线程完成执行；
  2. 而 CyclicBarrier 则是允许 N 个线程相互等待。
  3. CountDownLatch 的计数器无法被重置；CyclicBarrier 的计数器可以被重置后使用，因此它被称为是循环的 barrier 。

实现分析

java.util.concurrent.CountDownLatch 结构如下图：

J.U.C 之CountDownLatch

通过上面的结构图我们可以看到，CountDownLatch 内部依赖 Sync 实现，而 Sync 继承 AQS 。

CountDownLatch 仅提供了一个构造方法，代码如下：

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

构造一个用给定计数初始化的 CountDownLatch 。

Sync

sync 变量，为 CountDownLatch 的一个内部类 Sync ，其定义如下：

private static final class Sync extends AbstractQueuedSynchronizer {
        
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);
        }

        // 获取同步状态
        int getCount() {
            return getState();
        }

        // 获取同步状态
        @Override
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

        // 释放同步状态
        @Override
        protected boolean tryReleaseShared(int releases) {
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }
    
}
复制代码
  1. 通过这个内部类 Sync 实现类，我们可以清楚地看到， CountDownLatch 是采用共享锁来实现的。
  2. #tryAcquireShared(int acquires) 和 #tryReleaseShared(int releases) 方法，结合下文一起理解。

await

CountDownLatch 提供 #await() 方法，来使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断，定义如下：

public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}
复制代码

该方法内部使用 AQS 的 #acquireSharedInterruptibly(int arg) 方法，代码如下：

// AQS.java
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}
复制代码

在内部类 Sync 中重写了 #tryAcquireShared(int arg)方法，代码如下：

// Sync.java
@Override
protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}
复制代码

getState() 方法，获取同步状态，其值等于计数器的值。从这里我们可以看到，如果计数器值不等于 0，则会调用 #doAcquireSharedInterruptibly(int arg) 方法。该方法为一个自旋方法会尝试一直去获取同步状态，代码如下：

// AQS.java
private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                /**
                 * 对于CountDownLatch而言，如果计数器值不等于0，那么r 会一直小于0
                 */
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            //等待
            if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
复制代码

await

CountDownLatch 提供 #await(long timeout, TimeUnit unit) 方法，来使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断，或者等待超时，定义如下：

public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
复制代码

调用 AQS 的 tryAcquireSharedNanos(int acquires, long nanosTimeout) 方法。

countDown

CountDownLatch 提供 #countDown() 方法，递减锁存器的计数。如果计数到达零，则唤醒所有等待的线程。

public void countDown() {
    sync.releaseShared(1);
}
复制代码

内部调用 AQS 的 #releaseShared(int arg) 方法，来释放共享锁同步状态：

// AQS.java
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}
复制代码

tryReleaseShared(int arg) 方法，被 CountDownLatch 的内部类 Sync 重写，代码如下：

// Sync.java
@Overrride
protected boolean tryReleaseShared(int releases) {
    for (;;) {
        //获取锁状态
        int c = getState();
        //c == 0 直接返回，释放锁成功
        if (c == 0)
            return false;
        //计算新“锁计数器”
        int nextc = c-1;
        //更新锁状态（计数器）
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}
复制代码

getCount

public long getCount() {
    return sync.getCount();
}
复制代码

总结

CountDownLatch 内部通过共享锁实现。

  1. 在创建 CountDownLatch 实例时，需要传递一个int型的参数：count，该参数为计数器的初始值，也可以理解为该共享锁可以获取的总次数。
  2. 当某个线程调用 #await() 方法，程序首先判断 count 的值是否为 0 ，如果不为 0 的话，则会一直等待直到为 0 为止。
  3. 当其他线程调用 #countDown() 方法时，则执行释放共享锁状态，使 count 值 - 1。
  4. 当在创建 CountDownLatch 时初始化的 count 参数，必须要有 count 线程调用#countDown() 方法，才会使计数器 count 等于 0 ，锁才会释放，前面等待的线程才会继续运行。
  5. 注意 CountDownLatch 不能回滚重置。

应用示例

示例仍然使用开会案例。老板进入会议室等待 5 个人全部到达会议室才会开会。所以这里有两种线程：老板等待开会线程、员工到达会议室线程：

public class CountDownLatchTest {

    private static CountDownLatch countDownLatch = new CountDownLatch(5);

    /**
     * Boss线程，等待员工到达开会
     */
    static class BossThread extends Thread{
        @Override
        public void run() {
            System.out.println("Boss在会议室等待，总共有" + countDownLatch.getCount() + "个人开会...");
            try {
                //Boss等待
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("所有人都已经到齐了，开会吧...");
        }
    }

    // 员工到达会议室线程
    static class EmpleoyeeThread  extends Thread{
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + "，到达会议室....");
            //员工到达会议室 count - 1
            countDownLatch.countDown();
        }
    }

    public static void main(String[] args){
        //Boss线程启动
        new BossThread().start();

        for(int i = 0 ; i < countDownLatch.getCount() ; i++){
            new EmpleoyeeThread().start();
        }
    }
}
复制代码

运行结果：

J.U.C 之CountDownLatch
原文  https://juejin.im/post/5d8b0ffe6fb9a04ddb3b61f2
正文到此结束
所属分类： 编程技术 Java

热门推荐

相关文章

阿里云首购8折
说给你听

本文目录
随机标签
书籍教程
springboot-demo Java入门教程 bootstrap3 CSS Apache基础教程 php ionic 教程 Python mysql教程 eclipse Ubuntu VPS系统配置 AngularJS 教程 MongoDB教程 Struts2教程 Redis教程 Spring教程 Git教程 Jenkins进阶系列 openfire参考指南 Java设计模式 HBase教程 Maven教程 hibernate教程 Docker 教程 memcached教程 Quartz指南 Hive教程 Ant教程 ANTLR教程 SpringCloud java实例教程 springcloud-demo XStream教程 Hazelcast教程 Elastic-Job-Lite 深入浅出MyBatis SVN教程 ibaties教程 rabittmq教程 solr教程 Hadoop教程 WebService CXF学习 JPA教程 ActiveMQ中文指南 java-demo Java内存模型 dubbo教程 python3-demo Linux入门视频教程
近期评论
  1. 1 Spring Cloud Consul实现选举机制
  2. 2 Spring Boot集成ShedLock实现分布式定时任务
  3. 3 利用oss进行数据库和网站图片备份
  4. 4 Spring Cloud Stream实现数据流处理
  5. 5 Python3访问MySQL数据库快速入门Demo
  6. 6 Github开源项目作者可以免费申请 JetBrains 全家桶
  7. 7 Spring Cloud Vault快速入门Demo
  8. 8 Spring Cloud Gateway快速入门Demo
  9. 9 Spring Cloud Contract快速入门Demo
  10. 10 Spring Cloud Consul快速入门Demo
网站信息
  • 文章总数：82,714 篇
  • 文件总数：284,287 个
  • 标签总数：2,393 个
  • 分类总数：85 个
  • 留言数量：2,560 条
  • 在线人数：669 人
  • 运行天数：4,409天
  • 最后更新：2024年11月22日05点
Loading...

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

[HBLOG]公众号 HBLOG HBLOG