How does the java synchronizer AQS architecture release locks and synchronize queues

Introduction

AQS has too much content, so we divided it into two chapters. Students who have not read the first half of AQS can look back. Ha, the first half of the chapter talks about a lot of basic concepts of locks, basic properties, how to obtain locks, etc. In this chapter we mainly talk about how to release locks and synchronization queues.

1. Release the lock

The triggering time of releasing the lock is our commonly used Lock.unLock () method, the purpose is to let the thread release access to resources (see the purple route of the overall architecture diagram for the process ).

The release of locks is also divided into two categories, one is the release of exclusive locks, and the other is the release of shared locks. Let's look at them separately.

1.1. Release the exclusive lock release

The release of the exclusive lock is relatively simple. Start from the head of the queue and find its next node. If the next node is empty, just It will start from the end and find the node whose status is not canceled, and then release the node. The source code is as follows:

// unlock 的基础方法
public final boolean release(int arg) {
    // tryRelease 交给实现类去实现，一般就是用当前同步器状态减去 arg，如果返回 true 说明成功释放锁。
    if (tryRelease(arg)) {
        Node h = head;
        // 头节点不为空，并且非初始化状态
        if (h != null && h.waitStatus != 0)
            // 从头开始唤醒等待锁的节点
            unparkSuccessor(h);
        return true;
    }
    return false;
}
// 很有意思的方法，当线程释放锁成功后，从 node 开始唤醒同步队列中的节点
// 通过唤醒机制,保证线程不会一直在同步队列中阻塞等待
private void unparkSuccessor(Node node) {
    // node 节点是当前释放锁的节点，也是同步队列的头节点
    int ws = node.waitStatus;
    // 如果节点已经被取消了，把节点的状态置为初始化
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);
    // 拿出 node 节点的后面一个节点
    Node s = node.next;
    // s 为空，表示 node 的后一个节点为空
    // s.waitStatus 大于0，代表 s 节点已经被取消了
    // 遇到以上这两种情况，就从队尾开始，向前遍历，找到第一个 waitStatus 字段不是被取消的
    if (s == null || s.waitStatus > 0) {
        s = null;
        // 这里从尾迭代，而不是从头开始迭代是有原因的。
        // 主要是因为节点被阻塞的时候，是在 acquireQueued 方法里面被阻塞的，唤醒时也一定会在 acquireQueued 方法里面被唤醒，唤醒之后的条件是，判断当前节点的前置节点是否是头节点，这里是判断当前节点的前置节点，所以这里必须使用从尾到头的迭代顺序才行，目的就是为了过滤掉无效的前置节点，不然节点被唤醒时，发现其前置节点还是无效节点，就又会陷入阻塞。
        for (Node t = tail; t != null && t != node; t = t.prev)
            // t.waitStatus <= 0 说明 t 没有被取消，肯定还在等待被唤醒
            if (t.waitStatus <= 0)
                s = t;
    }
    // 唤醒以上代码找到的线程
    if (s != null)
        LockSupport.unpark(s.thread);
}

1.2. Release the shared lock releaseShared

The method to release the shared lock is releaseShared, which is mainly divided into Two steps:

tryReleaseShared tries to release the current shared lock. If it fails, it returns false. If it succeeds, go to 2;

Wake up the subsequent blocking node of the current node. We have seen this method before. The thread obtains the share When locking, it will wake up the node behind it. The method name is: doReleaseShared.

Let’s take a look at the source code of releaseShared:

// 共享模式下，释放当前线程的共享锁
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        // 这个方法就是线程在获得锁时，唤醒后续节点时调用的方法
        doReleaseShared();
        return true;
    }
    return false;
}

2. Important methods of conditional queue

Before looking at the methods of conditional queue, we must first understand why there are If you have a synchronization queue, do you still need a conditional queue?

Mainly because not all scenarios can be handled by a synchronization queue. When encountering a scenario where lock queues are combined, Lock Condition is required. First use Lock to determine which threads can obtain the lock. Which threads need to be queued and blocked in the synchronization queue; when the multiple threads that have obtained the lock encounter the queue being full or empty, you can use Condition to manage these threads, let these threads block and wait, and then be awakened normally at the appropriate time. .

Synchronization queues are used in conjunction with conditional queues, and are most often used in lock queue scenarios.

So the conditional queue is also an indispensable part.

Next let’s take a look at some of the more important methods of the condition queue. The following methods are all in the ConditionObject internal class.

2.1. Enter the queue and wait for await

// 线程入条件队列
public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 加入到条件队列的队尾
    Node node = addConditionWaiter();
    // 标记位置 A
    // 加入条件队列后，会释放 lock 时申请的资源，唤醒同步队列队列头的节点
    // 自己马上就要阻塞了，必须马上释放之前 lock 的资源，不然自己不被唤醒的话，别的线程永远得不到该共享资源了
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // 确认node不在同步队列上，再阻塞，如果 node 在同步队列上，是不能够上锁的
    // 目前想到的只有两种可能：
    // 1:node 刚被加入到条件队列中，立马就被其他线程 signal 转移到同步队列中去了
    // 2:线程之前在条件队列中沉睡，被唤醒后加入到同步队列中去
    while (!isOnSyncQueue(node)) {
        // this = AbstractQueuedSynchronizer$ConditionObject
        // 阻塞在条件队列上
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    // 标记位置 B
    // 其他线程通过 signal 已经把 node 从条件队列中转移到同步队列中的数据结构中去了
    // 所以这里节点苏醒了，直接尝试 acquireQueued
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        // 如果状态不是CONDITION，就会自动删除
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

await method has several points that need special attention:

The above code marks position A. Before the node is ready to enter the conditional queue, it must The currently held lock will be released first, otherwise it will enter the condition queue and other threads will not be able to obtain the lock. The above code marks position B. At this time, the node is awakened by the Condition.signal or signalAll method. At this time, the node has been It was successfully transferred to the synchronization queue (blue process in the overall architecture diagram), so the acquireQueued method can be executed directly; Node is named in the condition queue. The source code likes to use Waiter to name it, so we see in the condition queue Waiter is actually Node.

There are two important methods in the await method: addConditionWaiter and unlinkCancelledWaiters. Let’s take a look at them one by one.

2.1.1, addConditionWaiter

The addConditionWaiter method mainly puts the node into the condition queue. The method source code is as follows:

// 增加新的 waiter 到队列中，返回新添加的 waiter
// 如果尾节点状态不是 CONDITION 状态，删除条件队列中所有状态不是 CONDITION 的节点
// 如果队列为空，新增节点作为队列头节点，否则追加到尾节点上
private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    // 如果尾部的 waiter 不是 CONDITION 状态了，删除
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    // 新建条件队列 node
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    // 队列是空的，直接放到队列头
    if (t == null)
        firstWaiter = node;
    // 队列不为空，直接到队列尾部
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

The overall process is relatively simple, just append to the queue. At the end, there is an important method called unlinkCancelledWaiters. This method will delete all nodes in the condition queue whose status is not CONDITION. Let’s take a look at the source code of the unlinkCancelledWaiters method, as follows:

2.1.2, unlinkCancelledWaiters

// 会检查尾部的 waiter 是不是已经不是CONDITION状态了
// 如果不是，删除这些 waiter
private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    // trail 表示上一个状态,这个字段作用非常大，可以把状态都是 CONDITION 的 node 串联起来，即使 node 之间有其他节点都可以
    Node trail = null;
    while (t != null) {
        Node next = t.nextWaiter;
        // 当前node的状态不是CONDITION，删除自己
        if (t.waitStatus != Node.CONDITION) {
            //删除当前node
            t.nextWaiter = null;
            // 如果 trail 是空的，咱们循环又是从头开始的，说明从头到当前节点的状态都不是 CONDITION
            // 都已经被删除了，所以移动队列头节点到当前节点的下一个节点
            if (trail == null)
                firstWaiter = next;
            // 如果找到上次状态是CONDITION的节点的话，先把当前节点删掉，然后把自己挂到上一个状态是 CONDITION 的节点上
            else
                trail.nextWaiter = next;
            // 遍历结束，最后一次找到的CONDITION节点就是尾节点
            if (next == null)
                lastWaiter = trail;
        }
        // 状态是 CONDITION 的 Node
        else
            trail = t;
        // 继续循环，循环顺序从头到尾
        t = next;
    }
}

In order to make it easier for everyone to understand this method, I drew an interpretation diagram, as follows:

2.2. Single wake-up signal

The signal method is wake-up Meaning, for example, the queue was full before, and some threads were blocked in the conditional queue due to the take operation. Suddenly, the elements in the queue were consumed by thread A. Thread A will call the signal method to wake up the previously blocked threads, and then the thread will start. The head node of the conditional queue starts to wake up (see the blue part in the overall architecture diagram for the process). The source code is as follows:

// 唤醒阻塞在条件队列中的节点
public final void signal() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 从头节点开始唤醒
    Node first = firstWaiter;
    if (first != null)
        // doSignal 方法会把条件队列中的节点转移到同步队列中去
        doSignal(first);
}
// 把条件队列头节点转移到同步队列去
private void doSignal(Node first) {
    do {
        // nextWaiter为空，说明到队尾了
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        // 从队列头部开始唤醒，所以直接把头节点.next 置为 null，这种操作其实就是把 node 从条件队列中移除了
        // 这里有个重要的点是，每次唤醒都是从队列头部开始唤醒，所以把 next 置为 null 没有关系，如果唤醒是从任意节点开始唤醒的话，就会有问题，容易造成链表的割裂
        first.nextWaiter = null;
        // transferForSignal 方法会把节点转移到同步队列中去
        // 通过 while 保证 transferForSignal 能成功
        // 等待队列的 node 不用管他，在 await 的时候，会自动清除状态不是 Condition 的节点(通过 unlinkCancelledWaiters 方法)
        // (first = firstWaiter) != null  = true 的话，表示还可以继续循环， = false 说明队列中的元素已经循环完了
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

Let’s take a look at the most critical method: transferForSignal.

// 返回 true 表示转移成功， false 失败
// 大概思路：
// 1. node 追加到同步队列的队尾
// 2. 将 node 的前一个节点状态置为 SIGNAL，成功直接返回，失败直接唤醒
// 可以看出来 node 的状态此时是 0 了
final boolean transferForSignal(Node node) {
    /*
     * If cannot change waitStatus, the node has been cancelled.
     */
    // 将 node 的状态从 CONDITION 修改成初始化，失败返回 false
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;
    // 当前队列加入到同步队列，返回的 p 是 node 在同步队列中的前一个节点
    // 看命名是 p，实际是 pre 单词的缩写
    Node p = enq(node);
    int ws = p.waitStatus;
    // 状态修改成 SIGNAL，如果成功直接返回
    // 把当前节点的前一个节点修改成 SIGNAL 的原因，是因为 SIGNAL 本身就表示当前节点后面的节点都是需要被唤醒的
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        // 如果 p 节点被取消，或者状态不能修改成SIGNAL，直接唤醒
        LockSupport.unpark(node.thread);
    return true;
}

After reading the entire source code, we can see that waking up the nodes in the condition queue actually transfers the nodes in the condition queue to the synchronization queue and sets the status of its previous node to SIGNAL.

2.3. Wake up all signalAll

The function of signalAll is to wake up all nodes in the condition queue. The source code is as follows:

    public final void signalAll() {
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        // 拿到头节点
        Node first = firstWaiter;
        if (first != null)
            // 从头节点开始唤醒条件队列中所有的节点
            doSignalAll(first);
    }
    // 把条件队列所有节点依次转移到同步队列去
    private void doSignalAll(Node first) {
        lastWaiter = firstWaiter = null;
        do {
            // 拿出条件队列队列头节点的下一个节点
            Node next = first.nextWaiter;
            // 把头节点从条件队列中删除
            first.nextWaiter = null;
            // 头节点转移到同步队列中去
            transferForSignal(first);
            // 开始循环头节点的下一个节点
            first = next;
        } while (first != null);
    }

As can be seen from the source code, its essence is for The transferForSignal method is called in a loop to transfer the nodes in the condition queue to the synchronization queue.

3. Summary

The AQS source code is finally finished. Do you understand it? You can silently recall the AQS architecture diagram and see if you can understand it now.

The above is the detailed content of How does the java synchronizer AQS architecture release locks and synchronize queues. For more information, please follow other related articles on the PHP Chinese website!