AQS同步组件-Semaphore（信号量）解析和案例

AQS同步组件-Semaphore（信号量）解析和案例

@[toc]

基本概念

作用和使用场景

用于保证同一时间并发访问线程的数目。 信号量在操作系统中是很重要的概念，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); }

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

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

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

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。