AQS同步组件Semaphore信号量案例剖析

AQS同步组件Semaphore AQS Semaphore信号量 2022-11-13 14:11:19 170人浏览泡泡鱼

Python 官方文档：入门教程 => 点击学习

摘要

目录基本概念作用和使用场景源码分析构造函数常用方法使用案例acquire()获取单个许可acquire(int permits)获取多个许可tryAcquire()获取许可tryAc

基本概念

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

例如排队买票的情况，如果只有三个窗口，那么同一时间最多也只能有三个人买票。第四个人来了之后就必须在后面等着，只有其他人买好了，才可以去相应的窗口进行买票。

作用和使用场景

用于保证同一时间并发访问线程的数目。
信号量在操作系统中是很重要的概念，java并发库里的Semaphore就可以很轻松的完成类似操作系统信号量的控制。Semaphore可以很容易控制系统中某个资源被同时访问的线程个数。
在数据结构中我们学过链表，链表正常是可以保存无限个节点的，而Semaphore可以实现有限大小的列表。

使用场景：仅能提供有限访问的资源。比如数据库连接。

源码分析

构造函数


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

public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

常用方法

acquire() 获取一个许可

acquire(int permits) 获取指定个数的许可

tryAcquire()方法尝试获取1个许可证

tryAcquire(long timeout, TimeUnit unit) 最大等待许可的时间

tryAcquire(int permits) 获取指定个数的许可

tryAcquire(int permits, long timeout, TimeUnit unit) 最大等待许可的时间

availablePermits() ：返回此信号量中当前可用的许可证数

release() 释放许可

release(int permits) 释放指定个数的许可

int getQueueLength() 返回正在等待获取许可证的线程数。

boolean hasQueuedThreads() 是否有线程正在等待获取许可证。

void reducePermits(int reduction) 减少reduction个许可证。是个protected方法。

Collection getQueuedThreads() 返回所有等待获取许可证的线程集合。是个protected方法。

使用案例

acquire()获取单个许可


    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);
    }

输出结果：

根据输出结果的时间可以看出来同一时间最多只能3个线程执行，符合预期

acquire(int permits)获取多个许可


    private final static int threadCount = 15;
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        //信号量设置为3，也就是最大并发量为3，同时只允许3个线程获得许可
        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个许可，每个线程每次获取3个许可，因此同一时间只能有1个线程执行。

tryAcquire()获取许可

tryAcquire()尝试获取一个许可,如果未获取到，不等待，将直接丢弃该线程不执行


    private final static int threadCount = 15;
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        //信号量设置为3，也就是最大并发量为3，同时只允许3个线程获得许可
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    //尝试获取一个许可,如果未获取到，不等待，将直接丢弃该线程不执行
                    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个许可后，因为每个线程调用的方法要执行1秒中，最早的一个许可也要在1S后释放，剩下的17个线程未获取到许可，使用了semaphore.tryAcquire()方法，没有设置等待时间，所以便直接被丢弃，不执行了。

tryAcquire(long timeout, TimeUnit unit)

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


    private final static int threadCount = 15;
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newCachedThreadPool();
        //信号量设置为3，也就是最大并发量为3，同时只允许3个线程获得许可
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    //设置了获取许可等待时间为2秒，如果两秒后还是未获得许可的线程便得不到执行
                    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);
    }

输出结果：