AbstractQueuedSynchronizer理解(CountDownLatch)

简介: AbstractQueuedSynchronizer理解(CountDownLatch)

本文分析一下CountDownLatch是如何运用AQS的

CountDownLatch是什么

CountDownLatch顾名思义它是一个Latch(门闩),它是用一个计数器实现的,初始状态计数器的数值等于线程数,每当有线程完成任务后,计数器就会减一。当state为0时,锁就会被释放,凡是之前因抢占锁而等待的线程这时候就会被唤醒继续抢占锁。

CountDownLatch小栗子

public static void main(String[] args) throws InterruptedException{
    int threadSize = 3;
    CountDownLatch doneSignal = new CountDownLatch(threadSize);

    for (int i = 1; i <= threadSize; i++) {
        final int threadNum = i;
        new Thread(() -> {
            System.out.println("thread" + threadNum + ":start");

            try {
                Thread.sleep(1000 * threadNum);
            } catch (InterruptedException e) {
                System.out.println("thread" + threadNum + ":exception");
            }

            doneSignal.countDown();
            System.out.println("thread" + threadNum + ":complete");
        }).start();
    }

    System.out.println("main thread:await");
    doneSignal.await();
    System.out.println("main thread:go on");
}

例子中主线程启动了三条子线程,睡眠一段时间,此时主线程在等待所有子线程结束后才会继续执行下去;
看一下输出结果:

main thread:await
thread1:start
thread2:start
thread3:start
thread1:complete
thread2:complete
thread3:complete
main thread:go on

Process finished with exit code 0

CountDownLatch原理分析

既然CountDownLatch也是AQS的一种使用方式,我们看一下它的内部类Syc是怎么实现AQS的:

private static final class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 4982264981922014374L;
    
    //构造函数,初始化同步状态state的值,即线程个数
    Sync(int count) {
        setState(count);
    }

    int getCount() {
        return getState();
    }

    //这里重写了方法,在共享模式下,告诉调用者是否可以抢占state锁了,正数代表可以,负数代表否定;当state为0时返回正数
    protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
    }

    //共享模式下释放锁
    protected boolean tryReleaseShared(int releases) {
        // Decrement count; signal when transition to zero
        for (;;) {
            int c = getState();
            //state为0时说明没有什么可释放
            if (c == 0)
                return false;
            int nextc = c-1;
            if (compareAndSetState(c, nextc))
                //CAS对state操作成功后返回state值是否为0,为0则释放成功
                return nextc == 0;
        }
    }
}

看完了重写的AQS同步器后,我们了解了CountDownLatch对state锁的描述。接下来先看主线程调用的await方法,在await方法里调用了AQS的acquireSharedInterruptibly:

//在共享模式下尝试抢占锁
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    //线程中断抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //尝试抢占前先查询一下是否可以抢占,如果返回值大于0程序往下执行,小于0则等待
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}


private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    //在Reentrant解析中我们看过,往队列中新增node(共享模式)
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                //如果当前node的前继时head,马上尝试抢占锁
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    //如果state==0即允许往下执行,重新设置head并往下传播信号
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    //得到往下执行的允许
                    return;
                }
            }
            //以下都跟Reentrant一样
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    //将当前node设置为head,清空node的thread、prev
    setHead(node);
    /*
     * Try to signal next queued node if:
     *   Propagation was indicated by caller,
     *     or was recorded (as h.waitStatus either before
     *     or after setHead) by a previous operation
     *     (note: this uses sign-check of waitStatus because
     *      PROPAGATE status may transition to SIGNAL.)
     * and
     *   The next node is waiting in shared mode,
     *     or we don't know, because it appears null
     *
     * The conservatism in both of these checks may cause
     * unnecessary wake-ups, but only when there are multiple
     * racing acquires/releases, so most need signals now or soon
     * anyway.
     */
    //如果propagate大于0,或者原来head的等待状态小于0或者现在head的等待状态小于0
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        //准备唤醒下一个节点
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

private void doReleaseShared() {
    /*
     * Ensure that a release propagates, even if there are other
     * in-progress acquires/releases.  This proceeds in the usual
     * way of trying to unparkSuccessor of head if it needs
     * signal. But if it does not, status is set to PROPAGATE to
     * ensure that upon release, propagation continues.
     * Additionally, we must loop in case a new node is added
     * while we are doing this. Also, unlike other uses of
     * unparkSuccessor, we need to know if CAS to reset status
     * fails, if so rechecking.
     */
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                //如果head的状态为SIGNAL,更改状态为0
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                //唤醒后继节点
                unparkSuccessor(h);
            }
            //如果head状态为0,更改状态为PROPAGATE
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        //如果head没有改变,结束当前loop,如果遇到head被别的线程改变,继续loop
        if (h == head)                   // loop if head changed
            break;
    }
}

释放锁的信号一直向后传播,直到所有node被唤醒并继续执行,那第一个信号时何时发起的呢?我们来看一下CountDownLatch的countDown方法,该方法调用了sync的releaseShared方法:

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        //如果同步状态state为0时,调用doReleaseShared,在这里就发出了第一个唤醒所有等待node的信号,然后信号自动往后传播
        doReleaseShared();
        return true;
    }
    return false;
}

总结

CountDownLatch在调用await的时候判断state释放为0,如果大于0则阻塞当前线程,将当前线程的node添加到队列中等待;在调用countDown时当遇到state减到0时,发出释放共享锁的信号,从头节点的后记节点开始往后传递信号,将队列等待的线程逐个唤醒并继续往下执行;
在这里state跟Reentrant的state独占锁含义不同,state的含义是由AQS的子类去描述的。

相关文章
|
2天前
|
人工智能 弹性计算 运维
|
23天前
|
Linux 程序员 数据格式
【2026最新】Notepad++下载、安装和使用一篇搞定(附中文版安装包)
Notepad++ 是一款免费开源、轻量高效的 Windows 文本编辑器,支持 C/Python/HTML 等 80+ 语言语法高亮、代码折叠、正则替换、编码转换及插件扩展,专为程序员与文本处理用户打造,完美替代系统记事本。(239字)
|
9天前
|
人工智能 缓存 安全
Claude Code 封号真实原因曝光,这次彻底不装了,直接针对国内开发者的账号下手?
Claude Code 封号潮背后:逆向扒出客户端隐写区域标记,Anthropic 政策收紧叠加 DeepSeek 7 月涨价,国产替代更紧迫。
|
14天前
|
人工智能 JSON 自然语言处理
让教学更智慧:用阿里云百炼工作流,自动生成中小学教材内容#小有可为#有温度的AI
通过可视化工作流编排,将大模型推理能力转化为标准化的教学内容生成引擎。教师只需输入教材标题和适用学段,即可自动获得结构完整、符合课程标准的章节内容,大幅降低备课门槛,助力教育资源均衡化。
501 127
|
18天前
|
存储 人工智能 监控
QoderWork完全指南:从入门到精通,把“AI实习生”变成你的全能工作搭档
阿里云2026年推出的桌面端AI工作助手QoderWork,不止聊天,更可动手干活:本地运行、安全可控,支持文件整理、数据分析、PPT生成、网页开发等;内置专家套件、多Agent协作与自定义Skills,让AI真正成为你身边的“AI实习生”。
|
7天前
|
人工智能 编解码 物联网
2026 最新Stable Diffusion 本地部署教程 下载安装使用详细图解(含官网安装包)
Stable Diffusion(SD)是2022年发布的开源文生图模型,由Stability AI等联合开发。支持文生图、图生图、局部重绘等,依托VAE降低算力需求,可在消费级显卡运行。本文提供秋葉aaaki制作的Windows整合包(含图形界面与插件),开箱即用,零配置启动。
|
9天前
|
人工智能 安全 程序员
终于,Claude Code 封号的原因被曝光了!竟然针对中国用户,植入隐形代码?!
通俗易懂地揭秘 Claude Code 封号的手段,分享一些自己对 AI 编程困境的思考,Codex、Cursor、DeepSeek、智谱 GLM、甚至是豆包,都有所行动了
496 1
|
10天前
|
人工智能 安全 Cloud Native
Higress 新发布:AI Gateway 能力增强,Gateway API 及其推理扩展持续打磨
增强 AI 网关能力,持续打磨 Gateway API 及其推理扩展。
412 125
|
17天前
|
人工智能 弹性计算 API
什么是 AlibabaCloud Agent Toolkit
Alibaba Cloud Agent Toolkit 是面向AI Agent的阿里云智能工具套件,集成OpenAPI、Terraform、CLI与文档能力,提供MCP插件、场景化Skills及执行审计机制,助AI准确查API、生成代码、规划架构、校验部署,实现安全、可靠、可追溯的云上智能运维。
438 2