我们在Java并发下常常使用synchronized进行线程同步操作。synchronized的语义是互斥锁，就是在同一时刻，只有一个线程能获得执行代码的锁。但是现实生活中，有好多的场景，锁不止一把。

由于synchronized是悲观锁，同一时刻只能有一个线程处于运行状态。对于同一时刻要有n个线程同时工作这种需求，用synchronized显然不合适。

查看Java并发工具，发现有一个Semaphore类，天生就是处理这种情况的。

什么是Semaphore

Semaphore是一种基于计数的信号量。它可以设定一个阈值，基于此，多个线程竞争获取许可信号，做完自己的申请后归还，超过阈值后，线程申请许可信号将会被阻塞。Semaphore可以用来构建一些对象池，资源池之类的，比如数据库连接池，我们也可以创建计数为1的Semaphore，将其作为一种类似互斥锁的机制，这也叫二元信号量，表示两种互斥状态。

看个例子，业务场景为模拟售票窗口售票，同一时刻有五个窗口能够提供服务。

package semaphore;

import java.util.concurrent.Semaphore;

public class Ticket {

    public static void main(String[] args) {
        Semaphore windows = new Semaphore(5);  // 声明5个窗口

        for (int i = 0; i < 8; i++) {
            new Thread() {
                @Override
                public void run() {
                    try {
                        windows.acquire();  // 占用窗口
                        System.out.println(Thread.currentThread().getName() + ": 开始买票");
                        sleep(2000);  // 睡2秒，模拟买票流程
                        System.out.println(Thread.currentThread().getName() + ": 购票成功");
                        windows.release();  // 释放窗口
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }.start();
        }
    }
}

运行结果

Thread-1: 开始买票
Thread-3: 开始买票
Thread-4: 开始买票
Thread-2: 开始买票
Thread-0: 开始买票
Thread-1: 购票成功
Thread-5: 开始买票
Thread-3: 购票成功
Thread-2: 购票成功
Thread-4: 购票成功
Thread-7: 开始买票
Thread-6: 开始买票
Thread-0: 购票成功
Thread-7: 购票成功
Thread-5: 购票成功
Thread-6: 购票成功

从结果来看，最多只有5个线程在购票。而这么精确的控制，我们也只是调用了acquire和release方法。下面看看是如何实现的。

源码分析

从acquire方法进去，又可以看到老套路：具体调用的还是 AbstractQueuedSynchronizer这个类的逻辑

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

而tryAcquireShared方法留给了子类去实现，Semaphore类里面的两个内部类FairSync和NonfairSync都继承自AbstractQueuedSynchronizer。这两个内部类，从名字来看，一个实现了公平锁，另一个是非公平锁。

这里多说一句，所谓公平和非公平是这个意思：假设现在有一个线程A在等待获取锁，这时候又来了线程B，如果这个时候B不考虑A的感受，也去申请锁，显然不公平；反之，只要A是先来的，B一定要排在A的后面，不能马上去申请锁，就是公平的。

Semaphore默认是调用了NonfairSync的tryAcquireShared方法，主要逻辑：

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

这又是一个经典的 CAS 操作加无限循环的算法，用来保证共享变量的正确性。另外，此处的getState()方法很是迷惑人，你以为是获取状态，实则不然。我们先看看Semaphore的构造方法：

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}
// 内部类
NonfairSync(int permits) {
    super(permits);
}
// 内部类，NonfairSync的父类
Sync(int permits) {
    setState(permits);
}

我们传进去的参数5，最终传给了setState方法，而getState和setState方法都在AbstractQueuedSynchronizer类里面

/**
* The synchronization state.
*/
private volatile int state;

protected final int getState() {
    return state;
}

protected final void setState(int newState) {
    state = newState;
}

也就是说父类定义了一个属性state，并配有final的get和set方法，子类只需要继承该属性，想代表什么含义都可以，比如Semaphore里面的内部类Sync就把这个属性当作最大允许访问的permits，像CountDownLatch和CyclicBarrier都是这么干的。

这种方式似乎不太好理解，为什么不是每个子类都定义自己的具有明确语义的属性，而是把控制权放在父类？？？我猜是出于安全的考虑。

再回到tryAcquireShared方法，这个方法是有参数的—int型的acquires，代表你要一次占几个坑。我们调用的无参的acquire方法，默认是传入1作为参数调用的这个方法，一次只申请一个坑。但是有的情况下，你可能一次需要多个。方法的返回值是剩余的坑的数量，如果数量小于0，执行AbstractQueuedSynchronizer这个类的doAcquireSharedInterruptibly方法。

/**
 * Acquires in shared interruptible mode.
 * @param arg the acquire argument
 */
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) {
                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);
    }
}

这个方法的逻辑与独占模式下的逻辑差不多。当所有的坑都被占着的时候，再来的线程都会被封装成节点，添加到等待的队列里面去。不同的是，这里的节点都是共享模式，而共享模式是实现多个坑同时提供服务的核心。

再来看看坑的释放，从release方法进去，核心逻辑在tryReleaseShared方法：

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");
        if (compareAndSetState(current, next))
            return true;
    }
}

API解析

Semaphore有两种模式，公平模式和非公平模式。公平模式就是调用acquire的顺序就是获取许可证的顺序，遵循FIFO；而非公平模式是抢占式的，也就是有可能一个新的获取线程恰好在一个许可证释放时得到了这个许可证，而前面还有等待的线程。

1.构造方法

Semaphore有两个构造方法，如下

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

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

从上面可以看到两个构造方法，都必须提供许可的数量，第二个构造方法可以指定是公平模式还是非公平模式，默认非公平模式。
Semaphore内部基于AQS的共享模式，所以实现都委托给了Sync类。

2.获取许可acquire()

先从获取一个许可看起，并且先看非公平模式下的实现。首先看acquire方法，acquire方法有几个重载，但主要是下面这个方法

public void acquire(int permits) throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    sync.acquireSharedInterruptibly(permits);
}

从上面可以看到，调用了Sync的acquireSharedInterruptibly方法，该方法在父类AQS中，如下：

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    //如果线程被中断了，抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //获取许可失败，将线程加入到等待队列中
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
} 

3.释放许可

释放许可也有几个重载方法，但都会调用下面这个带参数的方法
public void release(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.releaseShared(permits);
}

releaseShared方法在AQS中，如下

public final boolean releaseShared(int arg) {
    //如果改变许可数量成功
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

AQS子类实现共享模式的类需要实现tryReleaseShared类来判断是否释放成功，实现如下：

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改变许可数量成功，返回true
        if (compareAndSetState(current, next))
            return true;
    }
}       

从上面可以看到，一旦CAS改变许可数量成功，那么就会调用doReleaseShared()方法释放阻塞的线程.

4.减小许可数量

Semaphore还有减小许可数量的方法，该方法可以用于用于当资源用完不能再用时，这时就可以减小许可证。代码如下：

protected void reducePermits(int reduction) {
    if (reduction < 0) throw new IllegalArgumentException();
    sync.reducePermits(reduction);
}

可以看到，委托给了Sync，Sync的reducePermits方法如下

final void reducePermits(int reductions) {
    for (;;) {
        //得到当前剩余许可数量
        int current = getState();
        //得到减完之后的许可数量
        int next = current - reductions;
        if (next > current) // underflow
            throw new Error("Permit count underflow");
        //如果CAS改变成功
        if (compareAndSetState(current, next))
            return;
    }
}

从上面可以看到，就是CAS改变AQS中的state变量，因为该变量代表许可证的数量。

5.获取剩余许可数量　

Semaphore还可以一次将剩余的许可数量全部取走，该方法是drain方法，如下：

public int drainPermits() {
    return sync.drainPermits();
}

Sync的实现如下：

final int drainPermits() {
    for (;;) {
        int current = getState();
        if (current == 0 || compareAndSetState(current, 0))
            return current;
    }
}

可以看到，就是CAS将许可数量置为0。

实例代码

尝试获取许可方式

Semaphore semaphore = new Semaphore(3); // 最大允许10个并发
ExecutorService exec = Executors.newCachedThreadPool();

for (int i = 0; i < threadCount; i++) {
    final int threadNum = i;
    exec.execute(() -> {
        try {
            if (semaphore.tryAcquire(5000, TimeUnit.MILLISECONDS)) {
                test(threadNum);
                semaphore.release();
            }

        } catch (Exception e) {
            LOGGER.warning("execption:" + e);
        } 
    });
}
LOGGER.info("finish !!!");
exec.shutdown();

普通方式

Semaphore semaphore = new Semaphore(3); // 最大允许3个并发
    ExecutorService exec = Executors.newCachedThreadPool();

    for (int i = 0; i < threadCount; i++) {
        final int threadNum = i;
        exec.execute(() -> {
            try {
                semaphore.acquire(3);
                test(threadNum);
                semaphore.release(3);
            } catch (Exception e) {
                LOGGER.warning("execption:" + e);
            } 
        });
    }
    LOGGER.info("finish !!!");
    exec.shutdown();

指定了同一时刻获取许可个数，一共3个，每次三个，释放三个，相当于单线程

小结

Semaphore是信号量，用于管理一组资源。其内部是基于AQS的共享模式，AQS的状态表示许可证的数量，在许可证数量不够时，线程将会被挂起；而一旦有一个线程释放一个资源，那么就有可能重新唤醒等待队列中的线程继续执行。

所有的并发核心控制逻辑都在AbstractQueuedSynchronizer这个类中，只有理解了这个类的设计思路，才能真正理解衍生出来的工具类的实现原理。

出处https://blog.csdn.net/qq_19431333/article/details/70212663