Java 동기화 장치 AQS 아키텍처는 어떻게 잠금을 해제하고 대기열을 동기화합니까?

Introduction

AQS의 내용이 너무 많아서 AQS 전반부를 읽지 않은 학생들이 다시 살펴볼 수 있도록 전반부에서는 AQS의 기본 개념에 대해 많이 이야기합니다. 잠금 기본 속성, 잠금 획득 방법 등 이 장에서는 잠금 및 동기화 대기열을 해제하는 방법에 대해 주로 설명합니다.

1. 잠금 해제

잠금 해제의 트리거 시간은 일반적으로 사용되는 Lock.unLock() 메서드입니다. 목적은 스레드가 리소스에 대한 액세스를 해제할 수 있도록 하는 것입니다(전체 아키텍처 다이어그램의 보라색 경로 참조). 과정).

잠금 해제도 두 가지로 나뉘는데, 하나는 전용 잠금 해제이고, 다른 하나는 공유 잠금 해제입니다.

1.1.배타적 잠금 해제

배타적 잠금 해제는 비교적 간단합니다. 대기열의 선두에서 시작하여 다음 노드를 찾으면 끝에서 시작하여 다음 노드를 찾습니다. 상태가 취소된 노드가 아니고 노드를 해제합니다. 소스 코드는 다음과 같습니다:

// 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. 공유 잠금 해제 releaseShared

공유 잠금을 해제하는 방법은 크게 두 가지로 나뉩니다. 단계:

tryReleaseShared는 현재 공유 잠금을 해제하려고 시도하고 실패하면 false를 반환합니다. Go 2 성공적으로

스레드가 공유 잠금을 얻을 때 이 방법을 본 적이 있습니다. 잠금을 해제하면 뒤에 있는 노드가 활성화됩니다. 메소드 이름은 doReleaseShared입니다.

releaseShared의 소스코드를 살펴보겠습니다:

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

2. 조건부 큐의 중요한 메소드

조건부 큐의 메소드를 살펴보기 전에 먼저 동기화 큐가 존재하는 이유와 조건부 큐가 존재하는 이유를 이해해야 합니다. 필요해?

주로 동기화된 대기열로 모든 시나리오를 처리할 수는 없기 때문에 잠금 + 대기열이 결합된 시나리오가 발생하면 먼저 잠금을 사용하여 잠금을 얻을 수 있는 스레드와 대기열에 넣어야 하는 스레드가 필요합니다. 잠금을 획득한 여러 스레드가 대기열이 가득 차거나 비어 있는 경우 조건을 사용하여 이러한 스레드를 관리하고 이러한 스레드가 차단 및 대기한 다음 적절한 시간에 정상적으로 깨어날 수 있습니다.

동기화 대기열 + 조건부 대기열을 함께 사용하는 시나리오는 잠금 + 대기열 시나리오에서 가장 자주 사용됩니다.

그래서 조건부 큐도 빼놓을 수 없는 부분입니다.

다음으로 조건 대기열의 더 중요한 메서드 중 일부를 살펴보겠습니다. 다음 메서드는 모두 ConditionObject 내부 클래스에 있습니다.

2.1. 대기열에 들어가서 wait

// 线程入条件队列
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 메서드를 기다립니다. 특별한 주의가 필요한 몇 가지 사항이 있습니다.

위 코드는 위치 A를 표시합니다. 노드가 조건부 대기열에 들어갈 준비가 되기 전에 먼저 해제해야 합니다. 그렇지 않으면 조건 대기열이 조건 대기열에 있고 다른 스레드는 잠금을 얻을 수 없습니다. 위의 코드는 위치 B를 표시합니다. 이때 노드는 Condition.signal 또는 signalAll 메서드는 이 시점에서 노드가 동기화 큐로 성공적으로 전송되었으므로(아키텍처 다이어그램의 전체 파란색 프로세스), 소스 코드는 Waiter를 사용하여 조건에 노드 이름을 지정하는 것을 좋아합니다. 따라서 조건 큐에 실제로는 Node인 Waiter가 표시됩니다.

await 메서드에는 addConditionWaiter와 unlinkCancelledWaiters라는 두 가지 중요한 메서드가 있습니다. 하나씩 살펴보겠습니다.

2.1.1, addConditionWaiter

addConditionWaiter 메소드는 주로 조건 큐에 노드를 넣습니다. 메소드 소스 코드는 다음과 같습니다.

// 增加新的 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;
}

전체 프로세스는 큐의 끝에 추가하는 것입니다. unlinkCancelledWaiters라는 중요한 메소드입니다. 이 메소드 상태가 CONDITION이 아닌 조건 대기열의 모든 노드는 다음과 같이 unlinkCancelledWaiters 메소드의 소스 코드를 살펴보겠습니다. 이 방법에 대한 모든 사람의 이해를 돕기 위해 다음과 같은 설명 다이어그램이 그려집니다.

2.2. 단일 깨우기 신호

신호 방법은 이전에 대기열이 꽉 찼을 때 일부 스레드를 의미합니다. take 작업으로 인해 조건부 큐에서 차단되었으며 갑자기 큐의 요소가 스레드 A에 의해 소비되었습니다. 스레드 A는 이전에 차단된 스레드를 깨우기 위해 신호 메서드를 호출합니다. 조건의 헤드 노드에서 깨어납니다. queue(프로세스의 전체 아키텍처 다이어그램에서 파란색 부분 참조) 소스 코드는 다음과 같습니다.

// 会检查尾部的 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;
    }
}

가장 중요한 메서드인 transferForSignal을 살펴보겠습니다.

// 唤醒阻塞在条件队列中的节点
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);
}

전체 소스 코드를 읽어보면 조건 큐에 있는 노드를 깨우면 실제로 조건 큐에 있는 노드가 동기화 큐로 전송되고 이전 노드의 상태가 SIGNAL로 설정되는 것을 알 수 있습니다.

2.3. 모두 깨우기 signalAll

signalAll은 조건 대기열의 모든 노드를 깨우는 데 사용됩니다. 소스 코드는 다음과 같습니다.

// 返回 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;
}

소스 코드에서 볼 수 있듯이 그 본질은 for 루프 호출입니다. 조건 대기열의 노드를 순환하는 transferForSignal 메소드. 동기화 대기열로 전송합니다.

3. 요약

AQS 소스 코드가 드디어 완성되었습니다. 이해가 되셨나요? 이제 AQS 아키텍처 다이어그램을 조용히 떠올려 보실 수 있습니다.

위 내용은 Java 동기화 장치 AQS 아키텍처는 어떻게 잠금을 해제하고 대기열을 동기화합니까?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!