JUC并发编程03——AQS条件队列源码剖析

逃离我推掉我的手 2023-09-23 21:15 154阅读 0赞

![compatible][]

[上一节][Link 1]我们在分析AQS源码时知道，AQS内部维护了一个同步状态位state和一个虚拟的同步队列来实现锁的争抢，其实除此以外还维护了一个**条件队列**。这个条件队列也不神秘，我们在开发中或多或少都使用过JUC的Conditon，用来做多个线程之间的等待唤醒，它的底层就是我们今天要说的AQS的条件队列。

我们先用一张图来看下AQS的内部结构：

![compatible 1][]

[上一节][Link 1]我们主要分析了AQS的CLH等待队列，今天我们着重来分析下**条件队列**的源码。

我们先看下AQS的ConditionObject类的结构：

publicclassConditionObjectimplementsCondition, java.io.Serializable {
        privatestaticfinallongserialVersionUID=1173984872572414699L;
        /** 
         * 条件队列的第一个节点
         */privatetransient Node firstWaiter;
        /**
         * 条件队列的最后一个节点
         */privatetransient Node lastWaiter;
        
        //重写Conditon接口的方法
        {
            //等待voidawait()throws InterruptedException;
    
            //.......//唤醒voidsignal();
    
            voidsignalAll();
    
        }
    }    
    复制代码

从类结构上不难看出，它也是使用Node作为队列的节点，前面我们分析[AQS源码][AQS]时已经看过了Node的组成结构，这里就不再看了。

我们以一个常见的面试题引出我们今天要讲的条件队列，就是3个线程顺序打印A-B-C, 我们先看下演示代码：

/**
     * @author qiuguan
     * @date 2022/12/10 22:48:23  星期六
     *
     * 控制线程打印A - B - C
     */publicclassLockConditionDemo {
    
        privatefinalLocklock=newReentrantLock();
        /**
         * lock.newCondition()就是创建了一个上面的 ConditionObject 对象
         * 一共创建了3个 ConditionObject 对象
         */privatefinalConditionc1= lock.newCondition();
        privatefinalConditionc2= lock.newCondition();
        privatefinalConditionc3= lock.newCondition();
    
        privateintflag=0;
    
        publicvoidprintA(){
            lock.lock();
            try {
                while (flag != 0) {
                    try {
                        c1.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
    
                System.out.println("--------------> A");
                flag++;
                c2.signal();
            } finally {
                lock.unlock();
            }
        }
    
        publicvoidprintB(){
            lock.lock();
            try {
                while (flag != 1) {
                    try {
                        c2.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
    
                System.out.println("--------------> B");
                flag++;
                c3.signal();
            } finally {
                lock.unlock();
            }
        }
    
        publicvoidprintC(){
            lock.lock();
            try {
                while (flag != 2) {
                    try {
                        c3.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
    
                System.out.println("--------------> C");
                flag = 0;
                c1.signal();
            } finally {
                lock.unlock();
            }
        }
    
        publicstaticvoidmain(String[] args) {
            LockConditionDemolc=newLockConditionDemo();
            newThread(lc::printC, "tc").start();
            newThread(lc::printB, "tb").start();
            newThread(lc::printA, "ta").start();
        }
    }
    复制代码

从代码中我们可以清晰的知道，**tc**,**tb**,**ta** 三个线程分别调用各自的方法，然后各自执行lock.lock()方法，其中只有一个线程能获得锁，获取不到锁的线程将会进入等待队列。我们建设**tc线程**率先获得锁，然后**tb线程**,**ta线程**先后进入等待队列。

![compatible 2][]

### tc线程优先获得锁 ###

publicvoidprintC(){
        //tc线程过来获取锁
        lock.lock();
        try {
            while (flag != 2) {
                try {
                    c3.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
    
            System.out.println("--------------> C");
            flag = 0;
            c1.signal();
        } finally {
            lock.unlock();
        }
    }
    复制代码

**tc线程**先执行，它会调用printC方法，进而执行lock.lock()方法，这个我们前面分析[AQS][AQS 1]的时候知道，**tc线程**进来会将同步状态位state从0修改为1，表示获取锁，然后继续往下执行，由于flag != 2条件成立，所以将会执行c3.await()的逻辑，到这里就要带领我们走进AQS的条件队列。接下来我们就看下c3.await() 具体都做了什么？我们直接看源码：

publicfinalvoidawait()throws InterruptedException {
        //如果线程是中断的，则抛出异常if (Thread.interrupted())
            thrownewInterruptedException();
        //1. 创建一个条件队列的节点    Nodenode= addConditionWaiter();
        //2. 释放锁，核心逻辑就是lock.unlock()方法的内容intsavedState= fullyRelease(node);
        intinterruptMode=0;
        //3. 判断是否在同步队列中，也就是前面说的CLH同步队列while (!isOnSyncQueue(node)) {
            //不在同步队列中，挂起
            LockSupport.park(this);
            //被唤醒后接着往下执行，然后checkInterruptWhileWaiting(node))方法会将当前//node加入到同步队列中（入队：入的是同步队列)if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        //acquireQueued(node, savedState)这个方法就比较熟悉了，前面分析AQS的时候着重说过//它就是加入同步队列后，等待被唤醒，然后去抢锁if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }
    复制代码

> 这个方法在AbstractQueuedSynchronizer（AQS）类中

我们先看下创建条件队列节点的方法：addConditionWaiter()

private Node addConditionWaiter() {
        Nodet= lastWaiter;
        // If lastWaiter is cancelled, clean out.if (t != null && t.waitStatus != Node.CONDITION) {
            unlinkCancelledWaiters();
            t = lastWaiter;
        }
        //创建一个Node节点，waitStatus 是 Node.CONDITIONNodenode=newNode(Thread.currentThread(), Node.CONDITION);
        //整个逻辑就是尾插法插入到条件队列的尾部if (t == null)
            firstWaiter = node;
        else
            t.nextWaiter = node;
        lastWaiter = node;
        
        return node;
    }
    复制代码

我们用一张图来看下入条件队列：

![compatible 3][]

我们用一张图来描述下**tc线程的await()方法**都干了什么

![compatible 4][]

从图中我们知道，**tc线程**加入条件队列后，会唤醒等待队列中等待的线程，然后将自己挂起。然后接下来我们看下唤醒**tb线程**(FIFO)后，**tb线程**都做了什么？

### 唤醒tb线程 ###

publicvoidprintB(){
        lock.lock();
        try {
            while (flag != 1) {
                try {
                
                    c2.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
    
            System.out.println("--------------> B");
            flag++;
            c3.signal();
        } finally {
            lock.unlock();
        }
    }
    复制代码

**tb线程**获得锁后，将从lock.lock()继续往下执行，然后来到c2的await()方法中，那这就比较熟悉了，因为它和**tc线程**是一样的。

1.  首先tb线程要加入条件队列（注意：这个条件队列和tc线程不是同一个，因为他们是分别newCondition获得的)

![compatible 5][]

1.  然后我们再看下**tb线程**的awiat()方法做了什么？

![compatible 6][]

从图中我们很清晰的知道，**tb线程**加入条件队列后，会唤醒等待队列中等待的线程，然后将自己挂起。然后接下来我们看下唤醒**ta线程**后，**ta线程**都做了什么？

### 唤醒ta线程 ###

publicvoidprintA(){
        lock.lock();
        try {
            while (flag != 0) {
                try {
                    c1.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
    
            System.out.println("--------------> A");
            flag++;
            c2.signal();
        } finally {
            lock.unlock();
        }
    }
    复制代码

**ta线程**获得锁后，将从lock.lock()继续往下执行，由于它不满足while条件，所以它先打印"--------------> A"，然后将执行**c2.signal()** 方法，那么接下来我们就看下这个方法都做了什么？

> 注意：是c2的signal()方法

publicfinalvoidsignal() {
        //判断当前的线程是否为持有锁的线程,tb线程if (!isHeldExclusively())
            thrownewIllegalMonitorStateException();
        //找到条件队列中的第一个first节点Nodefirst= firstWaiter;
        if (first != null)
            //唤醒
            doSignal(first);
    }
    复制代码

> 这里的firstWaiter节点是封装了tb线程的Node节点

![compatible 7][]

我们继续跟进去看下：

privatevoiddoSignal(Node first) {
        //整个do逻辑是将节点从头到尾一个一个的从条件队列中移除do {
            //这里的逻辑就是看下条件队列是不是只有一个节点if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            //if条件里已经完成了赋值操作，将下一个节点推到了首节点    
            first.nextWaiter = null;
                 //重点是transferForSignal(first) 方法
        } while (!transferForSignal(first) &&
                 (first = firstWaiter) != null);
    }
    复制代码

我们接着看下transferForSignal(first) 方法：

> 这个方法的逻辑就是将条件队列的节点转移到同步队列中

finalbooleantransferForSignal(Node node) {
        /*
         * 如果CAS修改失败，说明这个节点已经被取消了
         * 一旦被取消了，就重新执行上面的do-while循环
         */if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            returnfalse;
    
        /*
         * CAS状态修改成功，将条件状态修改为了初始值0
         *///入队：入的是"等待队列"的队 ,并返回前驱节点Nodep= enq(node);
        intws= p.waitStatus; //查看前驱节点的状态，默认值0//CAS 将前驱节点的状态修改为-1（具有唤醒下一个节点的责任)，如果成功，返回true,// 如果失败，返回false,取反后则将进入if判断的内部逻辑，也就是唤醒线程if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        returntrue;
    }
    复制代码

这个方法执行完

1.  条件队列变成了这样：

![compatible 8][]

1.  同步队列就变成了这样：

![compatible 9][]

然后退出c2.signal()法方法，然后执行printA()方法的lock.unlock();，这个方法就比较熟悉了，前面讲AQS基础的时候有说过，这里我们在简单看下：

publicfinalbooleanrelease(int arg) {
        //ta线程释放锁if (tryRelease(arg)) {
            Nodeh= head;
            if (h != null && h.waitStatus != 0)
                //唤醒后继节点
                unparkSuccessor(h);
            returntrue;
        }
        returnfalse;
    }
    复制代码

> 在这个方法中，将会调用LockSupport.unpark(s.thread);唤醒 **tb线程**

![compatible 10][]

我们看下之前**tb线程**park的地方:

![compatible 11][]

然后tb线程将会继续往下执行，由于它已经在同步队列中了，所以将会退出while循环，然后执行acquireQueued方法，这个方法我们也比较熟悉了，就是尝试获取锁，如果获取不到就继续在同步队列中park。

由于tb线程此时可以获取锁，所以它将不会再继续park,继续往下执行业务代码，将会打印"--------------> B"

然后tb线程继续执行c3.signal()方法，继续去唤醒tc线程，然后打印"--------------> C"

好了，关于AQS的条件队列就分析到这里吧，可能阅读起来比较复杂，限于作者水平有限，不能更好的表达，敬请谅解。

***限于作者水平，文中难免有错误之处，欢迎指正，勿喷，感谢感谢***

作者：秋官

原文链接：[https://juejin.cn/post/7182869415321403449][https_juejin.cn_post_7182869415321403449]

[compatible]: https://ksc-bj.ag.wps.cn/api/object/2_9915dbbda23e4fdea00a997648bcfd44/compatible
[Link 1]: https://juejin.cn/post/7174802301100490812
[compatible 1]: https://ksc-bj.ag.wps.cn/api/object/2_33bd51974cb942d38bf103de536fbd18/compatible
[AQS]: https://juejin.cn/post/7174802301100490812#heading-0
[compatible 2]: https://ksc-bj.ag.wps.cn/api/object/2_c94f652d160e495f819a634e1cb23074/compatible
[AQS 1]: https://juejin.cn/post/7174802301100490812#heading-7
[compatible 3]: https://ksc-bj.ag.wps.cn/api/object/2_7c1c128d674243a79143f8fffbbf2ad5/compatible
[compatible 4]: https://ksc-bj.ag.wps.cn/api/object/2_016714a56e8d428fbe9f3addf0c1cca3/compatible
[compatible 5]: https://ksc-bj.ag.wps.cn/api/object/2_a6edd920b8c14bec90a1b7b01cf448c0/compatible
[compatible 6]: https://ksc-bj.ag.wps.cn/api/object/2_70c07713c43440febb95cf7c6965cb55/compatible
[compatible 7]: https://ksc-bj.ag.wps.cn/api/object/2_8c017ac07393477c9740c404339889d6/compatible
[compatible 8]: https://ksc-bj.ag.wps.cn/api/object/2_28a64f7980794ac5af1819db8c023eb2/compatible
[compatible 9]: https://ksc-bj.ag.wps.cn/api/object/2_bb70505c6be9457eb9d3fa2616729690/compatible
[compatible 10]: https://ksc-bj.ag.wps.cn/api/object/2_37e7cbe8c8c943629ca50ffac4f413db/compatible
[compatible 11]: https://ksc-bj.ag.wps.cn/api/object/2_56071aa80b4940e9be5e95518fb6a8b1/compatible
[https_juejin.cn_post_7182869415321403449]: https://juejin.cn/post/7182869415321403449