AQS源码探究_03 成员方法解析(加锁资源竞争逻辑)

Posted 2021-05-21 兴趣使然の草帽路飞

文章参考：小刘老师源码

AQS成员方法解析

1. lock加锁方法

// 位于ReentrantLock类的静态内部类Sync中：加锁方法
final void lock() {
    // 令当前线程去竞争资源
    acquire(1);
}

2. acquire令当前线程竞争资源的方法

// 位于AQS下的acquire：令当前线程去竞争资源的方法
public final void acquire(int arg) {
    // 条件1：!tryAcquire(arg)方法 尝试获取锁，获取成功返回true，获取失败返回false
    // 条件2.1：addWaiter方法 将当前线程封装成node入队
    // 条件2.2：入队后调用 acquireQueued方法 (该方法包含挂起当前线程、以及线程唤醒后相关的逻辑)
    // 		   (令当前线程不断去竞争资源，直到成功获取锁才停止自旋)
    // acquireQueued方法返回boolean类型,true:表示挂起过程中线程中断唤醒过，false:表示未被中断唤醒过
    if (!tryAcquire(arg) && 
        // Node.EXCLUSIVE 当前节点是独占模式
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

3. tryAcquire尝试获取锁的方法

// 位于ReentrantLock类的静态内部类Sync中：位于尝试获取锁的方法，不会阻塞线程
// 返回true -> 尝试获取锁成功 | 返回false -> 尝试获取锁失败
protected final boolean tryAcquire(int acquires) {
    // 当前线程
    final Thread current = Thread.currentThread();
    // AQS中的state(加锁状态)值
    int c = getState();
    // 如果条件成立：c == 0 表示当前AQS处于无锁状态
    if (c == 0) {
        // 因为fairSync是公平锁，任何时候都需要检查一下在当前线程之前，队列中是否有等待者
        // 条件1：hasQueuedPredecessors 判断FIFO队列是否为空 
        // true -> 表示当前线程前面有等待者线程，当前线程需要入队等待
        // false -> 表示当前线程前面没有等待者线程，直接可以尝试获取锁
        if (!hasQueuedPredecessors() &&
            // 条件2：compareAndSetState(0, acquires) 基于CAS去更新state的值
            // state更新成功：说明当前线程抢占锁成功！
            // state更新失败：说明多个线程存在竞争，当前线程竞争失败，未能抢到锁的持有权
            compareAndSetState(0, acquires)) {
            // 条件1、2均成立时：说明当前线程抢夺锁的持有权成功！
            // 设置当前线程为独占线程(锁的持有者线程)
            setExclusiveOwnerThread(current);
            // true -> 当前线程尝试获取锁成功
            return true;
        }
    }
    // current == getExclusiveOwnerThread()：用于判断当current ！=0 或者 >0 的情况下
    // 当前线程是否是持有锁的线程(独占线程)，因为ReentrantLock是可重入的锁，获取锁的线程可以再次进入~
    // 如果条件成立：说明当前线程就是独占锁的线程
    else if (current == getExclusiveOwnerThread()) {
        // 获取当前线程的加锁状态，并累加
        int nextc = c + acquires;
        // 越界判断...当冲入的深度很深时，会导致 nextc < 0，因为 int值达到MAX最大之后，再+1，会变复数
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        // 更新加锁状态
        setState(nextc);
        // true -> 当前线程尝试获取锁成功
        return true;
    }
    // false -> 尝试获取锁失败的情况：
    // 1.CAS加锁失败 且 当前线程前面有等待的线程
    // 2.state > 0 且 当前线程不是占用锁的线程
    return false;
}

4.addWaiter将当前线程添加到阻塞队列的方法

// 位于AQS下：将当前线程添加到阻塞队列的方法
// 最终返回包装当前线程的node
private Node addWaiter(Node mode) {
    // 构建Node,把当前线程封装到Node中，mode：Node节点的模式，例如Node.EXCLUSIVE 当前节点是独占模式
    Node node = new Node(Thread.currentThread(), mode);
    // 线程快速入队方式：
    // 获取队尾节点，保存到pred
    Node pred = tail;
    if (pred != null) {// 如果条件成立：说明队列中已经有node了
        // 令当前节点node的前驱等于pred
        node.prev = pred;
        // 基于CAS更新队尾tail
        if (compareAndSetTail(pred, node)) {
            // tail更新成功：前驱节点等于node，完成双向绑定
            pred.next = node;
            // 返回node
            return node;
        }
    }
    // 线程完整入队方式(自旋入队)：
    // 执行到这里有以下2种情况：
    // 1.tail == null 当前队列是空队列
    // 2.cas设置当前newNode 为 tail 时失败了，被其他线程抢先一步了
    // 自旋入队，只有入队成功才结束自旋：
    enq(node);
    
    // 返回node
    return node;
}

5. enq当前线程完整入队的方法(自旋入队)

private Node enq(final Node node) {
    // 自旋~ 只有封装当前线程的node入队成功，才会跳出循环
    for (;;) {
        Node t = tail;
        // 第1种情况：空队列 ===> 即，当前线程是第一个抢占锁失败的线程
        // 当前持有锁的线程(注：tryAcquire方法直接获取到锁的线程，在该方法逻辑中，并没有将持锁线程入队，
        // 而按理说阻塞队列的head节点就应该是当前持有锁的线程才对)并没有设置过任何 node，
        // 所以作为该线程的第一个后驱next，需要给它擦屁股（给持锁线程补一个node节点并设置为阻塞队列的head
        // head节点任何时候，都代表当前占用锁的线程）
        if (t == null) {
            // 如果compareAndSetHead条件成立：说明当前线程给当前持有锁的线程，补充head操作成功了！
            if (compareAndSetHead(new Node()))
                // tail = head 表示当前队列只有一个元素，这里就表名当前持锁的线程被放入阻塞队列且为head了~
                tail = head;
            	// 注意：并没有直接返回，还会继续自旋，下次再进入循环时阻塞队列已经不为空，且head为持锁线程节点了...
        } else {
            // 其他情况，说明：当前队列中已经有node了，这里是一个追加node的过程

            // 如何入队呢？和 addWaiter方法入队逻辑一样~
            // 1.找到newNode的前置节点 pred
            // 2.更新newNode.prev = pred
            // 3.CAS更新tail为 newNode
            // 4.更新 pred.next = newNode

            // 前置条件：队列已经有等待者node了(不为空)，当前node并不是第一个入队的node
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                // 如果条件成立，说明当前线程成功入队！
                t.next = node;
                // 注意：入队成功，一定要return终止无限for循环~
                // 返回这个节点t
                return t;
            }
        }
    }
}

6. acquireQueued真正去竞争资源的方法

acquireQueued需要做什么呢？

• 1.当前节点如果没有被park挂起，则 ===> 挂起当前线程。
• 2.线程唤醒后 ===> 需要做一些线程唤醒之后的逻辑。

// 位于AQS中：真正去竞争资源的方法
// 参数final Node node：封装当前线程的node，且当前时刻该node已经入队成功了
// 参数arg：当前线程抢占资源成功后，更新state值时要用到
// 返回true:表示挂起过程中线程中断唤醒过，返回false:表示未被中断唤醒过
final boolean acquireQueued(final Node node, int arg) {
    // true：表示当前线程抢占锁成功
    // false：表示当前线程抢占锁失败，需要执行出队逻辑
    boolean failed = true;
    try {
        // 当前线程是否被中断
        boolean interrupted = false;
        // 自旋~
        for (;;) {
            // 什么情况下回执行到这里？
            // 1.进入for循环时，在线程尚未被park前会执行
            // 2.线程park后，被唤醒之后也会执行
            
            // 获取当前节点的前驱节点
            final Node p = node.predecessor();
            // p == head 条件成立时：说明当前node为head的节点的后驱(head.next)，head.next在任何时候都有权利去争夺锁。
            // tryAcquire 尝试去获取锁，如果条件成立，说明head对应的线程已经释放锁了，而作为head的后驱节点的线程，刚好可以获取锁。
            // tryAcquire 如果条件不成立：说明head对应的线程尚未释放锁，而作为head的后驱节点的线程，这时候仍需要继续park挂起~
            if (p == head && tryAcquire(arg)) {
                // 拿到锁~
                // 设置封装当前线程的节点为head节点(head无论什么时候都是持锁线程的节点)
                setHead(node);
                // 将上一个线程对应的node的next引用设置为null，帮助GC回收。即，老head出队~
                p.next = null; // help GC
                // 当前线程获取锁的过程中，没有发生异常
                failed = false;
                // 返回当前线程的中断标记~
                return interrupted;
            }
            
            // shouldParkAfterFailedAcquire: 判断当前线程获取锁资源失败后，是否需要挂起
            // true: 需要挂起 | false：不需要挂起
            if (shouldParkAfterFailedAcquire(p, node) &&
                // parkAndCheckInterrupt: 挂起当前线程，并在该线程唤醒之后，返回当前线程的interrupted中断标记
                // 唤醒该线程的方式：
                // 1.正常唤醒：其他线程调用 unpark方法，唤醒该线程
                // 2.其他线程给当前挂起的线程一个中断信号(中断挂起)
                parkAndCheckInterrupt())
                // interrupted = true 表示当前node对应的线程是被中断信号唤醒的
                interrupted = true;
        }
    } finally {
        // 当failed为true时：
        if (failed)
            // node节点的取消线程资源竞争
            cancelAcquire(node);
    }
}

7.shouldParkAfterFailedAcquire方法

// 位于AQS中: 判断当前线程获取锁资源失败后，是否需要挂起
// true: 需要挂起 | false：不需要挂起
// 参数1：Node pred 当前线程的前驱节点
// 参数2：Node node 封装当前线程的节点
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 获取前驱节点的状态waitStatus
    // 0: 默认状态 | -1：Signal状态(表示当前节点释放锁后会唤醒它的第一个后驱节点) |
    // >0：表示当前节点是CANCELED状态
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)// 如果条件成立，则表示前驱节点是可以唤醒当前线程节点的节点
        // 返回true后，在acquireQueue方法中会继续调用parkAndCheckInterrupt方法去park当前线程节点
        // 注意：一般情况下，第一次来到shouldParkAfterFailedAcquire方法中时，ws不会是-1
        return true;
    // 如果ws>0条件成立：表示当前节点是CANCELED状态
    if (ws > 0) {
        // 该循环是一个找pred.waitStatus > 0 的前驱节点的过程：
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        // 找到符合条件的前驱节点后，令其下一个节点为当前线程的node
        // 隐含着一种操作：即，CANCELED状态的节点会被出队 
        pred.next = node;
    } else {
        // 当前node前驱节点的状态就是0，即默认状态这种情况
        // 将当前线程node的前驱节点的状态，强制设置为SIGNAL，表示该节点释放锁后会唤醒它的第一个后驱节点
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

// 位于AQS中：挂起当前线程节点
private final boolean parkAndCheckInterrupt() {
    // 线程挂起
    LockSupport.park(this);
    return Thread.interrupted();
}

总结：

• 1.如果当前节点的前置节点是 CANCELED取消状态，则：
  • 第1次来到这个方法时，会越过取消状态的节点。
  • 第2次返回true，然后park挂起当前线程。
• 2.如果当前节点的前置节点是 0 默认状态，则：
  • 当前线程会设置前置节点的状态为 -1
  • 第2次自旋来到这个方发时，会返回true，然后park挂起当前线程。