【搞定Java并发编程】第26篇：Java中的并发工具类之控制并发线程数的 Semaphore

比眉伴天荒 2022-04-08 14:12 204阅读 0赞

### 上一篇：[Java中的并发工具类之同步屏障 CyclicBarrier][Java_ CyclicBarrier] ###

## **本文目录：** ##

**1、获取许可证**

**2、释放许可证**

--------------------

本文转载自：[https://mp.weixin.qq.com/s/LS8YBKpiJnHEY1kMWmwoxg][https_mp.weixin.qq.com_s_LS8YBKpiJnHEY1kMWmwoxg]

推荐阅读：**[剖析基于并发AQS的共享锁的实现(基于信号量Semaphore)][AQS_Semaphore]【写的非常好】**

**Semphore（信号量）是用来控制同时访问特定资源的线程数量，它通过协调各个线程，以保证合理的使用公共资源。**

Semaphore是JUC包中比较常用到的一个类，它是**AQS共享模式**的一个应用，可以允许多个线程同时对共享资源进行操作，并且可以有效的控制并发数，利用它可以很好的实现流量控制。

Semaphore提供了一个**许可证**的概念，可以把这个许可证看作公共汽车车票，只有成功获取车票的人才能够上车，并且车票是有一定数量的，不可能毫无限制的发下去，这样就会导致公交车超载。所以当车票发完的时候(公交车以满载)，其他人就只能等下一趟车了。如果中途有人下车，那么他的位置将会空闲出来，因此如果这时其他人想要上车的话就又可以获得车票了。

利用Semaphore可以实现各种池，我们在本篇末尾将会动手写一个简易的数据库连接池。

我们先看下Semaphore这个类的整体内部结构：

public class Semaphore implements java.io.Serializable {
        
            private final Sync sync;
    
            abstract static class Sync extends AbstractQueuedSynchronizer {...}
    	
    	static final class NonfairSync extends Sync {...}
    
            static final class FairSync extends Sync {...}
    	
    	public Semaphore(int permits) {
                sync = new NonfairSync(permits);
            }
    	
    	public Semaphore(int permits, boolean fair) {
                sync = fair ? new FairSync(permits) : new NonfairSync(permits);
            }
    	
    	// 获取许可证
    	public void acquire() throws InterruptedException {...}
    	
    	// 尝试获取许可证
    	public boolean tryAcquire() {...}
    	
    	// 归还许可证
    	public void release() {...}
    	
    	// 获取permits个许可证
    	public void acquire(int permits) throws InterruptedException {...}
    	
    	// 尝试获取permits个许可证
    	public boolean tryAcquire(int permits) {...}
    	
    	// 归还permits个许可证
    	public void release(int permits) {...}
    	
    	// 返回此信号量中当前可用的许可证数
    	public int availablePermits() {...}
    	
    	public int drainPermits() {...}
    	
    	// 减少reduction个许可证
    	protected void reducePermits(int reduction) {...}
    	 
    	// 是否有线程正在等待许可证 
    	public final boolean hasQueuedThreads() {...}
    	
    	// 返回正在等待许可证的线程数
    	public final int getQueueLength() {...}
    	
    	// 返回所有等待获取许可证的线程集合
    	protected Collection<Thread> getQueuedThreads() {...}
    }

可以得出下面关于 Semaphore 的类关系图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70][]

从上面的Semaphore类结构中可用发现它其实也是 AQS 中共享锁的使用，因为每个线程共享一个池嘛。

套路解读：创建 Semaphore 实例的时候，需要一个参数 permits，这个基本上可以确定是设置给 AQS 的 state 的，然后每个线程调用 acquire 的时候，执行 state = state - 1，release 的时候执行 state = state + 1，当然，acquire 的时候，如果 state = 0，说明没有资源了，需要等待其他线程 release。

首先我们来看一下Semaphore的构造器：

// 构造器1
    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
    
    // 构造器2
    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

Semaphore提供了两个带参构造器，没有提供无参构造器。这两个构造器都必须传入一个初始的许可证数量，使用构造器1构造出来的信号量在获取许可证时会采用非公平方式获取，使用构造器2可以通过参数指定获取许可证的方式(公平or非公平)。

Semaphore主要对外提供了两类API，**获取许可证**和**释放许可证**，默认的是获取和释放一个许可证，也可以传入参数来同时获取和释放多个许可证。在本篇中我们只讲每次获取和释放一个许可证的情况。

## 1、获取许可证 ##

// 获取一个许可证(响应中断)
    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));
    }

上面的API是Semaphore提供的默认获取许可证操作。每次只获取一个许可证，这也是现实生活中较常遇到的情况。除了直接获取还提供了尝试获取，直接获取操作在失败之后可能会阻塞线程，而尝试获取则不会。另外还需注意的是tryAcquire方法是使用非公平方式尝试获取的。在平时我们比较常用到的是acquire方法去获取许可证。下面我们就来看看它是怎样获取的。可以看到acquire方法里面直接就是调用sync.acquireSharedInterruptibly，这个方法是AQS里面的方法，我们简单讲一下。

acquireSharedInterruptibly方法首先就是去调用tryAcquireShared方法去尝试获取，tryAcquireShared在AQS里面是抽象方法，FairSync和NonfairSync这两个派生类实现了该方法的逻辑。FairSync实现的是公平获取的逻辑，而NonfairSync实现的非公平获取的逻辑。

abstract static class Sync extends AbstractQueuedSynchronizer {
        // 非公平方式尝试获取
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                // 获取可用许可证
                int available = getState();
                // 获取剩余许可证
                int remaining = available - acquires;
                // 1.如果remaining小于0则直接返回remaining
                // 2.如果remaining大于0则先更新同步状态再返回remaining
                if (remaining < 0 || compareAndSetState(available, remaining)) {
                    return remaining;
                }
            }
        }
    }
    
    // 非公平同步器
    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);
        }
    }
    
    // 公平同步器
    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()) {
                    // 如果有的话就直接返回-1，表示尝试获取失败
                    return -1;
                }
                // 获取可用许可证
                int available = getState();
                // 获取剩余许可证
                int remaining = available - acquires;
                // 1.如果remaining小于0则直接返回remaining
                // 2.如果remaining大于0则先更新同步状态再返回remaining
                if (remaining < 0 || compareAndSetState(available, remaining)) {
                    return remaining;
                }
            }
        }
    }

里需要注意的是NonfairSync的tryAcquireShared方法直接调用的是nonfairTryAcquireShared方法，这个方法是在父类Sync里面的。非公平获取锁的逻辑是先取出当前同步状态(同步状态表示许可证个数)，将当前同步状态减去传入的参数，如果结果不小于0的话证明还有可用的许可证，那么就直接使用CAS操作更新同步状态的值，最后不管结果是否小于0都会返回该结果值。

这里我们要了解tryAcquireShared方法返回值的含义，返回负数表示获取失败，零表示当前线程获取成功但后续线程不能再获取，正数表示当前线程获取成功并且后续线程也能够获取。我们再来看acquireSharedInterruptibly方法的代码。

// 以可中断模式获取锁(共享模式)
    public final void acquireSharedInterruptibly(int arg) 
        throws InterruptedException {
        // 首先判断线程是否中断, 如果是则抛出异常
        if (Thread.interrupted()) {
            throw new InterruptedException();
        }
        // 1.尝试去获取锁
        // 负数：表示获取失败
        // 零值：表示当前线程获取成功, 但是后继线程不能再获取了
        // 正数：表示当前线程获取成功, 并且后继线程同样可以获取成功
        if (tryAcquireShared(arg) < 0) {
            // 2. 如果获取失败则进人该方法
            doAcquireSharedInterruptibly(arg);
        }
    }

如果返回的remaining小于0的话就代表获取失败，因此tryAcquireShared(arg) < 0就为true，所以接下来就会调用doAcquireSharedInterruptibly方法，这个方法我们在讲AQS的时候讲过，它会将当前线程包装成结点放入同步队列尾部，并且有可能挂起线程。这也是当remaining小于0时线程会排队阻塞的原因。

而如果返回的remaining>=0的话就代表当前线程获取成功，因此tryAcquireShared(arg) < 0就为flase，所以就不会再去调用doAcquireSharedInterruptibly方法阻塞当前线程了。

以上是非公平获取的整个逻辑，而公平获取时仅仅是在此之前先去调用hasQueuedPredecessors方法判断同步队列是否有人在排队，如果有的话就直接return -1表示获取失败，否则才继续执行下面和非公平获取一样的步骤。

## 2、释放许可证 ##

// 释放一个许可证
    public void release() {
        sync.releaseShared(1);
    }

调用release方法是释放一个许可证，它的操作很简单，就调用了AQS的releaseShared方法，我们来看看这个方法。

// 释放锁的操作(共享模式)
    public final boolean releaseShared(int arg) {
        // 1.尝试去释放锁
        if (tryReleaseShared(arg)) {
            // 2.如果释放成功就唤醒其他线程
            doReleaseShared();
            return true;
        }
        return false;
    }

AQS的releaseShared方法首先调用tryReleaseShared方法尝试释放锁，这个方法的实现逻辑在子类Sync里面。

abstract static class Sync extends AbstractQueuedSynchronizer {
        ...
        // 尝试释放操作
        protected final boolean tryReleaseShared(int releases) {
            for (;;) {
                // 获取当前同步状态
                int current = getState();
                // 将当前同步状态加上传入的参数
                int next = current + releases;
                // 如果相加结果小于当前同步状态的话就报错
                if (next < current) {
                    throw new Error("Maximum permit count exceeded");
                }
                // 以CAS方式更新同步状态的值, 更新成功则返回true, 否则继续循环
                if (compareAndSetState(current, next)) {
                    return true;
                }
            }
        }
        ...
    }

可以看到tryReleaseShared方法里面采用for循环进行自旋，首先获取同步状态，将同步状态加上传入的参数，然后以CAS方式更新同步状态，更新成功就返回true并跳出方法，否则就继续循环直到成功为止，这就是Semaphore释放许可证的流程。

--------------------

**上一篇：[Java中的并发工具类之同步屏障 CyclicBarrier][Java_ CyclicBarrier]**

[Java_ CyclicBarrier]: https://blog.csdn.net/pcwl1206/article/details/85062706
[https_mp.weixin.qq.com_s_LS8YBKpiJnHEY1kMWmwoxg]: https://mp.weixin.qq.com/s/LS8YBKpiJnHEY1kMWmwoxg
[AQS_Semaphore]: https://blog.csdn.net/javazejian/article/details/76167357
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70]: /images/20220404/1c2df2edc43a424188166168233ac67a.png