AQS同步组件Semaphore信号量案例剖析
基本概念
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);
}
输出结果：
tryAcquire通过参数指定了2秒的等待时间。 上述代码中同一时间最多执行3个。第4个线程因前3个线程执行需要耗时一秒未释放许可，因此需要等待。
但是由于设置了2秒的等待时间，所以在5秒内等待到了释放的许可，继续执行，循环往复。
但是15个线程 ，每秒并发3个，2S是执行不完的。所以上面执行到第6个（0开始，显示是5）就结束了，【每次执行结果会有差异，取决于CPU】，并没有全部执行完15个线程。
以上就是AQS同步组件Semaphore信号量案例剖析的详细内容，更多关于AQS同步组件Semaphore的资料请关注其它相关文章！
相关文章