详解Java信号量Semaphore的原理及使用

/**
 * 保存某个AQS子类实例
 */
private final Sync sync;

/**
 * 创建具有给定的信号量数和非公平的公平模式的 Semaphore。
 *
 * @param permits 初始的可用信号量数目。此值可能为负数，在这种情况下，必须在授予任何获取信号量前进行释放信号量。
 */
public Semaphore(int permits) {
    //默认初始化NonfairSync实例
    sync = new NonfairSync(permits);
}

/**
 * 创建具有给定的信号量数和给定的公平设置的 Semaphore。
 *
 * @param permits 初始的可用信号量数目。此值可能为负数，在这种情况下，必须在授予任何获取信号量前进行释放信号量。
 * @param fair    如果此信号量保证在争用时按先进先出的顺序授予信号量，则为 true；否则为 false。
 */
public Semaphore(int permits, boolean fair) {
    //根据fair参数选择初始化一个公平FairSync类或者非公平NonfairSync类的实例
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}


/**
 * 非公平模式的实现
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = -2694183684443567898L;

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

    //…………其他方法后面再讲

}

/**
 * 公平模式的实现
 */
static final class FairSync extends Sync {
    private static final long serialVersionUID = 2014338818796000944L;

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

    //…………其他方法后面再讲

}

/**
 * 信号量的同步实现。 使用 AQS 的state状态表示信号量。子分类为公平和非公平模式。
 */
abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 1192457210091910933L;

    /**
     * 构造器
     *
     * @param permits 初始的可用信号量数目。
     */
    Sync(int permits) {
        //被设置为state值
        setState(permits);
    }

    //…………其他方法后面再讲
}

/**
 * Semaphore的acquire方法
 * 从信号量获取一个信号量，没有则一直阻塞，直到在其他线程提供信号量并唤醒或者线程被中断。
 *
 * @throws InterruptedException 如果调用此方法时已被中断或者等待时被中断
 */
public void acquire() throws InterruptedException {
    //内部调用AQS的acquireSharedInterruptibly方法
    //这实际上就是共享式可中断获取资源模版方法
    sync.acquireSharedInterruptibly(1);
}

/**
 * 从信号量获取permits个信号量，没有则一直阻塞，直到在其他线程提供信号量并唤醒或者线程被中断。
 *
 * @param permits 需要获取的信号量数量
 * @throws InterruptedException 如果调用此方法时已被中断或者等待时被中断
 */
public void acquire(int permits) throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    //参数就是permits
    sync.acquireSharedInterruptibly(permits);
}

/**
 1. AQS的acquireSharedInterruptibly方法
 2. 共享式可中断获取信号量资源的模版方法
 3.  4. @param arg 需要获取的信号量资源数量
 5. @throws InterruptedException 如果调用此方法时已被中断或者等待时被中断
 */
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    //最开始就检查一次，如果当前线程是被中断状态，直接抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //调用tryAcquireShared尝试获取共享信号量资源，这个方法是子类自己重写的
    //如果返回值小于0，表示当前线程共享信号量资源失败，否则表示成功
    //Semaphore的FairSync和NonfairSync对tryAcquireShared分别做出了公平和不公平的实现
    if (tryAcquireShared(arg) < 0)
        //获取不到就执行doAcquireSharedInterruptibly方法
        doAcquireSharedInterruptibly(arg);
}

/**
 * 公平模式
 */
static final class FairSync extends Sync {
    /**
     * 尝试公平的获取共享信号量资源
     *
     * @param acquires 获取信号量资源数量
     * @return 如果返回值小于0，表示当前线程共享信号量资源失败，否则表示成功
     */
    protected int tryAcquireShared(int acquires) {
        /*开启一个循环尝试获取共享信号量资源*/
        for (; ; ) {
            //这是AQS实现公平模式的预定义的方法，AQS帮我们实现好了。该方法用于查询是否有任何线程等待获取信号量资源的时间超过当前线程
            //如果该方法返回true，则表示有线程比当前线程更早地请求获取信号量资源。由于是公平的的，因此当前线程不应该获取信号量资源，直接返回-1，表示获取信号量资源失败
            if (hasQueuedPredecessors())
                return -1;
            //到这里，表示当前线程就是最早请求获取信号量资源，可以尝试获取

            //获取state的值available，我们知道state代表信号量资源数量
            int available = getState();
            //remaining为available减去需要获取的信号量资源数量之后的差值
            int remaining = available - acquires;
            //如果remaining小于0，那么返回remaining值，由于是负数，因此获取失败
            //如果大于等于0，那么表示可以获取成功，尝试CAS的更新state，更新成功之后同样返回remaining，由于是大于等于0的数，因此获取成功
            if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                return remaining;
            //如果remaining大于等于0，但是CAS更新state失败，那么循环重试
        }
    }
}

/**
 * 非公平模式
 */
static final class NonfairSync extends Sync {

    /**
     * 尝试非公平的获取共享信号量资源
     *
     * @param acquires 获取信号量资源数量
     * @return 如果返回值小于0，表示当前线程共享信号量资源失败，否则表示成功
     */
    protected int tryAcquireShared(int acquires) {
        //调用父类Sync的nonfairTryAcquireShared方法
        //为什么该方法要实现在父类中的，因为无论是指定的公平模式还是非公平模式，
        //它们的tryAcquire方法都是调用的nonfairTryAcquireShared方法，即非公平的，因此实现在父类中
        return nonfairTryAcquireShared(acquires);
    }
}

/**
 * AQS的实现，作为公平和非公平模式的父类，有一些共享方法
 */
abstract static class Sync extends AbstractQueuedSynchronizer {

    /**
     * 尝试非公平的获取共享信号量资源
     *
     * @param acquires 获取信号量资源数量
     * @return 如果返回值小于0，表示当前线程共享信号量资源失败，否则表示成功
     */
    final int nonfairTryAcquireShared(int acquires) {
        /*开启一个循环尝试获取共享信号量资源*/
        for (; ; ) {
            //相比于公平模式，少了hasQueuedPredecessors的实现
            //获取state的值available，我们知道state代表信号量资源数量
            int available = getState();
            //remaining为available减去需要获取的信号量资源数量之后的差值
            int remaining = available - acquires;
            //如果remaining小于0，那么返回remaining值，由于是负数，因此获取失败
            //如果大于等于0，那么表示可以获取成功，尝试CAS的更新state，更新成功之后同样返回remaining，由于是大于等于0的数，因此获取成功
            if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                return remaining;
            //如果remaining大于等于0，但是CAS更新state失败，那么循环重试
        }
    }
}

/**
 * 获取一个信号量，没有则一直阻塞，直到在其他线程提供信号量并唤醒该线程。
 * 获取一个信号量就立即返回，将可用的信号量数减 1。
 */
public void acquireUninterruptibly() {
    //内部调用AQS的acquireShared方法
    //这实际上就是共享式不可中断获取资源模版方法
    sync.acquireShared(1);
}

/**
 * AQS的acquireShared方法
 * 共享式不可中断获取资源模版方法
 *
 * @param arg 获取的资源数量
 */
public final void acquireShared(int arg) {
    //并没有检查中断
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}


/**
 * 获取permits个信号量，没有则一直阻塞，直到在其他线程提供信号量并唤醒该线程。
 *
 * @param permits 获取的信号量数量
 */
public void acquireUninterruptibly(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    //参数就是permits
    sync.acquireShared(permits);
}

/**
 * @param timeout 超时时间
 * @param unit    时间单位
 * @return 是否获取资源成功
 * @throws InterruptedException 如果调用此方法时已被中断或者等待时被中断
 */
public boolean tryAcquire(long timeout, TimeUnit unit)
        throws InterruptedException {
    //实际上就是调用的AQS的共享式超时获取资源的方法，获取1个资源
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

/**
 * @param permits 获取的资源数量
 * @param timeout 超时时间
 * @param unit    时间单位
 * @return 是否获取资源成功
 * @throws InterruptedException 如果调用此方法时已被中断或者等待时被中断
 */
public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
        throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    //实际上就是调用的AQS的共享式超时获取资源的方法，获取permits个资源
    return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
}

/**
 * AQS的共享式超时获取资源的模版方法，支持中断
 *
 * @param arg          参数
 * @param nanosTimeout 超时时间，纳秒
 * @return 是否获取资源成功
 * @throws InterruptedException 如果调用此方法时已被中断或者等待时被中断
 */
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    //最开始就检查一次，如果当前线程是被中断状态，直接抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //下面是一个||运算进行短路连接的代码，同样左边是调用子类实现的tryAcquireShared尝试获取资源，获取到了直接返回true
    //获取不到资源就执行doAcquireSharedNanos方法，这个方法是AQS的方法，因此超时机制是AQS帮我们实现的！
    return tryAcquireShared(arg) >= 0 ||
            doAcquireSharedNanos(arg, nanosTimeout);
}

public boolean tryAcquire(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    //调用nonfairTryAcquireShared方法
    return sync.nonfairTryAcquireShared(permits) >= 0;
}

public boolean tryAcquire(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    //调用nonfairTryAcquireShared方法
    return sync.nonfairTryAcquireShared(permits) >= 0;
}

/**
 * 释放一个信号量，信号量总数加1。
 */
public void release() {
    //内部调用AQS的releaseShared方法
    //这实际上就是共享式释放资源的模版方法
    sync.releaseShared(1);
}

/**
 * 释放permits个信号量，信号量总数加permits。
 *
 * @param permits 释放的信号量个数
 */
public void release(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    //参数就是permits
    sync.releaseShared(permits);
}


/**
 * AQS的共享模式下释放资源的模版方法。
 * 如果成功释放则会调用doReleaseShared
 */
public final boolean releaseShared(int arg) {
    //tryReleaseShared释放信号量资源，该方法由子类自己实现
    if (tryReleaseShared(arg)) {
        //释放成功,必定调用doReleaseShared尝试唤醒后继结点，即阻塞的线程
        doReleaseShared();
        return true;
    }
    return false;
}

/**
 * Sync的tryReleaseShared实现
 *
 * @param releases 要释放的资源数量
 * @return true 成功 false 失败
 */
protected final boolean tryReleaseShared(int releases) {
    for (; ; ) {
        //很简单，就是尝试CAS的增加state值，增加releases
        int current = getState();
        int next = current + releases;
        //这里可以知道，信号量资源数量不可超过int的最大值
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        //CAS的增加state值，CAS成功之后返回true，否则循环重试
        if (compareAndSetState(current, next))
            return true;
    }
}

/**
 * @author lx
 */
public class SemaphoreTest {

    /**
     * 机器数目，实际上就是信号量为5，非公平模式
     */
    private static Semaphore semaphore = new Semaphore(5, false);
    /**
     * 机器数目，实际上就是信号量为5，公平模式
     */
    //private static Semaphore semaphore = new Semaphore(5, true);

    /**
     * 当所有工人都完成任务，那么统计工作量
     */
    private static CountDownLatch countDownLatch = new CountDownLatch(10);

    /**
     * 工人数目，8
     */
    private static final int NUM = 10;

    /**
     * 当前时间
     */
    private static final long NOW = System.nanoTime();

    /**
     * 纳秒单位
     */
    private static final long NANOUNIT = 1000000000;

    /**
     * 工作量
     */
    private static final LongAdder WORKLOAD = new LongAdder();


    static class Worker implements Runnable {
        public Worker(int num) {
            this.num = num;
        }

        private int num;
        private long timed = 20 * NANOUNIT;

        @Override
        public void run() {
            while (true) {
                //获取信号量
                try {
                    if (semaphore.tryAcquire(timed, TimeUnit.NANOSECONDS)) {
                        System.out.println("工人" + this.num + "占用一个机器在生产...");
                        //占用一定时间
                        LockSupport.parkNanos((long) (NANOUNIT * num * 0.5));
                        //统一调整为2秒，将会看到更明显的Semaphore效果
                        //LockSupport.parkNanos((long) (NANOUNIT * 2));

                        System.out.println("工人" + this.num + "生产完毕，释放出机器");
                        //释放信号量
                        //每个工人最多执行20秒
                        WORKLOAD.increment();
                        if ((timed = timed - (System.nanoTime() - NOW)) <= 0) {
                            semaphore.release();
                            countDownLatch.countDown();
                            break;
                        }
                        semaphore.release();
                    } else {
                        countDownLatch.countDown();
                        break;
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < NUM; i++) {
            executorService.execute(new Worker(i));
        }
        executorService.shutdown();
        countDownLatch.await();
        System.out.println("工作完毕，空闲机器为：" + semaphore.availablePermits());
        System.out.println("总工作量为：" + WORKLOAD.sum());
    }
}