【JDK源码分析】并发包同步工具Semaphore

前言

这次分析信号量Semaphore，为什么称之为信号量呢？是因为它可以控制同时访问某个资源的操作数量或是同时执行某个指定操作的数量。就好比它像一个租赁汽车的公司，租赁公司的汽车的数量是固定的，用完需要归还，用之前需要去租借（acquire 前提是还有可用的汽车），如果汽车都被租出去了，那只能等到别人归还了才能租到。它是基于AQS的共享锁来实现的，其中使用了较多的AQS的方法，所以在这之前最好需要分析过AQS的源码，也可以查看本人之前AQS的源码分析，有些AQS方法没有在之前分析过的这里涉及到了会进行分析。

源码

构造器和属性

    // 继承了AQS,并且有2个子类实现（非公平锁和公平锁）
    private final Sync sync;

    // 构造器只有许可数量permits的信号量
    public Semaphore(int permits) {
        // 非公平锁
        sync = new NonfairSync(permits);
    }
    // 构造器带有许可数量permitsy以及是否是公平锁
    public Semaphore(int permits, boolean fair) {
        // fair为true时，为公平锁否则非公平
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }    

内部抽象类Sync

    // 继承AQS
    abstract static class Sync extends AbstractQueuedSynchronizer {
        Sync(int permits) {
            // 将许可值赋值给AQS的state
            setState(permits);
        }
        // 获取当前state值
        final int getPermits() {
            return getState();
        }
        // 非公平获取许可
        final int nonfairTryAcquireShared(int acquires) {
             // 保证CAS操作成功
            for (;;) {
                // state值，可用许可
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    // 小于0表示申请许可大于剩余的许可直接返回
                    // 大于0时会更新为之前剩余许可数减去申请许可数后的剩余值
                    return remaining;
            }
        }
        // 尝试归还
        protected final boolean tryReleaseShared(int releases) {
             // 保证CAS操作成功
            for (;;) {
                // 当前剩余许可数量
                int current = getState();
                // 当前剩余加上归还
                int next = current + releases;
                // int值溢出检查
                if (next < current) // overflow 
                    throw new Error("Maximum permit count exceeded");
                // CAS更新剩余
                if (compareAndSetState(current, next))
                    return true;
            }
        }
        // 减少许可
        final void reducePermits(int reductions) {
             // 保证CAS操作成功
            for (;;) {
                // 获取当前许可数
                int current = getState();
                // 减少reductions个许可
                int next = current - reductions;
                // 溢出
                if (next > current) // underflow
                    throw new Error("Permit count underflow");
                // CAS更新
                if (compareAndSetState(current, next))
                    return;
            }
        }
        // 将当前剩余许可全部扔掉
        final int drainPermits() {
            // 保证CAS操作成功
            for (;;) {
                int current = getState();
                // 当前剩余许可为0 直接返回0
                // 当前许可大于0，将其更新为0
                if (current == 0 || compareAndSetState(current, 0))
                    return current;
            }
        }
    }

内部类非公平锁实现类NonfairSync

    static final class NonfairSync extends Sync {   
        NonfairSync(int permits) {
            // 调用父类Sync的构造器
            super(permits);
        }
        // 尝试获取acquires个许可
        protected int tryAcquireShared(int acquires) {
            return nonfairTryAcquireShared(acquires);
        }
    }      

内部类公平锁实现类FairSync

    static final class FairSync extends Sync {

        FairSync(int permits) {
            // 同样调用父类的构造器
            super(permits);
        }

        protected int tryAcquireShared(int acquires) {
            // 保证操作成功
            for (;;) {
                // 获取许可之前查看是否有正在等待的
                if (hasQueuedPredecessors())
                    return -1;
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    // 小于0表示申请许可大于剩余的许可直接返回
                    // 大于0时会更新为之前剩余许可数减去申请许可数后的剩余值
                    return remaining;
            }
        }
    }

方法acquire 获取许可

acquire方法分为无参和有参，里面调用的方法都是一样只是参数不一样，都为可中断

    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    // 获取permits个许可
    public void acquire(int permits) throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireSharedInterruptibly(permits);
    }

AQS的acquireSharedInterruptibly方法

    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        // 线程中断时抛出异常
        if (Thread.interrupted())
            throw new InterruptedException();

        // arg 大于可用许可时会调用doAcquireSharedInterruptibly方法
        // doAcquireSharedInterruptibly 当还有可用许可时不会挂起当前线程，反之则会挂起当前线程，在上一篇博文有分析
        if (tryAcquireShared(arg) < 0)
            // doAcquireSharedInterruptibly 
            doAcquireSharedInterruptibly(arg);
    }

方法release 归还许可

    public void release() {
        // 归还一个许可
        sync.releaseShared(1);
    }
    public void release(int permits) {
        // 参数保护
        if (permits < 0) throw new IllegalArgumentException();
        // 归还permits个许可
        sync.releaseShared(permits);
    }

AQS的releaseShared方法

    public final boolean releaseShared(int arg) {
        // 归还许可，只要许可没让int溢出都返回true，然后会去唤醒正在等待许可的线程
        if (tryReleaseShared(arg)) {
            // 上一篇博文有分析此方法
            doReleaseShared();
            return true;
        }
        return false;
    }

总结

信号量可以用来实现资源池，比如数据库连接池、线程池等等，信号量的许可量在初始化后并不是还可以通过drainPermits将当前许可数量变为0.