java.util.concurrent.Semaphore 类源码的深入解读

怼烎@ 2022-12-14 13:54 105阅读 0赞

源码先贴上，后面是实际使用的时候代码的执行流程

import java.util.Collection;
    import java.util.concurrent.TimeUnit;
    
    public class Semaphore implements java.io.Serializable { 
        private static final long serialVersionUID = -3222578661600680210L; // 序列化版本号
        private final Sync sync; // 抽象队列同步器的子类成员常量
    
    
        // 构造方法
        /** * 构造一个有permits个信号量的非公平同步器 */
        public Semaphore(int permits) { 
            sync = new NonfairSync(permits);
        }
    
        /** * 构造一个有permits个信号量 * 如果传入的布尔值fair=true则是公平同步器，false则是非公平同步器 */
        public Semaphore(int permits, boolean fair) { 
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }
    
    
        /** * 下面是常用方法 */
        /** * 获取信号量 */
        // 获取一个信号量，如果信号量被使用完则抛异常InterruptedException
        public void acquire() throws InterruptedException { 
            sync.acquireSharedInterruptibly(1);
        }
    
        // 获取permits个信号量，如果不够则抛异常InterruptedException
        public void acquire(int permits) throws InterruptedException { 
            if (permits < 0) throw new IllegalArgumentException();
            sync.acquireSharedInterruptibly(permits);
        }
    
        // 获取permits个不可中断的信号量
        public void acquireUninterruptibly(int permits) { 
            if (permits < 0) throw new IllegalArgumentException();
            sync.acquireShared(permits);
        }
    
        // 阻塞式（就是等待）获取信号量，当其它信号量被释放，并执行当前代码则成功
        public void acquireUninterruptibly() { 
            sync.acquireShared(1);
        }
    
        /** * 释放信号量方法 */
        // 释放一个信号量
        public void release() { 
            sync.releaseShared(1);
        }
    
        // 释放permits个信号量
        public void release(int permits) { 
            if (permits < 0) throw new IllegalArgumentException();
            sync.releaseShared(permits);
        }
    
        /** * 常用信号量辅助方法 */
        // 获取所有信号量，并将所有信号量都使用掉，返回0
        public int drainPermits() { 
            return sync.drainPermits();
        }
    
        // 获取可用的信号量个数
        public int availablePermits() { 
            return sync.getPermits();
        }
    
        // 获取等待线程队列的长度
        public final int getQueueLength() { 
            return sync.getQueueLength();
        }
    
        /** * 判断信号量方法 */
        // 判断是否可以获取一个信号量
        public boolean tryAcquire() { 
            return sync.nonfairTryAcquireShared(1) >= 0;
        }
        // 判断是否可以获取permits个信号量
        public boolean tryAcquire(int permits) { 
            if (permits < 0) throw new IllegalArgumentException();
            return sync.nonfairTryAcquireShared(permits) >= 0;
        }
        // 判断在指定时间内可否获取一个信号量,超过等待时间timeout则抛异常InterruptedException
        public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException { 
            return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
        }
        // 判断在timeout时间内是否可以获取permits个信号量throws InterruptedException
        public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException { 
            if (permits < 0) throw new IllegalArgumentException();
            return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
        }
        // 判断是否是公平机制的信号同步器
        public boolean isFair() { 
            return sync instanceof FairSync;
        }
        // 判断线程队列是否有线程,有线程在等待获取信号量则返回true，否则返回false
        public final boolean hasQueuedThreads() { 
            return sync.hasQueuedThreads();
        }
    
        /** * toString方法得到信号量 */
        public String toString() { 
            return super.toString() + "[Permits = " + sync.getPermits() + "]";
        }
    
    
        /** * 获取线程队列的集合 */
        protected Collection<Thread> getQueuedThreads() { 
            return sync.getQueuedThreads();
        }
        // 减少reduction个信号量（总量减少）
        protected void reducePermits(int reduction) { 
            if (reduction < 0) throw new IllegalArgumentException();
            sync.reducePermits(reduction);
        }
    
        /** * 同步器的抽象方法 */
        abstract static class Sync extends AbstractQueuedSynchronizer { 
            private static final long serialVersionUID = 1192457210091910933L;  // 序列化版本号
    
            // 带一个参数的构造方法
            Sync(int permits) { 
                setState(permits);// 调用AQS中的setState
            }
            // 获得许可数量
            final int getPermits() { 
                return getState();//调用AQS的方法
            }
    
            // 使用acquires个信号量，失败则直接返回计算得到的剩余信号量的值
            final int nonfairTryAcquireShared(int acquires) { 
                for (; ; ) { 
                    int available = getState();//不断自旋获取当前剩余信号量
                    int remaining = available - acquires;//计算剩余信号量
                    //前面如果false则需要执行后面的cas操作更新信号量，如果剩余信号量大于0则将这个负值的信号量直接返回
                    if (remaining < 0 || compareAndSetState(available, remaining))
                        return remaining;
                }
            }
            // 释放releases个信号量 失败则会抛异常
            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))// 说明没有超过最大值，进行更新许可值，并返回true
                        return true;
                }
            }
    
            // 使用reductions个信号量减少的是总量，也就是最大值
            final void reducePermits(int reductions) { 
                for (; ; ) { 
                    int current = getState(); // 当前许可
                    int next = current - reductions;// 计算减少后的新的总量
                    if (next > current) // 新的总量不能比之前的总量大，大则抛异常
                        throw new Error("Permit count underflow");
                    if (compareAndSetState(current, next))
                        return;
                }
            }
    
            final int drainPermits() { 
                for (; ; ) { 
                    int current = getState();
                    if (current == 0 || compareAndSetState(current, 0)) // 将信号量更新为0，表示全部使用完了
                        return current;
                }
            }
        }
    
        /** * 实现同步器的公平同步器 */
        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()) //如果已经形成了AQS队列，说明信号量早用完了，直接返回-1
                        return -1;
                    int available = getState();  //当前可用信号量
                    int remaining = available - acquires;// 计算使用了acquires信号量的值
                    if (remaining < 0 || compareAndSetState(available, remaining))// 值小于0直接返回计算后的负值remaining，不小于0则说明不会进行阻塞
                        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); // 调用父类Sync种的final方法
            }
        }
    
    }

#### 探究Semaphore 的使用以及流程搞定原理 ####

此前你需要掌握AQS原理，以及CAS  
[AQS原理和CAS相关文章点这里][AQS_CAS]

在我们使用的时候使用`new Semaphore (信号量初始值)`得到一个信号量的实例对象。  
当我们调用`acquire()`或者`acquire(int permits)`使用信号量时，我们看下前面源码，实际上使用的时非公平锁的实例对象，也就是NonfairSync的实例。  
看下使用信号量会执行哪些代码，我们已空参的`acquire()`为例。

// 获取一个信号量，如果信号量被使用完则抛异常InterruptedException
    public void acquire() throws InterruptedException { 
        sync.acquireSharedInterruptibly(1);
    }

会发现调用的则是AQS的final方法，传入的`arg=1`

public final void acquireSharedInterruptibly(int arg) throws InterruptedException { 
        if (Thread.interrupted() 
        		|| (tryAcquireShared(arg) < 0 
        		&& acquire(null, arg, true, true, false, 0L) < 0))
            throw new InterruptedException();
    }

在这个方法中会发现会执行`tryAcquireShared(1)`这个方法则在子类中实现，需要返回到Semaphore的NonfairSync类中定义的该方法如下：

//尝试获取信号量（非公平锁机制）
    protected int tryAcquireShared(int acquires) { 
        return nonfairTryAcquireShared(acquires); // 调用父类Sync种的final方法
    }

会发现这个`nonfairTryAcquireShared(1)`又是NonfairSync的父类Sync中定义的

// 使用acquires个信号量，失败则直接返回计算得到的剩余信号量的值
    final int nonfairTryAcquireShared(int acquires) { 
        for (; ; ) { 
            int available = getState();//不断自旋获取当前剩余信号量
            int remaining = available - acquires;//计算剩余信号量
            //前面如果false则需要执行后面的cas操作更新信号量，如果剩余信号量大于0则将这个负值的信号量直接返回
            if (remaining < 0 || compareAndSetState(available, remaining))
                return remaining;
        }
    }

也就是返回使用了1个信号量剩余的信号量值，如果使用后信号量剩余>=0则使用成功，对信号量进行更新，小于0则直接将计算结果后的值直接返回（结果为负值的情况），>=0也会返回，但是会进行更新信号量操作。

这里会发现里面有一个`getState()`方法来自于AQS，也就是在我们实例Semaphore时会按照如下顺序将state值赋值为我们传入Semaphore的值，也就是信号量。

### 注意构造的时候是逆序构造的，需要先构造好父类才可以子类，这里面只是将这个信号量的初始值一直传到了AQS并赋值给了state ###

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

NonfairSync(int permits) { 
        super(permits);
    }

// 带一个参数的构造方法
    Sync(int permits) { 
        setState(permits);// 调用AQS中的setState
    }

protected AbstractQueuedSynchronizer() { 
    }

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

回到代码，看我加的注释

public final void acquireSharedInterruptibly(int arg) throws InterruptedException { 
        if (Thread.interrupted() 
        		|| (tryAcquireShared(arg) < 0  //这里如果返回的剩余信号量<0也就是true，
         // 意味着需要阻塞线程，也就是判断中后面一个判断acquire方法会执行，如果不小于0则说明信号量没有被过度消费
        		&& acquire(null, arg, true, true, false, 0L) < 0))
            throw new InterruptedException();
    }

`acquire()`我们在AQS就读过一遍，我们接着看下Semaphore的acquire流程

通过源码我们很容易发现原来和ReentrantLock差不多就是走的分支不同，但是最终同样和ReetrantLock形成的队列差不多  
换而言之，Semaphore如果信号量设置为1，可以相当于低配版的ReentrantLock。  
为了更方便的读源码，我们借助工具Idea进行调试，不然acquire中定义的变量、以及分支够你看的眼花。

public class SyncTest { 
    
        public static void main(String[] args) { 
            Semaphore semaphore=new Semaphore(1);
            for (int i = 0; i < 2; i++) { 
                    try { 
                        semaphore.acquire();//此处加断点，在第二次循环时进入这个方法，会一直到AQS的acquire方法最终会被阻塞因为信号量被使用完了
    					// semaphore.release(); //释放信号量调用这个
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
            }
        }
    }

根据我加的注释在semaphore.acquire();加断得，我们重点看第二次循环，看下Semaphore类的AQS队列是怎么形成的。  
前面变量定义省略，直接看进入自旋  
第一层的if判断不会进入  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center]  
看下面的if发现这里会进入，我们继续追踪  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 2]  
这个方法有子类非公平同步器实现  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 3]  
又间接调用父类Sync的方法，这个方法就是计算剩余信号量，和前面差不多逻辑返回计算后剩余的信号量  
根据我在Semaphore源码中的注释继续执行

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 5]  
也就意味着不执行`if(acquired)`内代码块  
然后和ReentrantLock一样创建了一样的头节点  
![在这里插入图片描述][20201119015533871.png_pic_center]  
后面的逻辑和ReentrantLock一致，唯一区别在于Semaphore走的分支不同。  
最终也会执行acquire方法中`LockSupport.park()`阻塞线程  
如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 6]

释放信号量，应该和ReentrantLock差不多，如果释放了信号量信号量>=1则会尝试从AQS队列中将对应线程唤醒

[AQS_CAS]: https://blog.csdn.net/qq_41813208/article/details/109501988
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/ddd9e908898c4fa9bccf743310230038.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/2b6b8e8acd494301a06534988d05e3a1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/9ac73aea94394a60abdd5ed1e1dfe2c3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/a61e3e81e2e44e3b8cb2ba79720854b1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/9e444541d6d14d32b292a235b0ec7371.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/a6b64dd01c6846f583dae4088cda40c7.png
[20201119015533871.png_pic_center]: /images/20221123/fd9ff62e20c64e08a1a2026a4f55942e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxODEzMjA4_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/6d1f23c977ac4e0cb87ac9272a93ae36.png