AQS synchronization component - Semaphore (semaphore) analysis and case

基本概念

Semaphore也是一个线程同步的辅助类,可以维护当前访问自身的线程个数,并提供了同步机制.使用Semaphore可以控制并发访问资源的线程个数.

For example, queuing up to buy tickets,If only three windows,Then at most three people can buy tickets at the same time.When the fourth person arrives, they have to wait behind,Only other people bought it,You can go to the corresponding window to buy tickets .

作用和使用场景

• Used to guarantee the number of concurrent access threads at the same time.
• Semaphores are an important concept in operating systems,JavaConcurrent librarySemaphoreYou can easily complete the control similar to the operating system semaphore.SemaphoreIt is easy to control the number of threads accessing a resource at the same time in the system.
• In data structures we learned about linked lists,A linked list can normally hold an infinite number of nodes,而Semaphore可以实现有限大小的列表.

使用场景：Resources that only provide limited access.比如数据库连接.

源码分析

构造函数

/***接受一个整型的数字,表示可用的许可证数量.Semaphore（10）表*show permission10个线程获取许可证,*也就是最大并发数是10. ** @param permits The initial number of available licenses.**/public Semaphore(int permits) {        sync = new NonfairSync(permits);}/** * Use the given number of licenses and the given fairness settings * * @param permits The initial number of available licenses. * * @param fair Specifies whether it is fair or unfair mode,默认非公平模式 . 公平模式：The thread that starts first gets priority * 许可. 非公平模式：The thread that starts first does not necessarily get permission first,Whoever grabs it gets a license. */public Semaphore(int permits, boolean fair) {    sync = fair ? new FairSync(permits) : new NonfairSync(permits);}

常用方法

• acquire() 获取一个许可
• acquire(int permits) Acquires the specified number of licenses
• tryAcquire()方法尝试获取1个许可证
• tryAcquire(long timeout, TimeUnit unit) Maximum time to wait for permission
• tryAcquire(int permits) Acquires the specified number of licenses
• tryAcquire(int permits, long timeout, TimeUnit unit) Maximum time to wait for permission
• availablePermits() ： 返回此信号量中当前可用的许可证数
• release() 释放许可
• release(int permits) 释放指定个数的许可
• int getQueueLength() 返回正在等待获取许可证的线程数.
• boolean hasQueuedThreads() 是否有线程正在等待获取许可证.
• void reducePermits(int reduction) 减少reduction个许可证.是个protected方法.
• Collection getQueuedThreads() 返回所有等待获取许可证的线程集合.是个protected方法.

使用案例

acquire()Get a single license

/** * 线程数量 */    private final static int threadCount = 15;    public static void main(String[] args) throws Exception {        ExecutorService exec = Executors.newCachedThreadPool();		        final Semaphore semaphore = new Semaphore(3);        for (int i = 0; i < threadCount; i++) {            final int threadNum = i;            exec.execute(() -> {                try {                    //获取一个许可                    semaphore.acquire();                    test(threadNum);                    //释放一个许可                    semaphore.release();                } catch (Exception e) {                    log.error("exception", e);                }            });        }        exec.shutdown();    }    private static void test(int threadNum) throws Exception {        // 模拟请求的耗时操作        Thread.sleep(1000);        log.info("{}", threadNum);    }

输出结果：

According to the time of the output result, it can be seen that the same time can only be at most3个线程执行,符合预期

acquire(int permits)获取多个许可

/** * 线程数量 */    private final static int threadCount = 15;    public static void main(String[] args) throws Exception {        ExecutorService exec = Executors.newCachedThreadPool();        //信号量设置为3,That is, the maximum concurrency is 3,同时只允许3Threads are licensed        final Semaphore semaphore = new Semaphore(3);        for (int i = 0; i < threadCount; i++) {            final int threadNum = i;            exec.execute(() -> {                try {                    //获取多个许可                    semaphore.acquire(3);                    test(threadNum);                    //释放多个许可                    semaphore.release(3);                } catch (Exception e) {                    log.error("exception", e);                }            });        }        exec.shutdown();    }    private static void test(int threadNum) throws Exception {        // 模拟请求的耗时操作        Thread.sleep(1000);        log.info("{}", threadNum);    }

输出结果：

设置了3个许可,Each fetch per thread3个许可,So there can only be at the same time1个线程执行 .

tryAcquire()尝试获取一个许可,如果未获取到,不等待,The thread will be discarded without execution

/** * 线程数量 */    private final static int threadCount = 15;    public static void main(String[] args) throws Exception {        ExecutorService exec = Executors.newCachedThreadPool();        //信号量设置为3,That is, the maximum concurrency is 3,同时只允许3Threads are licensed        final Semaphore semaphore = new Semaphore(3);        for (int i = 0; i < threadCount; i++) {            final int threadNum = i;            exec.execute(() -> {                try {                    //尝试获取一个许可,如果未获取到,不等待,The thread will be discarded without execution                    if(semaphore.tryAcquire()) {                        test(threadNum);                        //释放许可                        semaphore.release();                    }                } catch (Exception e) {                    log.error("exception", e);                }            });        }        exec.shutdown();    }    private static void test(int threadNum) throws Exception {        // 模拟请求的耗时操作        Thread.sleep(1000);        log.info("{}", threadNum);    }

输出结果：

从输出可以看到,在3个线程获取到3个许可后,Because each thread calls the method to be executed1秒中,One of the earliest licenses is also in1S后释放,剩下的17threads did not acquire permission,使用了semaphore.tryAcquire()方法,没有设置等待时间,So it was immediately discarded,不执行了.

tryAcquire(long timeout, TimeUnit unit)未获取到许可,设置等待时长

/** * 线程数量 */    private final static int threadCount = 15;    public static void main(String[] args) throws Exception {        ExecutorService exec = Executors.newCachedThreadPool();        //信号量设置为3,That is, the maximum concurrency is 3,同时只允许3Threads are licensed        final Semaphore semaphore = new Semaphore(3);        for (int i = 0; i < threadCount; i++) {            final int threadNum = i;            exec.execute(() -> {                try {                    //The waiting time for obtaining a license is set to 2秒,Threads that are still unlicensed after two seconds are not executed                    if(semaphore.tryAcquire(2000, TimeUnit.MILLISECONDS)) {                        test(threadNum);                        //释放许可                        semaphore.release();                    }                } catch (Exception e) {                    log.error("exception", e);                }            });        }        exec.shutdown();    }    private static void test(int threadNum) throws Exception {        // 模拟请求的耗时操作        Thread.sleep(1000);        log.info("{}", threadNum);    }

输出结果：

tryAcquirespecified by parameters2秒的等待时间. The above code executes at most at the same time3个.第4thread because of the former3Thread execution takes one second without releasing the license,因此需要等待. 但是由于设置了2秒的等待时间,所以在5seconds to wait until the release of the permission,继续执行,循环往复.但是15个线程 ,每秒并发3个,2Sis not complete.So the above is executed to the first6个（0开始,显示是5）就结束了,【The result of each execution will be different,取决于CPU】,Not all are executed15个线程.