给力！Java 并发之 Semaphore 源码剖析！
本文精选自 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) // overflowthrow 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) // underflowthrow 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 GCfailed = 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) // overflowthrow 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 changedbreak;}}}
全文完！
希望本文对大家有所帮助。如果感觉本文有帮助，有劳转发或点一下“在看”！让更多人收获知识！
长按识别下图二维码，关注公众号「Doocs 开源社区」，第一时间跟你们分享好玩、实用的技术文章与业内最新资讯。
评论