前言
并发 JUC 包提供了很多工具类，比如之前说的 CountDownLatch，CyclicBarrier ，今天说说这个 Semaphore——信号量，关于他的使用请查看往期文章并发编程之 线程协作工具类，今天的任务就是从源码层面分析一下他的原理。
源码分析
如果先不看源码，根据以往我们看过的 CountDownLatch CyclicBarrier 的源码经验来看，Semaphore 会怎么设计呢？
首先，他要实现多个线程线程同时访问一个资源，类似于共享锁，并且，要控制进入资源的线程的数量。
如果根据 JDK 现有的资源，我们是否可以使用 AQS 的 state 变量来控制呢？类似 CountDownLatch 一样，有几个线程我们就为这个 state 变量设置为几，当 state 达到了阈值，其他线程就不能获取锁了，就需要等待。当 Semaphore 调用 release 方法的时候，就释放锁，将 state 减一，并唤醒 AQS 上的线程。
以上，就是我们的猜想，那我们看看 JDK 是不是和我们想的一样。
首先看看 Semaphore 的 UML 结构：
image.png
内部有 3 个类，继承了 AQS。一个公平锁，一个非公平锁，这点和 ReentrantLock 一摸一样。
看看他的构造器：
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
两个构造器，两个参数，一个是许可线程数量，一个是是否公平锁，默认非公平。
而 Semaphore 有 2 个重要的方法，也是我们经常使用的 2 个方法：
semaphore.acquire();
// doSomeing.....
semaphore.release();
acquire 和 release 方法，我们今天重点看这两个方法的源码，一窥 Semaphore 的全貌。
acquire 方法源码分析
代码如下：
public void acquire() throws InterruptedException {
// 尝试获取一个锁
sync.acquireSharedInterruptibly(1);
}
// 这是抽象类 AQS 的方法
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 如果小于0，就获取锁失败了。加入到AQS 等待队列中。
// 如果大于0，就直接执行下面的逻辑了。不用进行阻塞等待。
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
// 这是抽象父类 Sync 的方法，默认是非公平的
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
// 非公平锁的释放锁的方法
final int nonfairTryAcquireShared(int acquires) {
// 死循环
for (;;) {
// 获取锁的状态
int available = getState();
int remaining = available - acquires;
// state 变量是否还足够当前获取的
// 如果小于 0，获取锁就失败了。
// 如果大于 0，就循环尝试使用 CAS 将 state 变量更新成减去输入参数之后的。
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
这里的释放就是对 state 变量减一（或者更多）的。
返回了剩余的 state 大小。
当返回值小于 0 的时候，说明获取锁失败了，那么就需要进入 AQS 的等待队列了。代码入下：
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 添加一个节点 AQS 队列尾部
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
// 死循环
for (;;) {
// 找到新节点的上一个节点
final Node p = node.predecessor();
// 如果这个节点是 head，就尝试获取锁
if (p == head) {
// 继续尝试获取锁，这个方法是子类实现的
int r = tryAcquireShared(arg);
// 如果大于0，说明拿到锁了。
if (r >= 0) {
// 将 node 设置为 head 节点
// 如果大于0，就说明还有机会获取锁，那就唤醒后面的线程，称之为传播
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 如果他的上一个节点不是 head，就不能获取锁
// 对节点进行检查和更新状态，如果线程应该阻塞，返回 true。
if (shouldParkAfterFailedAcquire(p, node) &&
// 阻塞 park，并返回是否中断，中断则抛出异常
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
// 取消节点
cancelAcquire(node);
}
}
总的逻辑就是：
-
创建一个分享类型的 node 节点包装当前线程追加到 AQS 队列的尾部。
-
如果这个节点的上一个节点是 head ，就是尝试获取锁，获取锁的方法就是子类重写的方法。如果获取成功了，就将刚刚的那个节点设置成 head。
-
如果没抢到锁，就阻塞等待。
release 方法源码分析
该方法用于释放锁，代码如下：
public void release() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
// 死循环释放成功
if (tryReleaseShared(arg)) {
// 唤醒 AQS 等待对列中的节点，从 head 开始
doReleaseShared();
return true;
}
return false;
}
// Sync extends AbstractQueuedSynchronizer
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
// 对 state 变量 + 1
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
这里释放锁的逻辑写在了抽象类 Sync 中。逻辑简单，就是对 state 变量做加法。
在加法成功后，执行 doReleaseShared方法，这个方法是 AQS 的。
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
// 设置 head 的等待状态为 0 ，并唤醒 head 上的线程
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
// 成功设置成 0 之后，将 head 状态设置成传播状态
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
该方法的主要作用就是从 AQS 的 head 节点开始唤醒线程，注意，这里唤醒是 head 节点的下一个节点，需要和 doAcquireSharedInterruptibly方法对应，因为 doAcquireSharedInterruptibly 方法唤醒的当前节点的上一个节点，也就是 head 节点。
至此，释放 state 变量，唤醒 AQS 头节点结束。
总结
总结一下 Semaphore 的原理吧。
总的来说，Semaphore 就是一个共享锁，通过设置 state 变量来实现对这个变量的共享。当调用 acquire 方法的时候，state 变量就减去一，当调用 release 方法的时候，state 变量就加一。当 state 变量为 0 的时候，别的线程就不能进入代码块了，就会在 AQS 中阻塞等待。
网友评论