给力!Java 并发之 Semaphore 源码剖析!
Doocs开源社区
共 1038字,需浏览 3分钟
·
2020-08-23 11:28
本文精选自 Doocs 开源社区旗下“源码猎人”项目,作者 AmyliaY。
项目将会持续更新,欢迎 Star 关注。
项目地址:https://github.com/doocs/source-code-hunter
Semaphore 信号量,可用于控制一定时间内,并发执行的线程数,基于 AQS 实现。可应用于网关限流、资源限制 (如 最大可发起连接数)。由于 release()
释放许可时,未对释放许可数做限制,所以可以通过该方法增加总的许可数量。
获取许可支持公平和非公平模式,默认非公平模式。公平模式无论是否有许可,都会先判断是否有线程在排队,如果有线程排队,则进入排队,否则尝试获取许可;非公平模式无论许可是否充足,直接尝试获取许可。
不多废话,下面直接挖源码。
核心内部类 Sync
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 1192457210091910933L;
/* 赋值state为总许可数 */
Sync(int permits) {
setState(permits);
}
/* 剩余许可数 */
final int getPermits() {
return getState();
}
/* 自旋 + CAS非公平获取 */
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
// 剩余可用许可数
int available = getState();
// 本次获取许可后,剩余许可
int remaining = available - acquires;
// 如果获取后,剩余许可大于0,则CAS更新剩余许可,否则获取失败失败
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
/**
* 自旋 + CAS 释放许可
* 由于未对释放许可数做限制,所以可以通过release动态增加许可数量
*/
protected final boolean tryReleaseShared(int releases) {
for (;;) {
// 当前剩余许可
int current = getState();
// 许可更新值
int next = current + releases;
// 如果许可更新值为负数,说明许可数量溢出,抛出错误
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
// CAS更新许可数量
if (compareAndSetState(current, next))
return true;
}
}
/* 自旋 + CAS 减少许可数量 */
final void reducePermits(int reductions) {
for (;;) {
// 当前剩余许可
int current = getState();
// 更新值
int next = current - reductions;
// 较少许可数错误,抛出异常
if (next > current) // underflow
throw new Error("Permit count underflow");
// CAS更新许可数
if (compareAndSetState(current, next))
return;
}
}
/* 丢弃所有许可 */
final int drainPermits() {
for (;;) {
int current = getState();
if (current == 0 || compareAndSetState(current, 0))
return current;
}
}
}
/**
* 非公平模式
*/
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898L;
NonfairSync(int permits) {
super(permits);
}
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
}
/**
* 公平模式
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = 2014338818796000944L;
FairSync(int permits) {
super(permits);
}
/**
* 公平模式获取许可
* 公平模式不论许可是否充足,都会判断同步队列中是否有线程在等地,如果有,获取失败,排队阻塞
*/
protected int tryAcquireShared(int acquires) {
for (;;) {
// 如果有线程在排队,立即返回
if (hasQueuedPredecessors())
return -1;
// 自旋 + cas获取许可
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
}
主要 API
public class Semaphore implements java.io.Serializable {
private static final long serialVersionUID = -3222578661600680210L;
/** All mechanics via AbstractQueuedSynchronizer subclass */
private final Sync sync;
/**
* 根据给定的 总许可数permits,创建 Semaphore
*/
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
/**
* fair为true表示使用公平锁模式,false使用非公平锁
*/
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
// --------------------- 获取许可 --------------------
/* 获取指定数量的许可 */
public void acquire(int permits) throws InterruptedException {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireSharedInterruptibly(permits);
}
/* 获取一个许可 */
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0) // 获取许可,剩余许可>=0,则获取许可成功,<0获取许可失败,进入排队
doAcquireSharedInterruptibly(arg);
}
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
/**
* @return 剩余许可数量。非负数,获取许可成功,负数,获取许可失败
*/
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
/**
* 获取许可失败,当前线程进入同步队列,排队阻塞
*/
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 创建同步队列节点,并入队
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
// 如果当前节点是第二个节点,尝试获取锁
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 阻塞当前线程
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
// --------------------- 释放归还许可 -------------------------
/* 释放指定数量的许可 */
public void release(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.releaseShared(permits);
}
/* 释放一个许可 */
public void release() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
// 归还许可成功
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
/**
* 释放许可
* 由于未对释放许可数做限制,所以可以通过release动态增加许可数量
*/
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
private void doReleaseShared() {
// 自旋,唤醒等待的第一个线程(其他线程将由第一个线程向后传递唤醒)
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
// 唤醒第一个等待线程
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
}
全文完!
希望本文对大家有所帮助。如果感觉本文有帮助,有劳转发或点一下“在看”!让更多人收获知识!
长按识别下图二维码,关注公众号「Doocs 开源社区」,第一时间跟你们分享好玩、实用的技术文章与业内最新资讯。
评论