Semaphore 源码分析 - 《Java 多线程》

前言
源码分析
总结

前言

并发 JUC 包提供了很多工具类，比如 CountDownLatch 和 CyclicBarrier，今天说说这个 Semaphore——信号量。
Semaphore 是一个同步工具类，用来控制同时访问某个特定资源的操作数量。当线程池的工作队列被填满后，如果没有预定义的饱和策略来阻塞任务的执行，则可以通过信号量来限制任务的到达率。如果线程池使用无界队列缓冲任务，当任务在某一时刻增长数量过快，容易导致内存耗尽。
它的 acquire 方法返回一个虚拟的许可，如果没有可用的许可，则阻塞该方法的调用线程，直到有可用的许可为止。

源码分析

Sync

Semaphore 内部和其它同步工具包类似，也有一个 Sync 内部类，继承于 AQS。Sync 类又被细分为 FairSync 和 NonfairSync，即公平锁和非公平锁，最大的区别在于获取公平锁时，如果有其它线程先于当前线程加入 Sync Queue，那么当前线程就会等待，让其它线程先尝试获取锁。
Semaphore 源码分析 - 图1

Sync 构造函数

Sync(int permits) {
    setState(permits);
}

setState 方法是 AQS 内部的方法。

semaphore 构造器

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

两个构造器，第二个构造器有两个参数。第一个参数是许可线程数量，第二个参数标识是否为公平锁，默认非公平。
Semaphore 有 2 个重要的方法，也是经常使用的方法：

semaphore.acquire();
// doSomeing.....
semaphore.release();

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);
}

公平 tryAcquireShared 方法

protected int tryAcquireShared(int acquires) {
    for (;;) {
        // 如果有其它线程先于当前线程，则返回-1
        if (hasQueuedPredecessors())
            return -1;
        int available = getState();
        //得到剩余的许可数量
        int remaining = available - acquires;
        // 剩余数量不能小于 0，CAS 更换 state 值的操作必须成功，否则进入循环重试
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

非公平 tryAcquireShared 方法

// 这是抽象父类 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;
    }
}

从代码中我们可以看到，公平锁只比非公平锁多了两行代码，公平锁会主动判断 Sync Queue 是否有其它节点参与竞争。

acquireSharedInterruptibly 方法

当返回值小于 0 时，就得调用 AQS 的 acquireSharedInterruptibly 方法，加入 Sync Queue，阻塞当前线程。

// AbstractQueuedSynchronizer
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);
    }
}

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 变量 + releases
        int next = current + releases;
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        if (compareAndSetState(current, next))
            return true;
    }
}

doReleaseShared 方法

许可数量被减去的时候，调用 AQS 的 doReleaseShared 方法，唤醒下一个线程，并进行传播。
这里的行为和 ReentrantReadWriteLock, CountDownLatch 释放共享锁的逻辑一致。

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                // 如果当前节点是 SIGNAL 意味着，它正在等待一个信号，  
                // 或者说，它在等待转换状态，因此做两件事，1 是重置 waitStatus 标志位，2 是重置成功后, 唤醒下一个节点。
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);
            }
            // 如果本身头节点的 waitStatus 是出于重置状态（waitStatus==0）的，将其设置为“传播”状态。
            // 意味着需要将状态向后一个节点传播。
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        // 防止新加入节点，更新了首节点
        if (h == head)                   // loop if head changed
            break;
    }
}

这里有几个问题需要解释一下：

为什么需要把 waitStatus 设置为 0？

如果 waitStatus = Node.SIGNAL，说明这个节点就是头节点，需要唤醒下一个节点。但是为什么要把 waitStatus 设置为 0 呢？如果不把状态设置为 0，则自旋的时候有可能重复触发唤醒同一个节点。先原子性地把状态设置为 0，然后唤醒下一个节点，如果原子行为失败，则自旋，直到成功把状态设置为 0，然后唤醒下一个节点。

为什么要设置 Node.PROPAGATE

上一步把 waitStatus 设置为 0 后，只是为了方便唤醒下一个节点。把旧的头节点的状态设置为 Node.PROPAGATE，表明还要继续唤醒下个节点，当前旧头节点已经成为过去了。
释放共享锁的流程图：
Semaphore 源码分析 - 图2

总结

总的来看，Semaphore 就是一个共享锁，通过设置 state 变量来控制线程并发数量。
当调用 acquire 方法时，state 变量减 1，当调用 release 方法时，state 变量加 1。当 state 变量为 0 时，别的线程获取不到许可，不能往下执行，只能阻塞在 AQS 的 Sync Queue 中，等待有其它线程释放许可量，唤醒下个线程，然后继续争夺许可。