Java并发编程之Semaphore

秒速五厘米 2023-10-09 23:24 8阅读 0赞

前言

Semaphore是一种计数信号量，用来限制访问资源的线程数量，通过acquire方法获取许可require后访问资源进行操作，操作结束调用release方法来释放许可，当同时获取许可的线程数大于设置的许可数量时，没获取到许可的线程就进入队列进行等待。

获取许可

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)
            doAcquireSharedInterruptibly(arg);
    }
    复制代码

通过Semaphore的acquire方法来获取许可，会调用同步器的acquireSharedInterruptibly(1)方法，这个方法是可中断的，所以如果被中断了就直接抛出异常。获取许可的流程主要分为两步：

*  尝试获取许可
 *  获取许可失败处理

尝试获取许可

获取许可时根据公平和非公平模式有不同的实现

[非公平获取许可][Link 1]

protected int tryAcquireShared(int acquires) {
        return nonfairTryAcquireShared(acquires);
    }
    final int nonfairTryAcquireShared(int acquires) {
        for (;;) {
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }
    复制代码

首先获取state的值，也就是当前可用的许可数量，然后减去要获取的许可数量，得到最后剩余的许可数remaining，如果remaining < 0表示已经没有可用的许可了，当前线程获取许可失败直接返回remaining；如果remaining >= 0表示当前线程可以获取许可，就通过CAS更新剩余的许可数量state，更新失败就一直重试，更新成功就返回剩余的许可数。

公平获取许可

protected int tryAcquireShared(int acquires) {
        for (;;) {
            if (hasQueuedPredecessors())
                return -1;
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }
    // 队列中是否有其他节点
    public final boolean hasQueuedPredecessors() {
        Node t = tail; // Read fields in reverse initialization order
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }
    复制代码

公平模式获取许可会先检查等待队列中是否有在等待的节点，遵循先来后到的原则，如果队列中有等待的节点的话当前线程就没有资格去获取许可，所以就直接返回-1表示获取许可失败。

如果队列中没有等待的节点，当前线程就可以去获取许可了，这里的流程就和非公平模式一样了，参考非公平获取许可

获取许可失败处理

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);
        }
    }
    复制代码

根据上面获取锁的流程可以分析出这个方法的入参是-1，代表两种情况：

*  许可用完了
 *  公平模式下队列中有等待的线程

进行失败处理的流程主要分为两部分：

*  在队列中添加节点
 *  尝试获取许可

添加节点

private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {  // 添加到队尾
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }
     private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // 队列初始化
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }
    复制代码

首先把当前线程封装成Node节点对象，然后判断如果当前有尾节点tail的话，就尝试把当前线程的节点添加到尾节点后面成为新的尾节点。

如果没有尾节点tail的话说明队列还没有初始化，就调用enq方法先初始化队列然后再把当前线程的节点添加到尾节点后面成为新的尾节点。

获取许可

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();
    }
    复制代码

如果前驱节点是head节点，也就是当前节点是第二个节点表示有资格获取许可，就重新调用tryAcquireShared尝试去获取锁，获取成功会返回可用的许可数量这里一定是大于等于0，然后修改头节点head并判断是否需要唤醒后继节点。

[唤醒后继节点][Link 1]

// 修改head节点并判断是否需要唤醒后继节点
    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head;
        setHead(node);                                                 
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }
    复制代码

// 尝试唤醒后继节点
    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;
        }
    }
    // 唤醒后继节点
    private void unparkSuccessor(Node node) {
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);                         // ③
        Node s = node.next;
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }
    复制代码

setHeadAndPropagate方法的目的是修改head节点并尝试唤醒后继节点。

setHeadAndPropagate方法的参数propagate表示可用的许可数量，这里传进来的取值范围是propagate>=0，当大于0时表示有可用许可，尝试去唤醒后继节点就行了。

如果传进来的propagate=0，不就表示没有许可了吗？为什么还要继续判断后面的条件再去唤醒后继节点呢？因为如果此时刚好有其他线程释放了许可，那实际许可的数量就大于0了，所以还是可以去尝试唤醒后继节点。

特殊情况

考虑下面一种情况：

如果当前线程t1是被唤醒的，也就是会执行上面的doReleaseShared中的①和unparkSuccessor中的③，所以头节点head的waitStatus会被修改成0。

如果此时有其他线程t2释放了许可，就会尝试去唤醒后继节点，也就是会执行到doReleaseShared，但是此时头节点head的waitStatus被t1修改成了0，所以t2会执行到②把head的waitStatus修改为Node.PROPAGATE（-3）。

> 这里可以理解为：我释放了一个许可然后想唤醒一个节点，但是因为其他线程把标记waitStatus改了，所以我就不能去唤醒了，我只能把标记waitStatus再改成Node.PROPAGATE，让其他线程有能力去唤醒的时候看一下这个标记，如果是Node.PROPAGATE就去唤醒。

然后再回过头来看propagate=0之后的条件判断：

*  h == null || h.waitStatus < 0

> 说明t2是在setHead方法之前修改的waitStatus，所以满足这里的h.waitStatus < 0，因为更新了head，所以原来的h可能会被回收，h == null。

*  (h = head) == null || h.waitStatus < 0

> 说明t2是在setHead方法之后修改的waitStatus，所以满足这里的h.waitStatus < 0，而这里新的head==null可能是有其他节点重新成为了head节点，当前的head节点被回收。

释放许可

public void release() {
        sync.releaseShared(1);
    }
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }
    复制代码

释放许可调用的是AQS的releaseShared方法，流程很简单主要分为两步：

*  尝试释放许可
 *  唤醒后继节点

尝试释放许可

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;
        }
    }
    复制代码

首先获取当前许可数量，然后加上释放的许可数量，结果如果比之前还少了说明数据溢出了，直接抛出错误。 否则就尝试更新许可数量。

唤醒后继节点

[Link 1]: https://link.juejin.cn?target=