Semaphore 源码解读

Posted 2021-01-07 aquariusm

之前一篇ReentrantLock的文章如果看过，并且对AQS的代码比较熟知的话，Semaphore的代码阅读起来就相对会轻松很多，如果不熟知的话，可以参考那篇文章或者自行学习下AQS的代码。

这里我们来分析下Semaphore的源码。

使用代码举例：

最早接触Semaphore是在前边翻阅Flume的代码的时候，看到的，后来借鉴这个做了一个业务上的提交通道。挺好用的就是 Semaphore + LinkedBlockingDeque实现的一个常用消费者。

大致代码：

1）：初始化 许可资源池及存储队列

    private void init(){
        if(this.capacity == 0 || this.capacity == null){
            capacity = defaultCapacity;
        }
        
        queue = new LinkedBlockingDeque<T>(capacity);
                queueStored = new Semaphore(0);
        
    }

2）：往队列里塞进资源，每塞进一个资源就调用queueStore.release()增加一个许可

public boolean put(T obj) throws Exception{
        if (!queue.offer(obj)) {
            throw new ChannelException(
                    "Put queue for Memory of capacity " + queue.size() + " full, consider committing more frequently, "
                            + "increasing capacity or increasing thread count");
        }
        queueStored.release();
        return false;
    }

3）：从队列中消费资源，获取相应量的许可成功，才能获取成功，否则线程失败阻塞

public List<T> take(int n) throws InterruptedException{
        if(!queueStored.tryAcquire(n, defaultKeepAlive, TimeUnit.MILLISECONDS)){
            return null;
        }
        List<T> ret = new ArrayList<T>();
        int j = n;
        synchronized (queueLock) {
            while(j>0){
                ret.add(queue.poll());
                j--;
            }
        }
        return ret;
    }

以上例子，是一个常用的使用Semaphore的场景。

 

Semaphore整理上思路可以理解为一个资源池，资源池中有相应数量的资源。这样当多线程争抢资源时，就可以通过Semaphore来限定资源数量或者说明当前资源多少，来进行资源争抢，那么问题来了。。。

1：为毛不能用volatile的int变量直接说明资源数量就可以了，而是搞一个这个呢？

事实上，看过代码就知道，实际的资源数量确实是通过volatile 内存可见的整型数据来表示资源数量的，但是Semaphore不仅仅是用来表示资源数量的，它是一个管理资源及相关争抢线程的数据结构及处理逻辑的能力者，比如如果只是一个变量来表明大小，那么线程争抢的时候，争抢不到该怎么办，是不是要客户线程要自己来处理继续争抢还是怎么样，如果没有资源了呢，是不是又要自己来处理继续等待还是放弃，这些事情就是Semaphore为我们提供的内容。

OK，解除了上边的疑问我们就来继续翻代码。

构造器：

    public Semaphore(int permits) {
        // 这里也区分公平锁、非公平锁，默认是非公平
        sync = new NonfairSync(permits);
    }

释放资源：

 

添加资源：

 

大于0，并且争抢成功    =>    直接返回，不做其他操作

大于0，并且争抢失败    =>    阻塞，继续争抢

小于0                         =>    直接返回，并进入等待序列，让出占用cpu

再来一篇，Semaphore VS ReentrantLock。