Semaphore

背景

Semaphore 用来控制同时访问某个特定资源的操作数量，或者同时执行某个指定操作的数量。信号量还可以用来实现某种资源池，或者对容器施加边界。

Semaphore管理着一组许可（permit），许可的初始数量可以通过构造函数设定，操作时要首先获得许可，才能进行操作，操作完成之后释放许可。如果没有获取许可，则阻塞直到有许可被释放。如果初始化一个许可为 1 的 Semaphore那就相当于初始化了一个不可重入的互斥锁。

源码分析

构造函数

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

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

两个构造方法都必须提供许可数量。第二个构造用法用于指定公平模式还是非公平模式。与 ReentrantLock 中的公平锁和非公平锁一样。公平信号量获取时，如果当前线程不在 CLH 队列的头部，则需要排队等候，对于非公平信号量而言，无论当前线程处于 CLH 队列的任何位置都可以直接获取。

获取许可

Semaphore 提供了四种获取许可的方法，分别如下：

//获取一个许可证(响应中断)，在没有可用的许可证时当前线程被阻塞。
public void acquire() throws InterruptedException {
   sync.acquireSharedInterruptibly(1);
}

//获取一个许可证(不响应中断)
public void acquireUninterruptibly() {
   sync.acquireShared(1);
}

//尝试获取许可证(非公平获取)，立即返回结果（非阻塞）。
public boolean tryAcquire() {
   return sync.nonfairTryAcquireShared(1) >= 0;
}

//尝试获取许可证(定时获取)
public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException {
   return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
    if (Thread.interrupted())//线程中断 说明信号量对线程中断敏感
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0) //获取信号量失败 线程进入同步队列自旋等待
        doAcquireSharedInterruptibly(arg);
}

tryAcquireShared 依赖的时 Sync 的实现，Sync 提供了公平和非公平的方式，先看非公平的方式。

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}
final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();//同步状态 当前的信号量许可数
        int remaining = available - acquires;//减去释放的信号量 剩余信号量许可数
        if (remaining < 0 ||//剩余信号量小于0 直接返回remaining 不做CAS
            compareAndSetState(available, remaining))//CAS更新
            return remaining;
    }
}

1. 获取 AQS 的 state。
2. 通过 state 计算剩余的信号量个数。
3. 如果还有剩余信号量，使用 CAS 尝试更新 state。

再看公平式的获取信号量。

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

最后来看一下如果没有获取到信号量时的处理方法 doAcquireSharedInterruptibly。

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) {//当前节点的前置节点是AQS的头节点 即自己是AQS同步队列的第一个节点
                int r = tryAcquireShared(arg); //再去获取信号量
                if (r >= 0) {//获取成功
                    setHeadAndPropagate(node, r);//退出自旋
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node); //获取失败 就取消获取
    }
}

释放许可

释放信号量的方法：

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

调用的是 AQS 的 releaseShared 方法：

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

releaseShared 由子类 Sync 实现：

protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        int current = getState();//当前信号量许可数
        int next = current + releases; //当前信号量许可数+释放的信号量许可数
        if (next < current) 
            throw new Error("Maximum permit count exceeded");
        if (compareAndSetState(current, next))//CAS更新当前信号量许可数
            return true;
    }
}

释放成功后，就继续调用 doReleaseShared，唤醒后续的节点来争取信号量了。

private void doReleaseShared() {
    for (;;) {
        Node h = head; //头节点
        if (h != null && h != tail) {//同步队列中存在线程等待
            int ws = h.waitStatus; //头节点线程状态
            if (ws == Node.SIGNAL) {//头节点线程状态为SIGNAL 唤醒后续线程节点
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h); //唤醒下个节点
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

CountDownLatch

背景

CountDownLatch 是并发包中用来控制一个或者多个线程等待其他线程完成操作的并发工具类。

比如：游戏一开始需要加载一些基础数据后才能开始，基础数据加载完毕后可以继续游戏服务，这个流程就可以使用 CountDownLatch 来实现。

总结 CountDownLatch 的作用就阻塞其他线程直到条件允许以后才释放该阻塞。

源码分析

await 方法

调用 await 方法会阻塞当前线程直到计数器的数值为 0，方法如下：

public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1); //共享式获取AQS的同步状态
}

调用 AQS 的 acquireSharedInterruptibly 方法：

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
    if (Thread.interrupted())//线程中断 说明闭锁对线程中断敏感
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0) //闭锁未使用完成 线程进入同步队列自旋等待 
        doAcquireSharedInterruptibly(arg);
}

tryAcquireShared 在 CountDownLatch 的 Sync 只提供了一种实现，那就是根据获取到的 state 是否为 0 来判断。

protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1; //AQS的同步状态为0则闭锁结束 可以进行下一步操作
}

doAcquireSharedInterruptibly 就是加入等待队列进行自旋，与 Semaphore 是一样的。

countDown 方法

调用 countDown 方法会将计数器的数值减 1 直到计数器为 0。

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

和 Semaphore 一样，调用的是 AQS 的 releaseShared 方法：

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {//减少闭锁的计数器
        doReleaseShared();//唤醒后续线程节点
        return true;
    }
    return false;
}

CountDownLatch 的 Sync 的 tryReleaseShared 只提供了一种方式，作用就是将 state 的值减去 1。

protected boolean tryReleaseShared(int releases) {
    // Decrement count; signal when transition to zero
    for (;;) {
        int c = getState();
        if (c == 0)
            return false; //计数器已经是0了
        int nextc = c-1; //计数器减1
        if (compareAndSetState(c, nextc)) //CAS更新同步状态
            return nextc == 0;
    }
}

CountDownLatch 类使用 AQS 同步状态来表示计数。在 await 时，所有的线程进入同步队列自旋等待，在countDown 时，获取闭锁成功的线程会减少闭锁的计数器，同时唤醒后续线程取获取闭锁，直到 await 中的计数器为 0，获取到闭锁的线程才可以通过，执行下一步操作。