Semaphore源码剖析:基于AQS的并发控制艺术-夜雨聆风

Semaphore源码剖析:基于AQS的并发控制艺术

大家好！今天我们来探讨Java并发编程中一个非常重要的工具 —— Semaphore（信号量）。

作为JUC包中的核心组件，Semaphore在多线程协作中扮演着流量控制器的角色。通过本文，你将彻底理解Semaphore如何基于AQS（AbstractQueuedSynchronizer）实现高效的并发控制。

1. Semaphore是什么？

想象一下停车场的管理场景：停车场有5个车位，门口有个门卫，手中有5把钥匙对应5个车位。来一辆车，门卫给一把钥匙；车离开时，司机归还钥匙。如果同时来了100辆车，门卫只能放5辆进入，其他车辆必须等待。

在这个比喻中：

门卫 = Semaphore
车钥匙 = 许可证（Permit）
车辆 = 线程
车位 = 共享资源

Semaphore就是这样一个”许可证管理器”，它维护了一定数量的许可证，线程在访问共享资源前必须先获取许可证，使用完毕后归还许可证。

2. Semaphore的核心设计：基于AQS共享模式

2.1 AQS共享模式 vs 独占模式

要理解Semaphore，必须先了解AQS的两种同步模式：

模式	特点	典型实现
独占模式(Exclusive)	资源只有1份，同一时间只能1个线程持有	ReentrantLock
共享模式(Shared)	资源有N份，同一时间可以有多个线程同时持有	Semaphore、CountDownLatch

Semaphore正是基于AQS的共享模式实现的。

2.2 Semaphore的类结构

public class Semaphore implements java.io.Serializable {    private final Sync sync;    abstract static class Sync extends AbstractQueuedSynchronizer {        Sync(int permits) {            setState(permits);  // permits = 许可证数量        }    }    // 非公平同步器（默认）    static final class NonfairSync extends Sync {        protected int tryAcquireShared(int acquires) {            return nonfairTryAcquireShared(acquires);        }    }    // 公平同步器    static final class FairSync extends Sync {        protected int tryAcquireShared(int acquires) {            // 先判断队列，再抢锁        }    }}

Semaphore内部维护一个Sync对象，而Sync继承自AQS。构造Semaphore时传入的permits参数，实际上就是设置AQS的state值。

3. 核心源码解析

3.1 获取许可证：acquire()的核心逻辑

让我们深入分析用户提到的关键问题：当state-permits为负数时，线程如何被封装成节点加入AQS等待队列并被park挂起？

// Semaphore的acquire()方法public void acquire() throws InterruptedException {    sync.acquireSharedInterruptibly(1);}// AQS的acquireSharedInterruptibly方法public final void acquireSharedInterruptibly(int arg) throws InterruptedException {    if (Thread.interrupted())        throw new InterruptedException();    if (tryAcquireShared(arg) < 0)  // 尝试获取许可证        doAcquireSharedInterruptibly(arg);  // 获取失败，进入队列等待}

关键在tryAcquireShared方法，这是Semaphore实现的核心：

// 非公平模式获取许可证finalintnonfairTryAcquireShared(int acquires){    for (;;) {        int available = getState();      // 获取当前可用许可证数量        int remaining = available - acquires;  // 计算剩余许可证        if (remaining < 0 ||             // 如果剩余许可证小于0            compareAndSetState(available, remaining))  // 或者CAS设置成功            return remaining;            // 返回剩余数量    }}

核心逻辑解析：

计算remaining：remaining = state - acquires
判断条件：

如果remaining < 0：直接返回负数，表示获取失败
如果remaining >= 0：尝试CAS更新state值，成功则返回remaining

当remaining为负数时会发生什么？

这正是用户描述的场景：当初始化Semaphore(0)时，state=0。第一个线程调用acquire()，计算remaining = 0-1 = -1，小于0，直接返回-1。

此时tryAcquireShared返回负数，触发doAcquireSharedInterruptibly方法，线程被封装成Node节点加入AQS等待队列，并调用LockSupport.park()挂起。

3.2 释放许可证：release()的唤醒机制

// Semaphore的release()方法public void release() {    sync.releaseShared(1);}// AQS的releaseShared方法public final boolean releaseShared(int arg) {    if (tryReleaseShared(arg)) {     // 尝试释放许可证        doReleaseShared();           // 释放成功，唤醒等待线程        return true;    }    return false;}// Semaphore的tryReleaseShared实现protected final boolean tryReleaseShared(int releases) {    for (;;) {        int current = getState();        int next = current + releases;  // state增加        if (compareAndSetState(current, next))            return true;    }}

唤醒流程：

state增加：next = current + releases
CAS更新：成功更新state值
唤醒线程：调用doReleaseShared()唤醒等待队列中的线程

被唤醒的线程会重新尝试获取许可证，再次计算remaining = state - acquires。只要此时remaining >= 0，线程就能成功获取许可证并继续执行。

3.3 AQS等待队列的节点封装

当线程获取许可证失败时，AQS会执行以下操作：

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) {                int r = tryAcquireShared(arg);  // 再次尝试获取                if (r >= 0) {                    setHeadAndPropagate(node, r);  // 设置头节点并传播唤醒                    p.next = null;                    failed = false;                    return;                }            }            // 检查是否需要阻塞            if (shouldParkAfterFailedAcquire(p, node) &&                parkAndCheckInterrupt())  // 挂起线程                throw new InterruptedException();        }    } finally {        if (failed)            cancelAcquire(node);    }}

共享模式的关键特性：传播唤醒

与独占模式不同，共享模式在唤醒时具有传播特性。当一个共享节点被唤醒并成功获取资源后，它会检查是否还有剩余资源，如果有，会继续唤醒后面的共享节点。

private void setHeadAndPropagate(Node node, int propagate) {    Node h = head;    setHead(node);    // 判断是否需要传播唤醒    if (propagate > 0 || h == null || h.waitStatus < 0 ||        (h = head) == null || h.waitStatus < 0) {        Node s = node.next;        if (s == null || s.isShared())            doReleaseShared();  // 唤醒后继共享节点    }}

4. 公平 vs 非公平模式

4.1 非公平模式（默认）

// 非公平同步器static final class NonfairSync extends Sync {    protected int tryAcquireShared(int acquires) {        return nonfairTryAcquireShared(acquires);    }}

非公平模式的特点：线程直接尝试获取许可证，不考虑等待队列中是否有其他线程在等待。这可能导致”插队”现象，但吞吐量更高。

4.2 公平模式

// 公平同步器static final class FairSync extends Sync {    protected int tryAcquireShared(int acquires) {        for (;;) {            if (hasQueuedPredecessors())  // 先检查队列中是否有等待线程                return -1;  // 有等待线程，必须排队            int available = getState();            int remaining = available - acquires;            if (remaining < 0 ||                compareAndSetState(available, remaining))                return remaining;        }    }}

公平模式的特点：在尝试获取许可证前，先检查AQS队列中是否有其他线程在等待。如果有，当前线程必须排队，遵循FIFO原则。

选择建议：

大多数业务场景：使用非公平模式（默认），吞吐量更高
需要严格避免线程饥饿：使用公平模式，保证先来先服务

5. 使用Semaphore的正确姿势

5.1 基本使用模板

Semaphore semaphore = new Semaphore(20);  // 最多20个线程同时执行try {    semaphore.acquire();      // 获取许可证    // 执行业务逻辑} finally {    semaphore.release();      // 必须释放许可证}

5.2 常见陷阱与解决方案

陷阱1：忘记释放许可证

// 错误示例：获取后忘记释放semaphore.acquire();// 业务逻辑// 忘记调用semaphore.release();// 正确做法：使用try-finally确保释放try {    semaphore.acquire();    // 业务逻辑} finally {    semaphore.release();}

陷阱2：获取许可失败时也释放

// 错误示例：获取失败时也释放（会导致许可证数量增加）boolean acquireSuccess = false;try {    semaphore.acquire();    acquireSuccess = true;    // 业务逻辑} finally {    if (acquireSuccess) {  // 必须检查是否获取成功        semaphore.release();    }}

陷阱3：误用为QPS限流工具

Semaphore控制的是同时执行的线程数，不是每秒请求数。如果需要QPS限流，应该使用令牌桶或漏桶算法。

5.3 超时获取与中断处理

// 尝试在指定时间内获取许可if (semaphore.tryAcquire(1, TimeUnit.SECONDS)) {    try {        // 获取成功，执行业务逻辑    } finally {        semaphore.release();    }} else {    // 超时未获取到许可，执行降级逻辑}

6. 实战应用场景

6.1 接口限流

// 限制最多20个并发请求private Semaphore apiSemaphore = new Semaphore(20);public Response callApi(Request request) {    try {        apiSemaphore.acquire();        return doCallApi(request);    } finally {        apiSemaphore.release();    }}

6.2 任务并发控制

// 控制同时执行的任务数量ExecutorService executor = Executors.newFixedThreadPool(100);Semaphore taskSemaphore = new Semaphore(10);  // 最多10个任务同时执行for (int i = 0; i < 1000; i++) {    executor.submit(() -> {        try {            taskSemaphore.acquire();            // 执行任务        } finally {            taskSemaphore.release();        }    });}

7. 性能优化建议

合理设置许可证数量：根据系统资源和业务需求设置合适的permits值
优先使用非公平模式：在大多数场景下，非公平模式性能更好
避免在锁内获取许可证：先获取许可证，再进入同步块，避免死锁
使用tryAcquire避免长时间阻塞：设置合理的超时时间，避免线程长时间等待

8. 总结

通过本文的深度剖析，我们了解了：

Semaphore的核心机制：基于AQS共享模式，通过state管理许可证数量
获取许可证流程：计算remaining值，负数时线程入队挂起
释放许可证流程：增加state值，唤醒等待线程
公平与非公平区别：是否检查等待队列
正确使用姿势：try-finally确保释放，避免常见陷阱

Semaphore作为JUC并发工具包中的重要组件，其设计体现了AQS框架的强大和灵活。理解Semaphore的底层实现，不仅有助于我们更好地使用这个工具，更能深入理解Java并发编程的精髓。

记住：并发控制不是限制，而是为了更好的协作。合理使用Semaphore，让你的多线程程序既高效又稳定！