AQS源码简单理解二：加锁

悠悠 2022-10-27 06:15 158阅读 0赞

（手头的活忙完了，来补一篇博客）[上篇提到][Link 1]多线程并发但却没有竞争的时候，AQS只是多执行了一行代码而已，没有线程队列，更没有重量级锁。所以它比优化前的`synchronized()` 效率高些。

这篇主要介绍如果多线程且有竞争，AQS 是怎么处理的。

AQS通过**自旋**、**CAS**、**park** 三种方法结合使用 ，尽量将多线程同步放在 JVM 层完成，实在搞不定了，再创建重量级锁实现多线程同步。

# 加锁核心代码 #

这里与公平锁为例（[公平与非公平的区别在这里][Link 2]）。

顺着 `reentrantLock.lock()` 一直往下走到 `AbstractQueuedSynchronizer.class` 中的 `acquire(1)`。（其实是 `FairSync.class` 中，因为被继承了 ）

public final void acquire(int arg) { 
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

这块是加锁的核心代码，它做了这么几件事情：

1.  `tryAcquire(arg)` : 线程尝试拿锁。
2.  （如果没有拿到） `addWaiter(Node.EXCLUSIVE)` 把当前线程包装成线程节点放到同步器的队列中， “addWaiter” ：“添加等待者”等待者的意思，挺顺耳。
3.  `acquireQueued(addWaiter,arg)` : 同步器对列中的线程尝试拿锁，（也不能一直躺里面呀）毕竟还是要出队执行代码块的。可能是刚入队的线程再尝试拿锁，如果还拿不到就“沉睡”，也可能是被唤醒的线程尝试拿锁。这个后面再分析。

## tryAcquire(arg) ##

protected final boolean tryAcquire(int acquires) { 
        // 获取当前线程。
        final Thread current = Thread.currentThread();
        // 获取同步器状态。
        int c = getState();
        // state == 0 表示同步器中没有线程正在运行代码
        if (c == 0) { 
            if (!hasQueuedPredecessors() &&         // 当前线程是否需要排队。
                compareAndSetState(0, acquires)) {   // cas 尝试加锁（设置 state == 1）。
                setExclusiveOwnerThread(current);   // 加锁成功，将当前线程赋给 exclusiveOwnerThread。
                return true;
            }
        }
        // 锁重入。判断当前线程是不是排它锁的持有者。
        // 这里可以证明 reentrantlLock 是支持锁重入的。
        else if (current == getExclusiveOwnerThread()) { 
            // 重入数 + 1
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            // 更新 state 的值，此时 state > 1。
            setState(nextc);
            return true;
        }
        return false;
    }

上面代码执行加锁和锁重入，有几个问题需要解释下：

### 1. “加锁”到底是做了什么事情，怎么才算“加锁”了？ ###

“锁”在哪呢？是 reentrantLock 对象么？好像点不太准确。所谓代码 codeA 被加锁是说，线程在执行 codeA 时得先去看一个对象 obj，看 obj 中的标志 flag（obj 中的一个变量） 有没有被改变，如果没改，那线程就可以执行 codeA，如果被改了，表示 codeA 正在被另一个线程执行着呢，当前线程先得等等，等到 obj 的 flag 改回原值了，当前线程才能执行 codeA，这样就实现多线程同步了。

obj 对象就被称为“锁”或者“锁对象”，flag 没有被改表示锁是自由的，flag 被改变了表示锁是不自由的。回到代码，在 ReentrantLock 中，`new ReentrantLock()` 出来的对象就是锁，它的 `state` 字段充当 flag 的角色（该字段在 reentrantLock 对象的 sync 字段里面），`state == 0` 表示锁是自由的，`state > 0` 表示锁不自由。在 `synchronize(obj1) {}` 中，obj1 就是锁，它的 markword 扮演了 flag 的作用（[详细介绍在这里][Link 3]）。

代码块被加锁，其实是锁对象的 flag 变了，线程A 正在执行加锁代码块，紧接着线程B 也要执行这段代码， B看到锁对象的 flag 值变了，就知道“代码块正在被别的线程执行着，咱得等等”。线程A 执行完代码块后将锁的 flag 改回原值（释放锁），线程B 看到 flag 变原值了，“咱可以执行了…”。

上面提到了 “线程B 要等等”，该怎么等呢？放在后面介绍。

### 2. 当前线程怎么判断它需不需要排队？ ###

1.  多线程并发时，任何时刻只允许一个线程执行代码块，其他线程都得在同步器的队列中等着。
2.  公平锁（公平同步器）讲究论资排辈、先进先出，越先入队的线程越先尝试拿锁（拿锁成功，就去执行加锁代码块）。所以，越靠近队头的线程（节点）越先拿锁。
3.  因为只允许一个线程执行代码块，再加上队列的第一个节点是空节点（作为对头使用），所以队列的第二个节点拿锁的优先级最高，后面的节点全是等待的。

所以即便是当前线程看到 `state == 0` ,它还需要判断下自己需不需要入队，因为这会可能不是它的出场时间。再看下源码：

// AbstractQueuedSynchronizer.class
    public final boolean hasQueuedPredecessors() { 
        Node t = tail; 
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }

先有个意识，当前线程能得到 `state == 0` 就证明没有线程正在执行加锁代码块。h 指向队头节点，t 指向队尾节点。再明确下：hasQueuedPredecessors 返回 true 表示要排队，返回 false 表示不用排队。多线程并发开始了…，我们看下几种情况：

1.  T1 是第一个线程。此时队列为空， `head == tail == null` ，`h != t` 不成立，方法返回false，不用排队。
2.  T1 正在执行加锁代码块，T2 也开执行了。你放心，T2 走不到这里，因为T1 没释放锁，这会 `state == 1` , T2 会通过 执行 `addWaiter()` 直接入队，最后的结果是同步器队列中有两个节点，一个是空的头结点，后面紧跟着 T2 节点。
3.  T3 也开始执行了，恰好碰到 T1 执行结束，T3 得到了 `state == 0` 。此时队列不空，`h != t` 成立。后半截为真则排队，后半截为假则不排队。h.next 表示队列的第二个节点，所以 S 表示 T2 节点 ，`(s = h.next) == null` 不成立，当前线程是 T3 线程，s.thread 是 T2 线程，`s.thread != Thread.currentThread()` 成立，所以后半截是真。整个方法返回真，所以 T3 需要排队。
4.  T1 执行结束，T2 被唤醒（你放心，唤醒的绝对是 T2， 这里不考虑等待线程被取消，后面在介绍[锁释放时，会说到唤醒][Link 4]）。唤醒时队列中有三个节点，空的头节点，T2 和 T3。T2 得到 `state == 0` , `h != t` 成立，`s == h.next == T2` 节点，`(s = h.next) == null` 不成立，s.thread == Thread.currentThread() == T2，所以 `s.thread != Thread.currentThread()` 不成立，则后半截不成立，那么整个方法返回 fasle。T2 不排队，可以直接拿锁。（脑子要清醒下，这块的代码都在 `reentrantLock.lock()` 中执行的，开发者写的业务逻辑在这个方法的下面。）拿到锁后是要出队的，所以现在队列值剩头节点和 T3 节点了。
5.  T2 执行结束，T3 被唤醒。和 T2 的流程是一样的。T3 出队后，队列中只剩了个队头节点。（“剩了队头节点”，这个描述不是很准确，应该是原来的队头被 gc 回收了，新队头是新产生的。新队头怎么来的？跟锁释放有关）。

## addWaiter(Node.EXCLUSIVE) ##

线程先执行 `tryAcquire(1)` 尝试加锁，如果加锁失败，则为自己构建一个线程节点，并且追加到队列中去。

// AbstractQueuedSynchronizer.class
    private Node addWaiter(Node mode) { 
        // 为当前线程构建一个线程节点。
        Node node = new Node(Thread.currentThread(), mode);
        Node pred = tail;
        // 队列已经被初始化了（已经有线程在队列中等待了）。
        if (pred != null) { 
            node.prev = pred;
            // cas 将当前线程节点追加到队尾。
            // 如果 cas 成功，则入队结束。如果失败，走下面的 enq(node) 再入队
            if (compareAndSetTail(pred, node)) { 
                pred.next = node;
                return node;
            }
        }
        // 队列未被初始化（当前线程是第一个需要等待的线程）
        // 或者上面的 cas 失败，要用过 enq(node)的方式重新入队。
        enq(node);
        return node;
    }

node 里面保存着当前线程，它入队需要分两种情况考虑。

（1）如果node 不是第一个要等待的线程，那么队列是已经被初始化了，只需要使用 cas 将node 追加到队尾，如果 cas 成功，则修改节点引用的指向，入队就结束了。最后 return node。如果 cas 失败，则走下面的 enq(node) 再入队。

（2）如果 node 是第一个要等待的线层，那么当下队列是不存在的，直接执行`enq(node)` 创建队列。最后再return node。

// AbstractQueuedSynchronizer.class
    private Node enq(final Node node) { 
        for (;;) { 
            Node t = tail;
            if (t == null) {  // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else { 
                node.prev = t;
                if (compareAndSetTail(t, node)) { 
                    t.next = node;
                    return t;
                }
            }
        }
    }

根据上面的入队情况，`enq(final Node node)` 有两种执行流程：

（1）队列非空，而且 `enq()` 外围的 cas 失败了，采用 enq 再次入队。

// AbstractQueuedSynchronizer.class
    private Node enq(final Node node) { 
        // 死循环
        for (;;) { 
            // 拿到队尾引用
            Node t = tail;
            // 队列非空，所以不会进 if 代码块
            if (t == null) { 
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else { 
                // 死循环 + cas，构成了标准的乐观锁。
                // 只有追加成功，才结束循环。
                node.prev = t;
                if (compareAndSetTail(t, node)) { 
                    t.next = node;
                    return t;
                }
            }
        }
    }

这种还是很简单的，死循环 + cas 构成了乐观锁，只有队列追加成功时，才结束循环。

（2）队列是空，要先初始化队列，再追加线程节点。

// 循环了两次，创建了两个节点。一个空节点做头，后面追加一个正真的线程节点。
    private Node enq(final Node node) { 
        for (;;) { 
            // 第一次循环，tail == null。
            // 第二次循环，tail == node0。
            Node t = tail;
            // 第一次循环，同步器的队列是空的，所以 head == tail == null。
            if (t == null) {  // Must initialize
                // new Node() 创建了 node0 ，但是该节点是空的（里面没有线程）。
                // compareAndSetHead 操作，将 node0 设置成队列头，并且 head 指向 node0。
                if (compareAndSetHead(new Node()))
                    // 同步器的 head 和 tail 引用都指向 node0。
                    tail = head;
            } else { 
                // 第二次循环才会到这里，此时 t == node0。
                // 线程节点的前驱引用指向 node0 。
                node.prev = t;
                // 将当前线程节点追加到(设置到)队尾。
                if (compareAndSetTail(t, node)) { 
                    t.next = node;
                    return t;
                }
            }
        }
    }

到这里，入队算是走完了。如果队列被初始化那直接追加，如果没有队列，先创建队列，再追加节点。

## acquireQueued(final Node node, int arg) ##

上面介绍了尝试拿锁，线程入队。重点是只有一个线程能执行加锁代码，其他的都停住了，他们都是在哪里停的呢？除此以外，怎么停下来？已经阻塞又被唤醒的节点接下来该怎么办？什么时候可以再拿锁？前面这些问题都 `acquireQueued(final Node node, int arg)` 方法控制。源码如下：

// AbstractQueuedSynchronizer.class
    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);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }

这个方法的入口有两个地方，第一个是对于刚入队的线程节点，从入口参数开始往下执行。第二个是对于队列中已经”睡眠“ 但被唤醒的线程节点，从第 14 行开始执行。下面详细走一遍代码的执行过程。

（1）对于刚入队的线程

// AbstractQueuedSynchronizer.class
    final boolean acquireQueued(final Node node, int arg) { 
        // 是否拿锁成功。
        boolean failed = true;
        try { 
            // 是否被打断。
            boolean interrupted = false;
            // 死循环
            for (;;) { 
                // 获取当前节点的前驱节点。
                final Node p = node.predecessor();
                // 如果 P==head 成立，那证明 当前 node 是队列的第二个节点。
                // (前面说过)第二个节点是最优先拿锁的节点。所以，如果当前节点
                // 是第二个节点，那就再尝试拿一次锁。
                if (p == head && tryAcquire(arg)) { 
                    // 尝试拿锁成功。
                    // 设置新的头结点（后面有解释）
                    setHead(node);
                    // 为了让 GC 回收原来的头节点。
                    p.next = null; 
                    failed = false;
                    return interrupted;
                }
                // 如果当前节点不是第二个节点，或者拿锁失败，
                // 那就要使当前线程再等等，那“等等”具体是自旋下，还是
                // 直接创建重量级索直接“睡”呢。
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }

对于刚入队的线程，它如果是队列的第二个节点，那它可以再尝试拿一次锁。因为某一时刻只能有一个线程执行加锁代码块，公平锁论资排辈，先进先出，队列中的第二个节点具有最高的拿锁资格。

如果拿锁成功：

// AbstractQueuedSynchronizer.class
    private void setHead(Node node) { 
        // 队头引用指向当前节点。
        head = node;
        // 将节点中的线程置空。
        node.thread = null;
        // 断开当前节点与前驱的指向。
        node.prev = null;
    }

从这个可以看出，如果第二个节点拿锁成功，那么第二个节点会变成新的队头（队头是空节点）。原来队头会被垃圾回收器回收掉（与它相关的引用都被置为 null）。最终的结果是第二个节点拿锁成功后，它也就“出队”了。

如果拿锁失败或者当前节点不是第二个节点：

这两种情况下当前线程都应该等待。稍微想一下，如果不是第二个节点，当前线程当然应该乖乖地去排在后面等待（公平锁是讲秩序的）。如果是第二个节点，但是拿锁失败，就表示加锁代码块正在被别的线程执行着，还没完呢，当前线程站在第二节点的位置上再等等。

### shouldParkAfterFailedAcquire(Node pred, Node node) 该不该 park？ ###

问题来了，“等等”是自旋下，还是直接创建重量级锁“睡”呢？

具体咋办，由 `shouldParkAfterFailedAcquire(p, node)` 来决定。这个函数名字翻译过来是“拿锁失败后应该 park 吗？”，看看里面怎么写的。

// AbstractQueuedSynchronizer.class
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            return true;
        if (ws > 0) { 
            do { 
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else { 
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

#### 先解释下 waitStatus 是什么意思？ ####

waitStatus 是 `AbstractQueuedSynchronizer.Node.class` 的属性，它表示等待线程的状态，取值有以下几种：

*  0：默认值，Node 创建时 waitSatus 的默认值。
 *  1： 表示线程被取消了。
 *  \-1：表示当前线点处于 park 状态，正在阻塞着。
 *  \-2：表示线程处于条件等待。
 *  \-3：用在共享锁里面。

#### park 还是 自旋？ ####

判断当前线程是该自旋还是该 park ？`shouldParkAfterFailedAcquire` 返回 true 表示 park，返回 fasle 表示自旋。

结论：当前节点是自旋还是park，取决于它的前驱节点的 waitSatus 的值：

*  \-1：当前线程 park。
 *  1：当前线程自旋。
 *  0：当前线程自旋。

具体的看下代码：

// node 是当前节点，pred 是当前节点的前驱节点。
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
        // 获取前驱的 ws
        int ws = pred.waitStatus;
        // Node.SIGNAL == -1
        if (ws == Node.SIGNAL)
            return true; // 当前节点park。
        // ws > 0, 其实是判断 ws == 1 是不是成立。
        if (ws > 0) { 
            // 如果前驱是的 ws == 1，那表示前驱线程被取消了。
            // do{}while{} 循环一直往前找，直到找到没有被取消的线程。
            // 修改节点的前后引用的指向，将当前节点追加到没有被取消的节点的后面。
            // 也是就是说，那些“被取消的节点”被从队列中清理出去了。
            // if{} 执行完会直接跳到最后的 return false。
            do { 
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else { 
            // 如果 ws == 0，将前驱的节点的 ws 设为 -1，后面就是 return false。
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        // 当前节点自旋。
        return false;
    }

#### 实际场景中是怎么走的？ ####

上面的代码虽说写了注解，估计看起来还是很迷糊。假设几种场景，看看代码到底是怎么走的。多线程并发开始啦…

（1）T1 正在执行加锁代码块，T2 也开始执行了：

T2 执行`tryAcquire(1)` 第一次尝试拿锁失败，执行 `addWaiter()` 入队，创建了两个节点，队列头 node0 和 node2，而且这两个 node 的 `waitstatus` 字段都是 0。执行 `acquireQueued(node2, 1)`, 因为 node2 是第二个节点，所以再尝试拿一次锁，当然依旧拿锁失败。执行 `shouldParkAfterFailedAcquire(node0, node2)`, 看下源码的怎么走的：

// pred == node0, node == node2
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
        // ws == 0
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            return true;
        if (ws > 0) { 
            do { 
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else { 
            // 直接到这里，设置 node0 的 ws == -1
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        // 返回 false，表示当前线程自旋。
        return false;
    }

外围是个 for 死循环，所以 node2 在第二个节点的位置上继续尝试拿锁，可依旧还是失败，又一次进到这个方法里。

// pred == node0, node == node2
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
        // ws == 1
        int ws = pred.waitStatus;
        // 成立
        if (ws == Node.SIGNAL)
            // 返回 true，表示当前线程应该 park。
            return true;
        if (ws > 0) { 
            do { 
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

如果再有T3、T4 启动，会分别重复 T2 的过程，将它的前驱节点的 waitStatus 标记成 -1，自己再自旋一次，第二次再进入 shouldParkAfterFailedAcquire() 返回 true。

最后 node0、node2、node3、node4的中的 waitStatus 值分别是 -1、-1、-1、0。

（2）T1 依旧在执行，T2、T4 被取消了，但 T5 开始执行了：

这里我们先不管 T2、T4 是怎么被取消的。此时可以确定，node0、node2、node3、node4的中的 waitStatus 值分别是 -1、1、-1、1。此时代码的执行流程：

// pred == node4, node == node5
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
        // ws == 1
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            return true;
        if (ws > 0) { 
            do { 
                //循环执行完后，队列中的节点顺序是：node0、node2、node3、node5。
                // (node3 在被唤醒后 尝试拿锁，第一次会失败，在判断是否需要自旋时，将node2 踢出队列，node3 自旋一次，再拿锁成功)
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else { 
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        // 返回 false, node5 再自旋一次。
        return false;
    }

**很明显，刚入队线程都会自旋一次，其目的还是谋求在 jvm 层面解决同步，尽量不要创建重量级锁。**

### parkAndCheckInterrupt() 在这里 park ###

当 `shouldParkAfterFailedAcquire(Node pred, Node node)` 返回 true 时，`parkAndCheckInterrupt()` 方法就真正的阻塞线程，看下源码：

// AbstractQueuedSynchronizer.class
    private final boolean parkAndCheckInterrupt() { 
        // 为当前线程创建重量级锁。
        LockSupport.park(this);
        return Thread.interrupted();
    }

线程执行到这里被阻塞住了，不再执行后面的代码。线程被唤醒（打断）时，也会从这里开始执行。

（2）对于被唤醒的线程  
线程在哪里 park 的，当它被唤醒时就从那里继续执行。（再贴一遍代码）

被唤醒的线程从 `parkAndCheckInterrupt()` 方法开始执行。因为在 for 循环里面，继续判断是不是队列中第二个节点，尝试拿锁，如果成功就出队，如果失败，就继续 park。  
**更多唤醒细节上的内容跟线程解锁有关，[在下一篇博客写][Link 4]。**

到这里整个加锁过程就结束了，只是调用了一句 `reentrantLock.lock()`,内部的执行还是很复杂的。

[Link 1]: https://blog.csdn.net/yy_diego/article/details/113095068
[Link 2]: https://blog.csdn.net/yy_diego/article/details/113095068#t8
[Link 3]: https://blog.csdn.net/yy_diego/article/details/112434339
[Link 4]: https://blog.csdn.net/yy_diego/article/details/113646162