AQS源码浅析(5)——锁方法

acquireQueued条件等待方法及独占模式获取

• 调用acquireQueued方法时，至少有一个node（就是参数中的node）在队列
• 同时可能有【新入队的线程】或者【由中断唤醒的线程】会与【当前线程】产生竞争
• 当前驱节点是head时，就是出队的时机
• 获取锁失败后会调用shouldParkAfterFailedAcquire（详细见下文）方法
• 如果退出自旋却未获得锁，说明出现了未知的异常，需要调用cancelAcquire（详细见下文）方法将当前节点取消

final boolean acquireQueued(final Node node, int arg) { //自旋获得锁，会阻塞当前线程
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();//前驱节点
            if (p == head && tryAcquire(arg)) {//前驱节点是头节点就获取锁
                setHead(node); //成功获取锁（其他线程失败，都会阻塞），当前node出队
                p.next = null; // help GC
                failed = false;
                return interrupted; //如果在竞争锁的过程中发生了中断，返回true
            }
            //线程获取锁失败后的操作
            if (shouldParkAfterFailedAcquire(p, node) &&//判断线程获取锁失败后，是否应该阻塞
                parkAndCheckInterrupt()) //阻塞线程，并在被唤醒后检查是否被中断
                interrupted = true;
        }
    } finally {
        if (failed) // tryAcquire发生未知的错误导致线程退出自旋，shouldParkAfterFailedAcquire、parkAndCheckInterrupt应该不会出错
            cancelAcquire(node); //取消当前线程
    }
}

shouldParkAfterFailedAcquire检查和更新获取锁失败的节点的状态

• 是所有acquire循环中的主要控制信号，检查并更新未能获取锁的节点的状态，要求pred必须是node的前驱节点
• 如果线程前驱节点状态是SIGNAL，那么可以安全的应该阻塞，返回 true
• 如果线程状态为CANCELLED，去除前面所有的已取消的节点（head节点一定不是CANCELLED）返回false
• 如果节点状态是0或者PROPAGATE的时候，将前驱状态设置为SIGNAL，那么下次循环获取锁失败后进入这个方法时，将会返回true
• 也就是说线程如果获取锁失败，在正常情况下前驱节点都应该是SIGNAL，代表自己准备好被唤醒了

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL) //pred节点是SIGNAL状态表示node应该被唤醒，当前node可以安全的阻塞，返回true
        /*
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        return true;
    if (ws > 0) { //跳过CANCELLED状态的前驱节点
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {//pred的waitStatus一定是0 或者 PROPAGATE，说明当前节点不需要阻塞，调用者需要重试
        /*
         * waitStatus must be 0 or PROPAGATE.  Indicate that we
         * need a signal, but don't park yet.  Caller will need to
         * retry to make sure it cannot acquire before parking.
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL); //前驱节点状态设置为SIGNAL
    }
    return false;
}

cancelAcquire取消节点尝试acquire

取消当前节点，如果节点在队列的中间就从队列中去掉，否则直接唤醒一个有效的后继节点

private void cancelAcquire(Node node) {
    // Ignore if node doesn't exist
    if (node == null)
        return;
    node.thread = null;//清除线程对象
    // Skip cancelled predecessors
    Node pred = node.prev;
    while (pred.waitStatus > 0) //跳过已取消的前驱节点
        node.prev = pred = pred.prev;
    // predNext is the apparent node to unsplice. CASes below will
    // fail if not, in which case, we lost race vs another cancel
    // or signal, so no further action is necessary.
    Node predNext = pred.next;
    // Can use unconditional write instead of CAS here.
    // After this atomic step, other Nodes can skip past us.
    // Before, we are free of interference from other threads.
    node.waitStatus = Node.CANCELLED;//不会有其他线程干扰，这一步后其他线程会跳过
    // If we are the tail, remove ourselves.
    if (node == tail && compareAndSetTail(node, pred)) { //如果node已经在队尾，设置尾节点
        compareAndSetNext(pred, predNext, null);//更新pred的next指针为null
    } else {
        // If successor needs signal, try to set pred's next-link
        // so it will get one. Otherwise wake it up to propagate.
        int ws;
        if (pred != head && //确定node前驱节点不是head，说明现在node已经不在队首了
            ((ws = pred.waitStatus) == Node.SIGNAL ||
             (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && //前驱状态小于0，或者将未取消的前驱设置为SIGNAL状态（标记后继节点需要唤醒）
            pred.thread != null) {//前驱拥有线程对象
            Node next = node.next;//if判断前驱节点不是虚节点
            if (next != null && next.waitStatus <= 0) //判断后继有效
                compareAndSetNext(pred, predNext, next);//连接node的前驱和后继
        } else {//如果node的前驱是虚节点，唤醒一个后继
            unparkSuccessor(node);
        }
        node.next = node; // help GC
    }
}

cancelAcquire与shouldParkAfterFailedAcquire一样，都有一段跳过CANCELLED状态的代码如下：

Node pred = node.prev;
while (pred.waitStatus > 0) //跳过已取消的前驱节点
    node.prev = pred = pred.prev;

为什么这里不需要用CAS呢？

• 首先只有线程可以把自己的节点状态设置为CANCELLED
• 其他的线程执行这段代码时，会直接跳过CANCELLED状态的节点
• 所以取消的线程与其他线程不会互相干扰

这也是cancelAcquire的逻辑：

• 如果头部同时有好几个节点取消，那么唤醒操作交给next链靠后的取消节点即可
• 如果在队列中间取消，那么只需要最靠后的节点连接未取消的前驱和后驱即可
• 如果在队列尾部取消，那么只需要最靠前的节点将前驱设置为尾部即可

doAcquireNanos和doAcquireInterruptibly

• doAcquireNanos与acquireQueued的区别是加入了超时时间限制，超时直接取消节点
• doAcquireInterruptibly与acquireQueued的区别是响应中断，发生中断直接取消节点并抛中断异常

private boolean doAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                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);
    }
}
private void doAcquireInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

doAcquireShared共享不可中断模式获取锁

共享模式的acquireQueued，doAcquireShared与acquireQueued的区别是会传播唤醒后继节点

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);//node变成头节点，传播唤醒后继
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

doAcquireSharedInterruptibly和doAcquireSharedNanos

• doAcquireSharedNanos与doAcquireShared的区别是加入了超时时间限制，超时直接取消节点
• doAcquireSharedInterruptibly与doAcquireShared的区别是响应中断，发生中断直接取消节点并抛中断异常

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