给力!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);    }    /**     * fairtrue表示使用公平锁模式,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 开源社区」,第一时间跟你们分享好玩、实用的技术文章与业内最新资讯。



浏览 41
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

举报