编程语言
首页 > 编程语言> > Semaphore源码解析

Semaphore源码解析

作者:互联网

Semaphore

Semaphore信号量,许可,用于控制在一段时间内,可并发访问执行的线程数量,基于AQS实现。

获取许可,支持公平和非公平模式,默认非公平模式。公平模式无论是否有许可,都会判断是否线程在排队,如果有线程排队,获取线程立即失败,进入排队;非公平模式无论许可是否充足,直接尝试获取许可。

可用于网关限流,或者资源限制(如最大可发起连接数)。

由于release()释放许可时,未对释放许可数做限制,所以可以通过该方法增加总的许可数量;reducePermits()方法可以减少总的许可数量,通过这两个方法可以达到动态调整许可的目的。

内部类同步器

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;
        }
    }
}

构造函数

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

/**
 * @param permits 总许可数
 * @param fair fair=true 公平锁  fair=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 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 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;
    }
}

标签:return,许可,permits,int,current,源码,Semaphore,解析,final
来源: https://www.cnblogs.com/QullLee/p/12247743.html