【JUC并发 | 第九篇】Semaphore 和 CountDownLatch

目录SemaphoreSemaphore 原理CountDownLatchCountDownLatch 作用CountDownLatch 原理【JUC并发 | 第八篇】AQS的底层原理https://blog.csdn.net/h52412224/article/details/159159242?spm1001.2014.3001.5502【JUC并发 | 第七篇】简析Future 和 CompletableFuture类https://blog.csdn.net/h52412224/article/details/159078192【JUC并发 | 第六篇】深入理解线程池https://blog.csdn.net/h52412224/article/details/159043558?spm1001.2014.3001.5502SemaphoreSemaphore 原理一、核心定位Semaphore 是 Java 并发包中用于控制并发访问线程数的同步工具底层完全基于 AQS 的共享模式实现——核心是通过 AQS 的state变量表示“可用许可数”利用 AQS 共享模式的线程排队、阻塞/唤醒机制实现多线程对许可的获取与释放。二、Semaphore底层核心实现1. 核心关联Semaphore与AQS的映射关系Semaphore 内部封装了一个继承 AQS 的同步器类Sync并分为公平版FairSync和非公平版NonfairSync核心映射关系Semaphore概念AQS底层实现许可数AQS 的state变量初始值由 Semaphore 构造函数指定如new Semaphore(3)则state3获取许可调用 AQS 的acquireShared()核心是tryAcquireShared()CAS 减少 state释放许可调用 AQS 的releaseShared()核心是tryReleaseShared()CAS 增加 state线程排队阻塞竞争许可失败的线程封装为 Node 节点加入 AQS 的 CLH 队列等待被唤醒2. 核心方法的底层实现1获取许可acquire()方法Semaphore 的acquire()本质是调用 AQS 的acquireSharedInterruptibly(1)核心逻辑在tryAcquireShared()中非公平版默认// NonfairSync 的 tryAcquireShared 实现 protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires); } final int nonfairTryAcquireShared(int acquires) { for (;;) { // 自旋 CAS // 1. 获取当前可用许可数 int available getState(); // 2. 计算获取后剩余许可数 int remaining available - acquires; // 3. CAS 修改 state剩余许可≥0 则修改成功否则失败 if (remaining 0 || compareAndSetState(available, remaining)) { return remaining; } } }返回值含义≥0表示获取许可成功0表示失败进入 AQS 队列阻塞。公平版仅多一步“检查队列是否有等待线程”避免新线程插队protected int tryAcquireShared(int acquires) { for (;;) { // 公平性检查如果队列有等待线程直接返回-1放弃竞争 if (hasQueuedPredecessors()) { return -1; } // 后续逻辑与非公平版一致 int available getState(); int remaining available - acquires; if (remaining 0 || compareAndSetState(available, remaining)) { return remaining; } } }2释放许可release()方法Semaphore 的release()调用 AQS 的releaseShared(1)核心逻辑在tryReleaseShared()中protected final boolean tryReleaseShared(int releases) { for (;;) { // 自旋 CAS // 1. 获取当前许可数 int current getState(); // 2. 计算释放后许可数 int next current releases; // 边界检查避免许可数溢出 if (next current) { throw new Error(Maximum permit count exceeded); } // 3. CAS 修改 state 成功则返回 true唤醒队列等待线程 if (compareAndSetState(current, next)) { return true; } } }释放许可成功后AQS 会唤醒 CLH 队列中等待的线程让它们重新竞争许可。3.Semaphore核心执行流程以new Semaphore(2)初始 2 个许可为例拆解完整流程场景线程 A、B、C 同时调用 acquire() 获取 1 个许可。 步骤 1线程 A 执行 tryAcquireShared(1)CAS 将 state 从 2→1返回 1≥0获取许可成功 步骤 2线程 B 执行 tryAcquireShared(1)CAS 将 state 从 1→0返回 0≥0获取许可成功 步骤 3线程 C 执行 tryAcquireShared(1)计算 remaining0-1-1CAS 失败返回 -1 步骤 4线程 C 进入 AQS 的 CLH 队列调用 LockSupport.park() 阻塞 步骤 5线程 A 执行 release()CAS 将 state 从 0→1返回 trueAQS 唤醒队列中的线程 C 步骤 6线程 C 被唤醒重新执行 tryAcquireShared(1)CAS 将 state 从 1→0获取许可成功退出队列。四、补充Semaphore与 ReentrantLock 的底层差异特性SemaphoreReentrantLockAQS模式共享模式独占模式state 含义可用许可数锁的重入次数竞争结果多线程可同时获取资源仅一个线程能获取资源核心方法tryAcquireShared/releaseSharedtryAcquire/release总结Semaphore信号量的底层完全基于 AQS 的共享模式实现核心逻辑如下许可数管理Semaphore 的“可用许可数”对应 AQS 的state变量初始许可数由构造函数指定获取许可调用acquire()时底层执行 AQS 的acquireShared()通过tryAcquireShared()自旋 CAS 减少state——剩余许可≥0 则获取成功否则线程入队阻塞公平版会先检查队列是否有等待线程非公平版直接抢许可释放许可调用release()时底层执行 AQS 的releaseShared()通过tryReleaseShared()自旋 CAS 增加state成功后唤醒 CLH 队列中等待的线程让其重新竞争许可核心特点支持多线程同时获取许可共享模式可灵活控制并发访问的线程数常用于限流、资源池控制等场景。CountDownLatchCountDownLatch 作用CountDownLatch 原理一、核心定位CountDownLatch 是 Java 并发包中用于等待多个线程完成任务的同步工具底层完全基于 AQS 的共享模式实现——核心是通过 AQS 的state变量表示“剩余待完成的任务数”利用 AQS 共享模式的线程排队、阻塞/唤醒机制实现“主线程等待所有子线程完成后再执行”的逻辑。二、CountDownLatch 底层核心实现1. 核心关联CountDownLatch 与AQS的映射关系CountDownLatch 内部封装了一个继承 AQS 的同步器类Sync无公平/非公平之分核心映射关系CountDownLatch 概念AQS底层实现倒计时数AQS 的state变量由构造函数指定如new CountDownLatch(3)则state3等待线程阻塞调用await()时底层执行 AQS 的acquireSharedInterruptibly(1)竞争失败则入队阻塞计数器减 1调用countDown()时底层执行 AQS 的releaseShared(1)CAS 减少 state唤醒等待线程当 state 减到 0 时AQS 唤醒所有阻塞在队列中的线程2. 核心方法的底层实现1初始化CountDownLatch(int count)构造函数直接将传入的“倒计时数”赋值给 AQS 的state且state一旦初始化无法重置CountDownLatch 是一次性的public CountDownLatch(int count) { if (count 0) throw new IllegalArgumentException(count 0); this.sync new Sync(count); // Sync 是内部 AQS 子类 } // 内部 Sync 类的构造 Sync(int count) { setState(count); // 调用 AQS 的 setState() 给 state 赋值 }2等待线程阻塞await() 方法await()是 CountDownLatch 的核心等待方法底层调用 AQS 的acquireSharedInterruptibly(1)核心逻辑在tryAcquireShared()中// CountDownLatch 的 await() public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } // 内部 Sync 重写的 tryAcquireShared核心逻辑 protected int tryAcquireShared(int acquires) { // 核心规则state0 时返回 1获取成功无需阻塞否则返回 -1失败入队阻塞 return (getState() 0) ? 1 : -1; }当线程调用await()时会循环执行tryAcquireShared()如果state≠0任务未完成返回 -1线程被封装为 Node 节点加入 AQS 的 CLH 队列调用LockSupport.park()阻塞如果state0所有任务完成返回 1线程直接执行无需阻塞。3计数器减 1countDown() 方法countDown()是触发倒计时的核心方法底层调用 AQS 的releaseShared(1)核心逻辑在tryReleaseShared()中// CountDownLatch 的 countDown() public void countDown() { sync.releaseShared(1); } // 内部 Sync 重写的 tryReleaseShared核心逻辑 protected boolean tryReleaseShared(int releases) { for (;;) { // 自旋 CAS 保证原子性多线程同时 countDown 不冲突 int c getState(); // 1. 如果 state 已经为 0直接返回 false无需处理 if (c 0) return false; // 2. 计数器减 1 int nextc c - 1; // 3. CAS 修改 state保证原子性 if (compareAndSetState(c, nextc)) { // 关键只有当 state 减到 0 时返回 true触发 AQS 唤醒所有等待线程 return nextc 0; } } }每次调用countDown()都会通过 CAS 将state减 1只有当state从 1 减到 0 时tryReleaseShared()返回 trueAQS 才会唤醒 CLH 队列中所有阻塞的线程。3. CountDownLatch 核心执行流程以new CountDownLatch(2)等待 2 个子线程完成为例拆解完整流程场景主线程调用 await() 等待子线程 1、子线程 2 执行任务后调用 countDown()。 步骤 1CountDownLatch 初始化state2 步骤 2主线程调用 await()执行 tryAcquireShared()state2≠0返回 -1主线程入队阻塞 步骤 3子线程 1 完成任务调用 countDown()CAS 将 state 从 2→1nextc1≠0返回 false无唤醒操作 步骤 4子线程 2 完成任务调用 countDown()CAS 将 state 从 1→0nextc0返回 true 步骤 5AQS 收到返回的 true唤醒队列中阻塞的主线程 步骤 6主线程被唤醒重新执行 tryAcquireShared()state0返回 1主线程退出阻塞继续执行。public class CountDownLatchDemo { public static void main(String[] args) throws InterruptedException { // 初始化计数器值为3代表3个任务 CountDownLatch latch new CountDownLatch(3); // 启动3个线程执行任务 for (int i 1; i 3; i) { new Thread(() - { try { System.out.println(Thread.currentThread().getName() 执行任务); Thread.sleep(1000); // 模拟任务执行时间 } finally { // 任务完成计数器减1 latch.countDown(); } }, 线程 i).start(); } // 主线程等待所有任务完成 latch.await(); System.out.println(所有子任务完成主线程执行汇总逻辑); } }总结CountDownLatch 的底层完全基于 AQS 的共享模式实现核心逻辑如下计数器管理CountDownLatch 的“倒计时数”对应 AQS 的state变量由构造函数初始化且一旦初始化无法重置等待阻塞调用await()时底层执行 AQS 的acquireSharedInterruptibly()通过tryAcquireShared()判断 state 是否为 0——非 0 则线程入 AQS 队列阻塞0 则直接执行倒计时触发调用countDown()时底层执行 AQS 的releaseShared()通过自旋 CAS 将 state 减 1只有当 state 减到 0 时才返回 true触发 AQS 唤醒所有阻塞的等待线程核心特点CountDownLatch 是一次性的state 减到 0 后无法重置支持多线程等待适用于“主线程等待多个子线程完成任务”的场景。补充CountDownLatch 与 Semaphore 的底层差异特性CountDownLatchSemaphorestate 含义剩余倒计时数0 触发唤醒可用许可数核心逻辑等待 state 变为 0竞争有限的许可数复用性一次性state 到 0 失效可复用释放后许可恢复唤醒逻辑state0 时唤醒所有线程释放许可时唤醒单个线程上述内容也同步在我的飞书欢迎访问https://my.feishu.cn/wiki/QLauws6lWif1pnkhB8IcAvkhncc?fromfrom_copylink如果我的内容对你有帮助请点赞评论收藏。创作不易你们的支持就是我坚持下去的动力