AQS之共享锁分析 (三)

文章目录

• ​​1.AQS 内部体系架构​​
• ​​2.获取资源​​

• ​​2.1 尝试失败后的处理 doAcquireShared()​​
• ​​2.2 处理队列中的节点 setHeadAndPropagate()​​

• ​​3.释放资源​​

• ​​3.1 尝试成功后的处理 doReleaseShared()​​

• ​​4.其他获取资源​​

• ​​4.1 响应中断 acquireSharedInterruptibly()​​
• ​​4.2 超时 doAcquireSharedNanos()​​

1.AQS 内部体系架构

• FairSync: 公平锁
• NoFairSync: 非公平锁

• Shared: 共享模式
• Exclusive: 排他模式

2.获取资源

尝试获取共享资源, 返回值为整数, 负数为失败, 0为成功, 但是其他线程无法再成功, 正数为成功, 其他线程也可以成功。
由于共享锁允许多个线程同时获取成功，因此可以用 返回值代表还能有几个线程可以继续获取资源，但并不是强制性的。

2.1 尝试失败后的处理 doAcquireShared()

private void doAcquireShared(int arg) {
    //加到队列里，和排斥锁一样
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    //成功，处理队列中的节点
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    //和排斥锁一样，不响应中断
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            //和排斥锁一样
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        //和排斥锁一样
        if (failed)
            cancelAcquire(node);
    }
}

获取资源成功后，需要执行setHeadAndPropagate(node, r);

2.2 处理队列中的节点 setHeadAndPropagate()

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    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();
    }
}

1. 设置新的头结点，把获取到资源的当前节点设置为头结点，即头结点是最后一个获取到资源的线程节点
2. 唤醒节点
   （1）三个条件满足其一
   a. 尝试获取资源的返回值propagate 大于0，还有其他线程有希望获取到资源
   b. 原头结点或新头结点是null或者状态小于0，说明有其他操作进行了修改
   c.状态小于0说明是SIGNAL或者PROPAGATE，即使PROPAGATE也有可能会在其他的操作中变成了SIGNAL，因此需要进行唤醒。
   （2）下一个节点为共享节点或者null

3.释放资源

3.1 尝试成功后的处理 doReleaseShared()

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            //signal需要唤醒阻塞线程
            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;
    }
}

对头节点的处理。

if (h == head) 的判断则说明如果处理过程中头结点发生变化则需要重试。

4.其他获取资源

4.1 响应中断 acquireSharedInterruptibly()

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

在尝试获取资源和阻塞结束后检测到中断后抛出异常。

4.2 超时 doAcquireSharedNanos()

private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
            throws InterruptedException {
    //检测超时时间
    if (nanosTimeout <= 0L)
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;
    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 true;
                }
            }
            nanosTimeout = deadline - System.nanoTime();
            //检测超时时间
            if (nanosTimeout <= 0L)
                return false;
            if (shouldParkAfterFailedAcquire(p, node) &&
                //如果时间太短了，直接自旋而不是阻塞，提高性能
                nanosTimeout > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

1. 排队获取资源时先判断是否已经超时
2. 每次判断是否需要阻塞时都先判断是否超时
3. 阻塞时增加检查，时间不足只进行自旋，减少阻塞的性能消耗
4. 阻塞时阻塞指定时间