Java 并发系列（二）——Semaphore源码分析

Semaphore 通常叫它信号量, 可以用来控制同时访问特定资源的线程数量，也可以用来保持资源生产者和消费者之间的资源限制关系，通过协调各个线程，以保证合理的使用资源。Semaphore 具有公平和非公平两种模式，本文通过源码对 Semaphore 的实现做简单分析。

使用 Semaphore 控制最多有 2 个线程同时运行。

public class SemaphoreTest {
    private static AtomicInteger integer = new AtomicInteger(0);

    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(2);
        for (int i = 0; i < 6; i++) {
            new Thread(() -> {
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName() + "："
                                       + integer.incrementAndGet());
                    Thread.sleep((long) (Math.random() * 1000));
                    integer.decrementAndGet();
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, "Thread-" + i).start();
        }
    }
}

输出结果：

Thread-1：1
Thread-2：2
Thread-5：2
Thread-0：2
Thread-4：2
Thread-3：2

Semaphore 内部继承了 AbstractQueuedSynchronizer 类， 使用 state 来表示可用资源数，通过 AbstractQueuedSynchronizer 获取共享锁流程进行资源控制。

Semaphore 每获取一个资源 state 就会扣除，应该算是一个独占的场景，不是很理解为什么用共享锁流程，而不是用独占锁的流程。

Lock 中 state 表示锁的持有计数，成功获取锁后 state 增大，在 Semaphore 中成功获取资源之后，可用资源数减少，state 变小。

可以看到，tryReleaseShared 方法并没有判断当前是否是持有锁线程的相关逻辑，所以是允许在任意情况下执行来增加可用资源数的。

abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 1192457210091910933L;
    Sync(int permits) {
        // 初始化锁状态码，作为资源数
        setState(permits);
    }
    final int getPermits() {
        // 取得可用资源数
        return getState();
    }
    // 非公平的方式占用acquires个资源，资源数-1
    final int nonfairTryAcquireShared(int acquires) {
        for (;;) {
            int available = getState();
            int remaining = available - acquires;
            // 如果占用后资源数小于0，表示资源数已用尽，获取锁失败
            if (remaining < 0 ||
                // CAS修改state
                compareAndSetState(available, remaining))
                return remaining;
        }
    }
    // 释放releases个资源
    protected final boolean tryReleaseShared(int releases) {
        for (;;) {
            int current = getState();
            // 资源数相加
            int next = current + releases;
            // 资源数溢出
            if (next < current) // overflow
                throw new Error("Maximum permit count exceeded");
            // CAS修改state
            if (compareAndSetState(current, next))
                return true;
        }
    }
    // 减少reductions个资源
    final void reducePermits(int reductions) {
        for (;;) {
            int current = getState();
            //资源数相减
            int next = current - reductions;
            // 资源数溢出
            if (next > current) // underflow
                throw new Error("Permit count underflow");
            // CAS修改state
            if (compareAndSetState(current, next))
                return;
        }
    }
    // 返回可用资源数，并将可用资源数设置为0
    final int drainPermits() {
        for (;;) {
            int current = getState();
            // 可用资源数设置为0
            if (current == 0 || compareAndSetState(current, 0))
                return current;
        }
    }
}

非公平的实现方式

static final class NonfairSync extends Sync {
    private static final long serialVersionUID = -2694183684443567898L;
    NonfairSync(int permits) {
        super(permits);
    }
    protected int tryAcquireShared(int acquires) {
        return nonfairTryAcquireShared(acquires);
    }
}

公平的实现方式，相比于非公平的实现方式，公平的实现方式在获取资源前会先执行 hasQueuedPredecessors 判断当前线程是否是队列中第一个等待的节点。

static final class FairSync extends Sync {
    private static final long serialVersionUID = 2014338818796000944L;
    FairSync(int permits) {
        super(permits);
    }
    protected int tryAcquireShared(int acquires) {
        for (;;) {
            // 判断当前是否是队列首个等待的节点
            if (hasQueuedPredecessors())
                return -1;
            int available = getState();
            // 资源数相减
            int remaining = available - acquires;
            if (remaining < 0 ||
                // CAS修改资源数
                compareAndSetState(available, remaining))
                return remaining;
        }
    }
}

关于 AbstractQueuedSynchronizer 中共享锁和独占锁流程的实现，见锁系列文章。