JUC之Semaphore

我不是女神ヾ 2022-10-05 06:51 143阅读 0赞

Semaphore，信号量。从翻译的“信号量”来看，并不是很清楚这个类的作用。

![20210607191718338.png][]

如果你之前对**ReentrantLock**有所认识的话，学习Semaphore，你会发现两个类在实现内容上大致相同。（关于ReentrantLock的介绍，请戳《[JUC之ReentrantLock][JUC_ReentrantLock]》）

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

ReentrantLock使用**AbstractQueuedSynchronizer**中的**state**字段来表示锁的数量，因为ReentrantLock是互斥锁，因此state的值只能为0和1。state=0时，表示锁未被任何线程占用，此时某个线程可以获取锁成功，将state的值增加1变成1。state=1表示锁已经被某个线程占用，这样其他线程想要去获取锁的话就只能等待，直到持有锁的线程释放了锁，将state的值减少1变成0。

从另一个方面来说的话，在ReentrantLock中，state可以看做的共享资源的数量。state最大值为1，也就表示共享资源只有一个，线程之间想要同时使用这个共享资源是不可能的，只能一个一个轮流来使用。

将state的值看做共享资源的数量来看，这样就可以引出Semaphore。在Semaphore中也是借助AbstractQueuedSynchronizer的state字段，以此来表示共享资源的数量。不同于ReentrantLock中state的值只能为0和1，在Semaphore中state的值范围为0到int类型的最大值（2147483647），即0到2147483647。假设设置state的值为10，即表明当前共享资源的数量为10，线程们可以去并发争抢这10个共享资源，一个线程抢到一个共享资源，总的共享资源数量就要减去1，当10个共享资源都被抢走了，那么剩下的线程就没有共享资源可使用了，此时他们只能等待，直到有了共享资源。既然可以抢共享资源，那就也要有归还共享资源，当持有共享资源的线程归还了一个共享资源，那么总的共享资源数量需要加上相应的数量。在Semaphore中，每个线程每次不是只能争抢一个共享资源，在共享资源数量允许的情况下，其是可以争抢多个共享资源的，有个原则，线程争抢了几个共享资源，那归还的时候也要归还相应数量的共享资源。

举个例子，假设当前有10把椅子，来了一个人就去坐这个椅子。当来了10个人，10把椅子都被坐了，那后面新来的人就没有椅子可以坐了，他们要想坐到椅子上，只能等椅子上有人离开，他们再去争抢可坐的椅子。此时，坐在椅子上有个人离开了，那么先前没有获取到椅子的人便开始争抢这把椅子，争抢到椅子的人可以坐在椅子上，没争抢到椅子的人只能继续等待。上面的情况是一个人只能占用一把椅子，但在实际情况中，会有人占座的情况，他一个人竟然占了3把椅子，这种情况也是允许的。占用了3把椅子的这个人在归还椅子的时候，需要将3把椅子都归还了。

ReentrantLock中有**公平锁**和**非公平锁**，Semaphore也是有公平和非公平去争抢共享资源的区分。

根据Semaphore提供的功能，Semaphore可以用来实现限流的功能。接口在一定时间内只允许N个人访问。

从上面的叙述中，我们可以从以下几个方面来学习Semaphore。

**（1）设置共享资源的数量**

**（2）争抢共享资源（公平/非公平）**

**（3）归还共享资源**

**（4）一次争抢和归还多个共享资源**

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

# **█ **Semaphore #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70][]

（0）**Sync**

Semaphore中持有一个**Sync**字段用来标记同步器类型。

private final Sync sync;

Semaphore功能的实现是依赖于AbstractQueuedSynchronizer的，按照套路，Semaphore在维护一个Sync内部类，Sync继承AbstractQueuedSynchronizer来实现相应功能。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1][]

Semaphore也提供了两种方式来争抢共享资源，公平和非公平。

**FairSync**以公平的方式竞争共享资源，**NonfairSync**以非公平的方式竞争非共享资源。

在前面关于争抢共享资源的描述中，可以看出是在共享资源数量允许的情况下，是可以同时有多个线程争抢到共享资源的。因此Sync继承AbstractQueuedSynchronizer，需要重新AbstractQueuedSynchronizer中共享模式下获取资源的方法：**tryAcquireShared**、**tryReleaseShared**。（Sync直接重新tryReleaseShared方法，tryAcquireShared方法分别由子类实现）

**（1）设置共享资源的数量**

关于设置共享资源的数量，是通过Sync的构造器来设置的。在构造器中调用AbstractQueuedSynchronizer的**setState**方法来初始化state的值，即实现设置共享资源的数量。

Sync(int permits) {
        setState(permits);
    }
    
    // AbstractQueuedSynchronizer中的方法
    protected final void setState(int newState) {
        state = newState;
    }

**（2）争抢共享资源（公平/非公平）**

*  **acquire**

通过调用**acquire**方法来争抢共享资源，且默认每次只争抢一个共享资源。其是通过调用Sync的**acquireSharedInterruptibly**方法。acquireSharedInterruptibly是由AbstractQueuedSynchronizer提供的模板方法，子类需要重写tryAcquireShared来自定义获取锁的逻辑。

public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

Sync是分FairSync与NonfairSync。下面分别来看FairSync、NonfairSync中对方法tryAcquireShared的重新。

**①NonfairSync**

protected int tryAcquireShared(int acquires) {
        return nonfairTryAcquireShared(acquires);
    }

**nonfairTryAcquireShared**是Sync中的方法：

// Sync中的方法
    final int nonfairTryAcquireShared(int acquires) {
        // 通过自旋操作去获取共享资源
        for (;;) {
            // available获取可用的共享资源数量
            int available = getState();
            // 将共享资源的数量减1
            int remaining = available - acquires;
            // 如果remaining<0，表示没有了可用的共享资源，方法会直接返回一个负数
            if (remaining < 0 ||
            // CAS操作更改state的值
                compareAndSetState(available, remaining))
                return remaining;
        }
    }

关于tryAcquireShared的返回值，返回小于0的值，则表示获取资源失败。否则表示获取资源成功。所以当没有了可用资源时，nonfairTryAcquireShared的返回值为负数，即没有获取到资源成功，后面在AbstractQueuedSynchronizer的acquireSharedInterruptibly方法中会将当前线程加入等待队列中。nonfairTryAcquireShared的返回值>=0，表示争抢资源成功。

**②FairSync**

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

公平的方式比非公平的方式就多了一个逻辑，**hasQueuedPredecessors**方法的逻辑判断，即判断在等待队列中是否已经有正在等待的线程了，如果有，当前线程要先让其前面的线程去获取资源，tryAcquireShared返回-1，表示争抢资源失败。

**（3）归还共享资源**

*  **release**

**release**方法释放资源，其是与acquire()方法对应，每次释放一个资源。

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

**releaseShared**是AbstractQueuedSynchronizer中提供的模板方法，会在其中调用子类的重新方法tryReleaseShared。释放资源就没有公平和非公平的区分了，Sync重写了tryReleaseShared。

protected final boolean tryReleaseShared(int releases) {
        // 自旋释放资源
        for (;;) {
            // 获取当前的共享资源数量
            int current = getState();
            // 将共享资源的数量加上释放的数量
            int next = current + releases;
            // 一方面的避免传进来的参数releases值为负数，
            // 另一方面避免数值过大的操作，即超过了int的最大值范围2147483647
            if (next < current) 
                throw new Error("Maximum permit count exceeded");
            // CAS操作更改state值
            if (compareAndSetState(current, next))
                return true;
        }
    }

上面的代码不复杂，主要就是next < current的问题，主要有两个方面的考虑，一方面是如果方法传进来的参数releases值为负数，那么（current + releases）< current成立，所以releases的值不能为负数。另一方面是releases是正数了，但是经过current + releases操作之后的结果超过了int的最大值2147483647，也会变成负数，如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2][]

**（4）一次争抢和归还多个共享资源**

*  **acquire(int permits)**

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

Semaphore还支持每次一个线程去争抢多个资源，重载了acquire方法，指定要争抢资源的数量。如果剩余的资源不够争抢的，则会争抢失败。即当前剩余资源为2，但一个线程想要一次获取3个资源，此时资源是不够的，该线程争抢资源失败。（要么成功要么失败，不存在部分成功、部分失败的情况）

*  **release(int permits)**

public void release(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.releaseShared(permits);
    }

与acquire(int permits)对应，release也重载了一个方法，可以指定释放的资源数量。

**（5）tryAcquire重载**

补充介绍下tryAcquire方法。Semaphore提供了几个tryAcquire重载方法，用于尝试获取资源。此方法会尝试去竞争一次资源。抢到资源返回true，未抢到资源返回false。

①竞争一次资源，每次只竞争一个资源：

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 boolean tryAcquire(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.nonfairTryAcquireShared(permits) >= 0;
    }

④带有超时时间吗，且可以指定竞争的资源的个数：

public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
        throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
    }

[20210607191718338.png]: /images/20221005/5582efbc83244842bce55ad32d89cc00.png
[JUC_ReentrantLock]: https://blog.csdn.net/weixin_50518271/article/details/117265771?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70]: /images/20221005/bbe38e01b7964912ad1c7d52f9b6c916.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1]: /images/20221005/2edbde9d81ce49d9bf469518b0e0a292.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2]: /images/20221005/9729cc1925064d559b8968b32f3b0f4b.png