Semaphore 信号量

基于 JDK 1.8 源码的 Semaphore 核心原理分析。

一、核心定位

1.1 一句话定义

Semaphore 是基于 AQS 共享模式（Shared Mode）实现的流量控制工具，用于限制同时访问特定资源的线程数量。

1.2 物理模型：停车场

停车场容量：3 个车位（permits = 3）

车辆1 进场 → permits = 2
车辆2 进场 → permits = 1
车辆3 进场 → permits = 0
车辆4 进场 → permits < 0，排队等待

车辆1 离场 → permits = 1 → 唤醒车辆4

核心语义：

• acquire()：进场，许可证减 1
• release()：离场，许可证加 1
• 许可证用完，后续线程排队

---

二、AQS 架构映射

2.1 State 的语义

public Semaphore(int permits) {
    sync = new NonfairSync(permits);  // 默认非公平
}

// Sync 构造函数
Sync(int permits) {
    setState(permits);  // state = 许可证数量
}

同步工具	state 含义
ReentrantLock	0=未锁定，>0=重入次数
CountDownLatch	剩余计数
Semaphore	剩余许可证数量（permits）

2.2 共享模式本质

Semaphore 允许多个线程同时 acquire 成功（只要 state > 0），这与互斥锁形成对比：

对比	ReentrantLock	Semaphore
AQS 模式	独占模式	共享模式
同时持有	只能 1 个线程	可多个线程
state 范围	0 或 >0	0 到 N

---

三、源码级原理解析

3.1 acquire() 的本质

public void acquire() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

nonfairTryAcquireShared 源码

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {  // CAS 自旋
        int available = getState();  // 当前剩余许可证
        int remaining = available - acquires;  // 计算剩余
        
        // 关键逻辑：
        // remaining < 0 → 返回负数 → AQS 将线程入队阻塞
        // remaining >= 0 且 CAS 成功 → 返回剩余值（非负）→ 获取成功
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

执行流程：

acquire(1)
    │
    ▼
remaining = state - 1
    │
    ├── remaining < 0  ← 许可证不足
    │       │
    │       └── 返回负数 → AQS 封装为 Shared Node 入队 → park 阻塞
    │
    └── remaining >= 0  ← 许可证充足
            │
            └── CAS 更新 state → 返回 remaining → 获取成功

公平模式差异

// FairSync.tryAcquireShared
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;
    }
}

3.2 release() 的本质

public void release() {
    sync.releaseShared(1);
}

tryReleaseShared 源码

protected final boolean tryReleaseShared(int releases) {
    for (;;) {  // CAS 自旋
        int current = getState();
        int next = current + releases;  // 归还许可证
        
        // 溢出检查
        if (next < current)
            throw new Error("Maximum permit count exceeded");
        
        if (compareAndSetState(current, next))
            return true;  // 返回 true → 触发 AQS 唤醒等待线程
    }
}

执行流程：

release(1)
    │
    ▼
next = state + 1
    │
    ▼
CAS 更新 state
    │
    ▼
返回 true
    │
    ▼
AQS.doReleaseShared()
    │
    ├── 唤醒 CLH 队头的 Shared Node
    │
    └── 传播机制：被唤醒的节点继续唤醒后续 Shared Node

3.3 传播机制

当一个线程释放许可证后，如果有多个等待线程，AQS 会通过传播机制（Propagate）依次唤醒：

// AQS.setHeadAndPropagate
private void setHeadAndPropagate(Node node, int propagate) {
    setHead(node);
    
    // propagate > 0 表示还有剩余许可证，继续唤醒
    if (propagate > 0 || ...) {
        Node s = node.next;
        if (s.isShared())
            doReleaseShared();  // 继续唤醒
    }
}

---

四、关键差异：Semaphore vs Lock

4.1 所有权（Ownership）

维度	Lock（ReentrantLock）	Semaphore
所有权	有，必须持有者释放	无所有权
释放约束	谁加锁谁解锁	任意线程可 release
exclusiveOwnerThread	记录持锁线程	不存在

Semaphore 无所有权的意义：

// 合法操作：线程A获取，线程B释放
Thread A:
    semaphore.acquire();
    // ...

Thread B:
    semaphore.release();  // 合法！

应用场景：

• 死锁恢复
• 跨线程协作
• 许可证转让

4.2 State 上限

维度	Lock	Semaphore
state 范围	0 或 >0（重入次数）	0 到 N（许可证数）
并发度	1（互斥）	N（可配置）

4.3 思维模型

维度	Lock	Semaphore
思维模型	钥匙（互斥访问）	车位/资源池
适用场景	临界区保护	流量控制、资源池限制

---

五、总结与思考

5.1 Semaphore 为什么"简单"

因为 AQS 承担了所有复杂工作：

AQS 负责	Semaphore 只需
CLH 队列管理	定义 state 语义
线程阻塞/唤醒	实现 tryAcquireShared
传播机制	实现 tryReleaseShared
公平/非公平队列	选择 Fair 或 Nonfair Sync

5.2 典型应用：数据库连接池限流

public class ConnectionPool {
    private final Semaphore semaphore;
    private final List<Connection> pool;
    
    public ConnectionPool(int poolSize) {
        this.semaphore = new Semaphore(poolSize);
        this.pool = new ArrayList<>();
        for (int i = 0; i < poolSize; i++) {
            pool.add(createConnection());
        }
    }
    
    public Connection acquire() throws InterruptedException {
        semaphore.acquire();  // 获取许可证
        synchronized (pool) {
            return pool.remove(0);
        }
    }
    
    public void release(Connection conn) {
        synchronized (pool) {
            pool.add(conn);
        }
        semaphore.release();  // 归还许可证
    }
}

5.3 核心要点速查

知识点	要点
底层实现	AQS 共享模式
state 语义	剩余许可证数量
acquire	state - 1，<0 则入队阻塞
release	state + 1，唤醒等待线程
所有权	无，任意线程可 release
传播机制	唤醒后继续唤醒后续 Shared Node