likes
comments
collection

AQS同步组件-Semaphore(信号量)解析和案例

作者站长头像
站长
· 阅读数 13

基本概念

Semaphore也是一个线程同步的辅助类,可以维护当前访问自身的线程个数,并提供了同步机制。使用Semaphore可以控制并发访问资源的线程个数。 AQS同步组件-Semaphore(信号量)解析和案例 例如排队买票的情况,如果只有三个窗口,那么同一时间最多也只能有三个人买票。第四个人来了之后就必须在后面等着,只有其他人买好了,才可以去相应的窗口进行买票 。

作用和使用场景

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

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

源码分析

构造函数

/**
*接受一个整型的数字,表示可用的许可证数量。Semaphore(10)表*示允许10个线程获取许可证,
*也就是最大并发数是10。 
*
* @param permits 可用许可证的初始数量。
**/
public Semaphore(int permits) {
        sync = new NonfairSync(permits);
}

/**
 * 使用给定的许可数量和给定的公平性设置
 *
 * @param permits 可用许可证的初始数量。
 *
 * @param fair 指定是公平模式还是非公平模式,默认非公平模式 . 公平模式:先启动的线程优先得到
 * 许可。 非公平模式:先启动的线程并不一定先获得许可,谁抢到谁就获得许可。
 */
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);
    }

输出结果: AQS同步组件-Semaphore(信号量)解析和案例 根据输出结果的时间可以看出来同一时间最多只能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);
    }

输出结果: AQS同步组件-Semaphore(信号量)解析和案例 设置了3个许可,每个线程每次获取3个许可,因此同一时间只能有1个线程执行 。

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

输出结果: AQS同步组件-Semaphore(信号量)解析和案例 从输出可以看到,在3个线程获取到3个许可后,因为每个线程调用的方法要执行1秒中,最早的一个许可也要在1S后释放,剩下的17个线程未获取到许可,使用了semaphore.tryAcquire()方法,没有设置等待时间,所以便直接被丢弃,不执行了。

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

输出结果: AQS同步组件-Semaphore(信号量)解析和案例 tryAcquire通过参数指定了2秒的等待时间。 上述代码中同一时间最多执行3个。第4个线程因前3个线程执行需要耗时一秒未释放许可,因此需要等待。 但是由于设置了2秒的等待时间,所以在5秒内等待到了释放的许可,继续执行,循环往复。但是15个线程 ,每秒并发3个,2S是执行不完的。所以上面执行到第6个(0开始,显示是5)就结束了,【每次执行结果会有差异,取决于CPU】,并没有全部执行完15个线程。

本文内容到此结束了,

如有收获欢迎点赞👍收藏💖关注✔️,您的鼓励是我最大的动力。

如有错误❌疑问💬欢迎各位大佬指出。

主页共饮一杯无的博客汇总👨‍💻

保持热爱,奔赴下一场山海。🏃🏃🏃

AQS同步组件-Semaphore(信号量)解析和案例