Semaphone应用&源码分析

发布时间:2023年12月23日

3.1 Semaphore介绍

sync,ReentrantLock是互斥锁,保证一个资源同一时间只允许被一个线程访问Semaphore(信号量)保证1个或多个资源可以被指定数量的线程同时访问底层实现是基于AQS去做的。
Semaphore底层也是基于AQS的state属性做一个计数器的维护。state的值就代表当前共享资源的个数。如果一个线程需要获取的1或多个资源,直接查看state的标识的资源个数是否足够,如果足够的,直接对state - 1拿到当前资源。如果资源不够,当前线程就需要挂起等待。知道持有资源的线程释放资源后,会归还给Semaphore中的state属性,挂起的线程就可以被唤醒。
Semaphore也分为公平和非公平的概念。
使用场景:连接池对象就可以基础信号量去实现管理。在一些流量控制上,也可以采用信号量去实现。再比如去迪士尼或者是环球影城,每天接受的人流量是固定的,指定一个具体的人流量,可能接受10000人,每有一个人购票后,就对信号量进行--操作,如果信号量已经达到了0,或者是资源不足,此时就不能买票。

3.2 Semaphore应用

以上面环球影城每日人流量为例子去测试一下。

public static void main(String[] args) throws InterruptedException {
// 今天环球影城还有人个人流量
Semaphore semaphore = new Semaphore(10);
new Thread(() -> {
System.out.println("一家三口要去~~");
try {
 semaphore.acquire(3);
System.out.println("一家三口进去了~~~");
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("一家三口走了~~~");
semaphore.release(3);
}
}).start();
for (int i = 0; i < 7; i++) {
int j = i;
new Thread(() -> {
System.out.println(j + "大哥来了。");
try {
semaphore.acquire();
System.out.println(j + "大哥进去了~~~");
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println(j + "大哥走了~~~");
semaphore.release();
}
}).start();
}
 Thread.sleep(10);
System.out.println("main大哥来了。");
if (semaphore.tryAcquire()) {
System.out.println("main大哥进来了。");
}else{
System.out.println("资源不够,main大哥进来了。");
}
Thread.sleep(10000);
System.out.println("main大哥又来了。");
if (semaphore.tryAcquire()) {
System.out.println("main大哥进来了。");
semaphore.release();
}else{
System.out.println("资源不够,main大哥进来了。");
}
}

其实Semaphore整体就是对构建Semaphore时,指定的资源数的获取和释放操作
获取资源方式:
● acquire():获取一个资源,没有资源就挂起等待,如果中断,直接抛异常
● acquire(int):获取指定个数资源,资源不够,或者没有资源就挂起等待,如果中断,直接抛异常
● tryAcquire():获取一个资源,没有资源返回false,有资源返回true
● tryAcquire(int):获取指定个数资源,没有资源返回false,有资源返回true
● tryAcquire(time,unit):获取一个资源,如果没有资源,等待time.unit,如果还没有,就返回false
● tryAcquire(int,time,unit):获取指定个数资源,如果没有资源,等待time.unit,如果还没有,就返回false
● acquireUninterruptibly():获取一个资源,没有资源就挂起等待,中断线程不结束,继续等
● acquireUninterruptibly(int):获取指定个数资源,没有资源就挂起等待,中断线程不结束,继续等
归还资源方式:
● release():归还一个资源
● release(int):归还指定个数资源

3.3 Semaphore源码分析

先查看Semaphore的整体结构,然后基于获取资源,以及归还资源的方式去查看源码

3.3.1 Semaphore的整体结构

Semaphore内部有3个静态内类。
首先是向上抽取的Sync
其次还有两个Sync的子类NonFairSync以及FairSync两个静态内部类
Sync内部主要提供了一些公共的方法,并且将有参构造传入的资源个数,直接基于AQS提供的setState方法设置了state属性。
NonFairSync以及FairSync区别就是tryAcquireShared方法的实现是不一样。

3.3.2 Semaphore的非公平的获取资源

在构建Semaphore的时候,如果只设置资源个数,默认情况下是非公平。
如果在构建Semaphore,传入了资源个数以及一个boolean时,可以选择非公平还是公平。

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

从非公平的acquire方法入手
首先确认默认获取资源数是1个,并且acquire是允许中断线程时,抛出异常的。获取资源的方式,就是直接用state - 需要的资源数,只要资源足够,就CAS的将state做修改。如果没有拿到锁资源,就基于共享锁的方式去将当前线程挂起在AQS双向链表
中。如果基于doAcquireSharedInterruptibly拿锁成功,会做一个事情。会执行**setHeadAndPropagate**方法。一会说

// 信号量的获取资源方法(默认获取一个资源)
public void acquire() throws InterruptedException {
// 跳转到了AQS中提供共享锁的方法
sync.acquireSharedInterruptibly(1);
}
// AQS提供的
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
// 判断线程的中断标记位,如果已经中断,直接抛出异常
if (Thread.interrupted())
throw new InterruptedException();
 // 先看非公平的tryAcquireShared实现。
// tryAcquireShared:
// 返回小于0,代表获取资源失败,需要排队。
// 返回大于等于0,代表获取资源成功,直接执行业务代码
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
// 信号量的非公平获取资源方法
final int nonfairTryAcquireShared(int acquires) {
// 死循环。
for (;;) {
// 获取state的数值,剩余的资源个数
int available = getState();
// 剩余的资源个数 - 需要的资源个数
int remaining = available - acquires;
// 如果-完后,资源个数小于0,直接返回这个负数
if (remaining < 0 ||
// 说明资源足够,基于CAS的方式,将state从原值,改为remaining
compareAndSetState(available, remaining))
return remaining;
}
}
// 获取资源失败,资源不够,当前线程需要挂起等待
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
// 构建Node节点,线程和共享锁标记,并且到AQS双向链表中
final Node node = addWaiter(Node.SHARED);
 boolean failed = true;
try {
for (;;) {
// 拿到上一个节点
final Node p = node.predecessor();
// 如果是head.next,就抢一手
if (p == head) {
// 再次基于非公平的方式去获取一次资源
int r = tryAcquireShared(arg);
// 到这,说明拿到了锁资源
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null;
failed = false;
return;
}
}
// 如果上面没拿到,或者不是head的next节点,将前继节点的状态改为-1,并挂起当前线程
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
// 如果线程中断会抛出异常
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}

acquire()以及acquire(int)的方式,都是执行acquireSharedInterruptibly方法去尝试获取资源,区别只在于是否传入了需要获取的资源个数。
tryAcquire()以及tryAcquire(int因为这两种方法是直接执行tryAcquire,**只使用非公平的实现**,只有非公平的情况下,才有可能在有线程排队的时候获取到资源
但是tryAcquire(int,time,unit)这种方法是正常走的AQS提供的acquire。因为这个tryAcquire可以排队一会,即便是公平锁也有可能拿到资源。这里的挂起和acquire挂起的区别仅仅是挂起的时间问题。
● acquire是一直挂起直到线程中断,或者线程被唤醒。
● tryAcquire(int,time,unit)是挂起一段时间,直到线程中断,要么线程被唤醒,要么阻塞时间到了
还有acquireUninterruptibly()以及acquireUninterruptibly(int)只是在挂起线程后,不会因为线程的中断而去抛出异常

文章来源:https://blog.csdn.net/qq_45309297/article/details/135060106
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。