Java并发编程(十) : JUC 之 AQS原理、ReentrantLock原理

心已赠人 2023-01-02 05:28 168阅读 0赞

# 一、AQS 原理 #

> [什么是AQS及其原理][AQS]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTg5OTgw_size_16_color_FFFFFF_t_70] **`AQS核心思想是，如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并且将共享资源设置为锁定状态`。如果被请求的共享资源被占用，那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制，这个机制AQS是用CLH队列锁实现的，即将暂时获取不到锁的线程加入到队列中。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTg5OTgw_size_16_color_FFFFFF_t_70 1]

*  特点：
    
     *  用 `state` 属性来表示`资源的状态`（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁
        
         *  **`getState - 获取 state 状态`**
         *  **`setState - 设置 state 状态`**
         *  **`compareAndSetState - cas 机制设置 state 状态`**
         *  **`独占模式是只有一个线程能够访问资源，而共享模式可以允许多个线程访问资源`**
     *  **提供了基于 FIFO 的等待队列，类似于 `Monitor` 的 `EntryList`**
     *  **条件变量来实现等待、唤醒机制，支持多个条件变量，类似于 `Monitor` 的 `WaitSet`**

子类主要实现这样一些方法（默认抛出 UnsupportedOperationException）

*  **tryAcquire**
 *  **tryRelease**
 *  **tryAcquireShared**
 *  **tryReleaseShared**
 *  **isHeldExclusively**

//获取锁的姿势
    // 如果获取锁失败
    if (!tryAcquire(arg)) { 
     // 入队, 可以选择阻塞当前线程 park unpark
    }
    
    //释放锁的姿势
    // 如果释放锁成功
    if (tryRelease(arg)) { 
     // 让阻塞线程恢复运行
    }

*  下面实现一个不可重入的阻塞式锁：使用`AbstractQueuedSynchronizer`自定义一个同步器来实现自定义锁！

/** * Description: * * @author guizy * @date 2021/1/4 23:10 */
    @Slf4j(topic = "guizy.TestAQS")
    @SuppressWarnings("all")
    public class TestAqs { 
        public static void main(String[] args) { 
            MyLock lock = new MyLock();
            new Thread(() -> { 
                lock.lock();
                log.debug("locking...");
                // 不可重入锁, 同一线程在锁释放前, 只能加一次锁
    // lock.lock();
    // log.debug("locking...");
                try { 
                    log.debug("locking...");
                    Sleeper.sleep(1);
                } finally { 
                    log.debug("unlocking...");
                    lock.unlock();
                }
            }, "t1").start();
    
            new Thread(() -> { 
                lock.lock();
                try { 
                    log.debug("locking...");
                } finally { 
                    log.debug("unlocking...");
                    lock.unlock();
                }
            }, "t2").start();
        }
    }
    
    // 自定义锁（不可重入锁）
    class MyLock implements Lock { 
    
        // 独占锁 同步器类
        class MySync extends AbstractQueuedSynchronizer { 
            @Override
            protected boolean tryAcquire(int arg) { 
                // 确保原子性
                if (compareAndSetState(0, 1)) { 
                    // 加上了锁，并设置 owner 为当前线程
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
    
            @Override
            protected boolean tryRelease(int arg) { 
                // 这里不需要确定原子性, 因为是持锁者进行释放
                // 在setState(0)上面设置Owner为null, 防止指令重排序带来的问题
                setExclusiveOwnerThread(null);
                setState(0); // state是volatile修饰的, 在setState(0)前面的属性修改, 对于其他线程也是可见的, 具体见volatile原理(写屏障)
                return true;
            }
    
            @Override // 是否持有独占锁
            protected boolean isHeldExclusively() { 
                return getState() == 1;
            }
    
            public Condition newCondition() { 
                return new ConditionObject();
            }
        }
    
        private MySync sync = new MySync();
    
        @Override // 加锁（不成功会进入等待队列）
        public void lock() { 
            sync.acquire(1);
        }
    
        @Override // 加锁，可打断
        public void lockInterruptibly() throws InterruptedException { 
            sync.acquireInterruptibly(1);
        }
    
        @Override // 尝试加锁（一次）
        public boolean tryLock() { 
            return sync.tryAcquire(1);
        }
    
        @Override // 尝试加锁，带超时
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { 
            return sync.tryAcquireNanos(1, unit.toNanos(time));
        }
    
        @Override // 解锁
        public void unlock() { 
            sync.release(1);
        }
    
        @Override // 创建条件变量
        public Condition newCondition() { 
            return sync.newCondition();
        }
    }

--------------------

# 二、 ReentrantLock 原理 #

*  ReentrantLock提供了两个同步器，实现`公平锁`和`非公平锁`，默认是非公平锁！

![1595043973690][]

#### 1. 非公平锁实现原理 ####

##### 图解流程 #####

*  `加锁, 解锁`流程，先从构造器开始看，默认为非公平锁实现

public ReentrantLock() { 
     	sync = new NonfairSync();
    }

**NonfairSync** extends Sync extends AbstractQueuedSynchronizer

*  没有竞争时
    
     *  Thread-0成为锁的持有者  
        ![1595045253140][]
 *  第一个竞争出现时，查看源码的`NonfairSync`的`lock`方法

![1595045270516][]

abstract void lock();
    
    // 非公平锁的lock
    final void lock() { 
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }
    
    public final void acquire(int arg) { 
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

Thread-1 执行了

1.  **lock方法中CAS 尝试将 state 由 0 改为 1，结果失败 (因为此时CAS操作, 已经state已经为1了)**
2.  lock方法中进一步调用`acquire`方法，进入 `tryAcquire` 逻辑，这里我们认为这时 state 已经是1，结果仍然失败
3.  接下来进入 acquire方法的`addWaiter` 逻辑，构造 `Node 队列 (双向链表实现)`
    
    1.  下图中`黄色三角`表示该 Node 的`waitStatus`状态，其中 0 为默认正常状态
    2.  Node 的创建是懒惰的
    3.  **其中第一个 Node 称为 Dummy（哑元）或哨兵，用来占位，并不关联线程**

![1595045451872][]

当前线程进入 acquire方法的 `acquireQueued` 逻辑

1.  acquireQueued 会在一个死循环中不断尝试获得锁，失败后进入 `park` 阻塞
2.  如果自己是紧邻着 head（排第二位），那么再次 tryAcquire 尝试获取锁，我们这里设置这时 state 仍为 1，失败
3.  进入 `shouldParkAfterFailedAcquire` 逻辑，将`前驱 node`，即 head 的 waitStatus 改为 `-1`，这次返回 false
    
    ![1595046768331][]
4.  shouldParkAfterFailedAcquire 执行完毕回到 `acquireQueued` ，再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
5.  当再次进入 shouldParkAfterFailedAcquire 时，这时因为其前驱 node 的 waitStatus 已经是 -1，这次返回 true
6.  进入`parkAndCheckInterrupt`， Thread-1 `park`（灰色表示已经阻塞）  
    ![在这里插入图片描述][20210105234001801.png]

![1595046786213][]

*  再次有多个线程经历上述过程竞争失败，变成这个样子

![1595046803755][]  
Thread-0 调用`unlock方法`里的`release方法`释放锁，进入`tryRelease(使用ctrl+alt+b查看tryRelease方法的具体`ReentrantLock`实现)`流程，如果成功，**设置 exclusiveOwnerThread 为 null，state = 0**

![1595046828330][]

unlock方法里的release方法方法中，如果当前队列不为 null，并且 head 的 waitStatus = -1，进入 unparkSuccessor 流程： **`unparkSuccessor中会找到队列中离 head 最近的一个 Node（没取消的），unpark 唤醒Thread-1 恢复其运行`**，本例中即为 Thread-1 回到 Thread-1 阻塞的位置继续执行, 会继续执行 acquireQueued 流程  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTg5OTgw_size_16_color_FFFFFF_t_70 2]

![1595046840247][]

**如果加锁成功（没有竞争）**，会设置 （acquireQueued 方法中）

1.  exclusiveOwnerThread 为 Thread-1，state = 1
2.  head 指向刚刚 Thread-1 所在的 Node，该 Node 清空 Thread
3.  原本的 head 因为从链表断开，而可被垃圾回收

**如果这时候有其它线程来竞争（非公平的体现）**，例如这时有 Thread-4 来了

![1595046854757][]

如果不巧又被 Thread-4 占了先

1.  Thread-4 被设置为 exclusiveOwnerThread，state = 1
2.  `Thread-1 再次进入 acquireQueued 流程，获取锁失败，重新进入 park 阻塞`

##### `加锁`源码 #####

// Sync 继承自 AQS
    static final class NonfairSync extends Sync { 
        private static final long serialVersionUID = 7316153563782823691L;
    
         // 加锁实现
        final void lock() { 
            // 首先用 cas 尝试（仅尝试一次）将 state 从 0 改为 1, 如果成功表示获得了独占锁
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                // 如果尝试失败，进入 ㈠
                acquire(1);
        }
    
        // ㈠ AQS 继承过来的方法, 方便阅读, 放在此处
        public final void acquire(int arg) { 
            // ㈡ tryAcquire
            if (
                    !tryAcquire(arg) &&
                	// 当 tryAcquire 返回为 false 时, 先调用 addWaiter ㈣, 接着 acquireQueued ㈤
                     acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
            ) { 
                selfInterrupt();
            }
        }
    
        // ㈡ 进入 ㈢
        protected final boolean tryAcquire(int acquires) { 
            return nonfairTryAcquire(acquires);
        }
    
        // ㈢ Sync 继承过来的方法, 方便阅读, 放在此处
        final boolean nonfairTryAcquire(int acquires) { 
            final Thread current = Thread.currentThread();
            int c = getState();
            // 如果还没有获得锁
            if (c == 0) { 
                // 尝试用 cas 获得, 这里体现了非公平性: 不去检查 AQS 队列
                if (compareAndSetState(0, acquires)) { 
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            // 如果已经获得了锁, 线程还是当前线程, 表示发生了锁重入
            else if (current == getExclusiveOwnerThread()) { 
                // state++
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            // 获取失败, 回到调用处
            return false;
        }
    
        // ㈣ AQS 继承过来的方法, 方便阅读, 放在此处
        private Node addWaiter(Node mode) { 
    // 将当前线程关联到一个 Node 对象上, 模式为独占模式，新建的Node的waitstatus默认为0，因为waitstatus是成员变量，默认被初始化为0
            Node node = new Node(Thread.currentThread(), mode);
            // 如果 tail 不为 null, cas 尝试将 Node 对象加入 AQS 队列尾部
            Node pred = tail;
            if (pred != null) { 
                node.prev = pred;
                if (compareAndSetTail(pred, node)) { 
                    // 双向链表
                    pred.next = node;
                    return node;
                }
            }
            //如果tail为null，尝试将 Node 加入 AQS, 进入 ㈥
            enq(node);
            return node;
        }
    
        // ㈥ AQS 继承过来的方法, 方便阅读, 放在此处
        private Node enq(final Node node) { 
            for (;;) { 
                Node t = tail;
                if (t == null) { 
                    // 还没有, 设置 head 为哨兵节点（不对应线程，状态为 0）
                    if (compareAndSetHead(new Node())) { 
                        tail = head;
                    }
                } else { 
                    // cas 尝试将 Node 对象加入 AQS 队列尾部
                    node.prev = t;
                    if (compareAndSetTail(t, node)) { 
                        t.next = node;
                        return t;
                    }
                }
            }
        }
    
        // ㈤ AQS 继承过来的方法, 方便阅读, 放在此处
        final boolean acquireQueued(final Node node, int arg) { 
            boolean failed = true;
            try { 
                boolean interrupted = false;
                for (;;) { 
                    final Node p = node.predecessor();
                    // 上一个节点是 head, 表示轮到自己（当前线程对应的 node）了, 尝试获取
                    if (p == head && tryAcquire(arg)) { 
                        // 获取成功, 设置自己（当前线程对应的 node）为 head
                        setHead(node);
                        // 上一个节点 help GC
                        p.next = null;
                        failed = false;
                        // 返回中断标记 false
                        return interrupted;
                    }
                    if (
                        // 判断是否应当 park, 进入 ㈦
                        shouldParkAfterFailedAcquire(p, node) &&
                        // park 等待, 此时 Node 的状态被置为 Node.SIGNAL ㈧
                        parkAndCheckInterrupt()
                    ) { 
                        interrupted = true;
                    }
                }
            } finally { 
                if (failed)
                    cancelAcquire(node);
            }
        }
    
        // ㈦ AQS 继承过来的方法, 方便阅读, 放在此处
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
            // 获取上一个节点的状态
            int ws = pred.waitStatus;
            if (ws == Node.SIGNAL) { 
                // 上一个节点都在阻塞, 那么自己也阻塞好了
                return true;
            }
            // > 0 表示取消状态
            if (ws > 0) { 
                // 上一个节点取消, 那么重构删除前面所有取消的节点, 返回到外层循环重试
                do { 
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                pred.next = node;
            } else { 
                // 这次还没有阻塞
                // 但下次如果重试不成功, 则需要阻塞，这时需要设置上一个节点状态为 Node.SIGNAL
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }
    
        // ㈧ 阻塞当前线程
        private final boolean parkAndCheckInterrupt() { 
            LockSupport.park(this);
            return Thread.interrupted();
        }
    }

##### `解锁`源码 #####

// Sync 继承自 AQS
    static final class NonfairSync extends Sync { 
        // 解锁实现
        public void unlock() { 
            sync.release(1);
        }
    
        // AQS 继承过来的方法, 方便阅读, 放在此处
        public final boolean release(int arg) { 
            // 尝试释放锁, 进入 ㈠
            if (tryRelease(arg)) { 
                // 队列头节点 unpark
                Node h = head;
                if (
                    // 队列不为 null
                    h != null &&
                    // waitStatus == Node.SIGNAL 才需要 unpark
                    h.waitStatus != 0
                ) { 
                    // unpark AQS 中等待的线程, 进入 ㈡
                    unparkSuccessor(h);
                }
                return true;
            }
            return false;
        }
    
        // ㈠ Sync 继承过来的方法, 方便阅读, 放在此处
        protected final boolean tryRelease(int releases) { 
            // state--
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            // 支持锁重入, 只有 state 减为 0, 才释放成功
            if (c == 0) { 
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }
    
        // ㈡ AQS 继承过来的方法, 方便阅读, 放在此处
        private void unparkSuccessor(Node node) { 
            // 如果状态为 Node.SIGNAL 尝试重置状态为 0, 如果线程获取到了锁那么后来头结点会被抛弃掉
            // 不成功也可以
            int ws = node.waitStatus;
            if (ws < 0) { 
                compareAndSetWaitStatus(node, ws, 0);
            }
            // 找到需要 unpark 的节点, 但本节点从 AQS 队列中脱离, 是由唤醒节点完成的
            Node s = node.next;
            // 不考虑已取消的节点, 从 AQS 队列从后至前找到队列最前面需要 unpark 的节点
            if (s == null || s.waitStatus > 0) { 
                s = null;
                for (Node t = tail; t != null && t != node; t = t.prev)
                    if (t.waitStatus <= 0)
                        s = t;
            }
            if (s != null)
                LockSupport.unpark(s.thread);
        }
    }

#### 2. 可重入原理 ####

>  *  同一个线程, 锁重入, 会对`state`进行自增. 释放锁的时候, state进行自减; 当state自减为0的时候. 此时线程才会将`锁`释放成功, 才会进一步去唤醒`其他线程`来竞争锁

static final class NonfairSync extends Sync { 
        // ...
    
        // Sync 继承过来的方法, 方便阅读, 放在此处
        final boolean nonfairTryAcquire(int acquires) { 
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) { 
                if (compareAndSetState(0, acquires)) { 
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            // 如果已经获得了锁, 线程还是当前线程, 表示发生了锁重入
            else if (current == getExclusiveOwnerThread()) { 
                // state++
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    
        // Sync 继承过来的方法, 方便阅读, 放在此处
        protected final boolean tryRelease(int releases) { 
            // state--
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            // 支持锁重入, 只有 state 减为 0, 才释放成功
            if (c == 0) { 
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }
    }

#### 3. 可打断原理 ####

*  不可打断模式：在此模式下，**即使它被打断，仍会驻留在 AQS 队列中，一直要等到获得锁后方能得知自己被打断了**

// Sync 继承自 AQS
    static final class NonfairSync extends Sync { 
        // ...
    
        private final boolean parkAndCheckInterrupt() { 
            // 如果打断标记已经是 true, 则 park 会失效
            // 被park阻塞的线程, 可以被别的线程调用它的interrupt方法打断该park阻塞
            LockSupport.park(this);
            // interrupted 会清除打断标记; 下次park仍然可以阻塞
            return Thread.interrupted();
        }
    
        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;
                        failed = false;
                        // 还是需要获得锁后, 才能返回打断状态
                        return interrupted;
                    }
                    if (
                            shouldParkAfterFailedAcquire(p, node) &&
                                    parkAndCheckInterrupt()
                    ) { 
                        // 如果是因为 interrupt 被唤醒, 返回打断状态为 true
                        interrupted = true;
                    }
                }
            } finally { 
                if (failed)
                    cancelAcquire(node);
            }
        }
    
        public final void acquire(int arg) { 
            if (
                    !tryAcquire(arg) &&
                            acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
            ) { 
                // 如果打断状态为 true
                selfInterrupt();
            }
        }
    
        static void selfInterrupt() { 
            // 重新产生一次中断，这时候线程是如果正常运行的状态，那么不是出于sleep等状态，interrupt方法就不会报错
            Thread.currentThread().interrupt();
        }
    }
    }

可打断模式

static final class NonfairSync extends Sync { 
        public final void acquireInterruptibly(int arg) throws InterruptedException { 
            if (Thread.interrupted())
                throw new InterruptedException();
            // 如果没有获得到锁, 进入 ㈠
            if (!tryAcquire(arg))
                doAcquireInterruptibly(arg);
        }
    
        // ㈠ 可打断的获取锁流程
        private void doAcquireInterruptibly(int arg) throws InterruptedException { 
            final Node node = addWaiter(Node.EXCLUSIVE);
            boolean failed = true;
            try { 
                for (;;) { 
                    final Node p = node.predecessor();
                    if (p == head && tryAcquire(arg)) { 
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                            parkAndCheckInterrupt()) { 
                        // 在 park 过程中如果被 interrupt 会进入此
                        // 这时候抛出异常, 而不会再次进入 for (;;)
                        throw new InterruptedException();
                    }
                }
            } finally { 
                if (failed)
                    cancelAcquire(node);
            }
        }
    }

#### 4. 公平锁实现原理 ####

> 看AQS队列中, 自己(线程) 有没有前驱节点(线程), (该节点不是占位的哨兵节点); 如果有就不去竞争锁. 如果没有, 才会去CAS操作
> 
>  *  if (!hasQueuedPredecessors() &&  
>     compareAndSetState(0, acquires)) \{  
>     setExclusiveOwnerThread(current);  
>     return true;  
>     \}

static final class FairSync extends Sync { 
        private static final long serialVersionUID = -3000897897090466540L;
        final void lock() { 
            acquire(1);
        }
    
        // AQS 继承过来的方法, 方便阅读, 放在此处
        public final void acquire(int arg) { 
            if (
                    !tryAcquire(arg) &&
                            acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
            ) { 
                selfInterrupt();
            }
        }
        // 与非公平锁主要区别在于 tryAcquire 方法的实现
        protected final boolean tryAcquire(int acquires) { 
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) { 
                // 先检查 AQS 队列中是否有前驱节点, 没有才去竞争
                if (!hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) { 
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) { 
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    
        // ㈠ AQS 继承过来的方法, 方便阅读, 放在此处
        public final boolean hasQueuedPredecessors() { 
            Node t = tail;
            Node h = head;
            Node s;
            // h != t 时表示队列中有 Node
            return h != t &&
                    (
                            // (s = h.next) == null 表示队列中还有没有老二
                            (s = h.next) == null || // 或者队列中老二线程不是此线程
                                    s.thread != Thread.currentThread()
                    );
        }
    }

#### 5. 条件变量实现原理 ####

##### 图解流程 #####

每个条件变量其实就对应着一个`等待队列`，其实现类是 `ConditionObject`

*  `await 流程` 开始 Thread-0 持有锁，conditionObject对象调用 `await`，进入 ConditionObject 的 `addConditionWaiter` 流程 创建新的 Node 状态为 `-2`（Node.CONDITION），关联 Thread-0，加入等待队列尾部

![1595079373121][]

接下来进入 AQS 的 `fullyRelease` 流程，释放同步器上的所有的锁 (因为可能线程发生可重入, 锁有很多层)

![1595079397323][]

`unparkSuccessor(h);` —> unpark唤醒 AQS 队列中的下一个节点，竞争锁，假设没有其他竞争线程，那么 `Thread-1` 竞争成功

![1595079457815][]  
LockSupport.park(this); —> park 阻塞 Thread-0

![1595079481112][]

*  `signal 流程`
    
     *  假设 Thread-1 要来唤醒 Thread-0

// 如果没有持有锁，会抛出异常 --> 这里表示Thread-1要持有锁, 
    	//才可以去条件变量中去唤醒等待的线程
            if (!isHeldExclusively())

![1595079499939][]

进入 ConditionObject 的 `doSignal` 流程，取得等待队列中第一个 Node，即 Thread-0 所在 Node

private void doSignal(Node first) { 
        do { 
        	// 去firstWaiter条件变量中将等待的线程拿出来.
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            first.nextWaiter = null;
            // 转移到AQS的队列中, 等待竞争锁
        } while (!transferForSignal(first) &&
                 (first = firstWaiter) != null);
    }

![1595079518690][]

执行 `transferForSignal` 流程，**将该 Node 加入 AQS 队列尾部，将 Thread-0 的 waitStatus 改为 0，Thread-3 的waitStatus 改为 -1** , 改为`-1`就有责任去唤醒自己的后继节点

![1595079772187][]

Thread-1 释放锁，进入 unlock 流程，略

##### 源码分析 #####

public class ConditionObject implements Condition, java.io.Serializable { 
        private static final long serialVersionUID = 1173984872572414699L;
    
        // 第一个等待节点
        private transient Node firstWaiter;
    
        // 最后一个等待节点
        private transient Node lastWaiter;
        public ConditionObject() {  }
        // ㈠ 添加一个 Node 至等待队列
        private Node addConditionWaiter() { 
            Node t = lastWaiter;
            // 所有已取消的 Node 从队列链表删除, 见 ㈡
            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;
        }
        // 唤醒 - 将没取消的第一个节点转移至 AQS 队列
        private void doSignal(Node first) { 
            do { 
                // 已经是尾节点了
                if ( (firstWaiter = first.nextWaiter) == null) { 
                    lastWaiter = null;
                }
                first.nextWaiter = null;
            } while (
                // 将等待队列中的 Node 转移至 AQS 队列, 不成功且还有节点则继续循环 ㈢
                    !transferForSignal(first) &&
                            // 队列还有节点
                            (first = firstWaiter) != null
            );
        }
    
        // 外部类方法, 方便阅读, 放在此处
        // ㈢ 如果节点状态是取消, 返回 false 表示转移失败, 否则转移成功
        final boolean transferForSignal(Node node) { 
            // 设置当前node状态为0（因为处在队列末尾），如果状态已经不是 Node.CONDITION, 说明被取消了
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                return false;
            // 加入 AQS 队列尾部
            Node p = enq(node);
            int ws = p.waitStatus;
            if (
                // 插入节点的上一个节点被取消
                    ws > 0 ||
                            // 插入节点的上一个节点不能设置状态为 Node.SIGNAL
                            !compareAndSetWaitStatus(p, ws, Node.SIGNAL)
            ) { 
                // unpark 取消阻塞, 让线程重新同步状态
                LockSupport.unpark(node.thread);
            }
            return true;
        }
    // 全部唤醒 - 等待队列的所有节点转移至 AQS 队列
    private void doSignalAll(Node first) { 
        lastWaiter = firstWaiter = null;
        do { 
            Node next = first.nextWaiter;
            first.nextWaiter = null;
            transferForSignal(first);
            first = next;
        } while (first != null);
    }
    
        // ㈡
        private void unlinkCancelledWaiters() { 
            // ...
        }
        // 唤醒 - 必须持有锁才能唤醒, 因此 doSignal 内无需考虑加锁
        public final void signal() { 
            // 如果没有持有锁，会抛出异常
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignal(first);
        }
        // 全部唤醒 - 必须持有锁才能唤醒, 因此 doSignalAll 内无需考虑加锁
        public final void signalAll() { 
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignalAll(first);
        }
        // 不可打断等待 - 直到被唤醒
        public final void awaitUninterruptibly() { 
            // 添加一个 Node 至等待队列, 见 ㈠
            Node node = addConditionWaiter();
            // 释放节点持有的锁, 见 ㈣
            int savedState = fullyRelease(node);
            boolean interrupted = false;
            // 如果该节点还没有转移至 AQS 队列, 阻塞
            while (!isOnSyncQueue(node)) { 
                // park 阻塞
                LockSupport.park(this);
                // 如果被打断, 仅设置打断状态
                if (Thread.interrupted())
                    interrupted = true;
            }
            // 唤醒后, 尝试竞争锁, 如果失败进入 AQS 队列
            if (acquireQueued(node, savedState) || interrupted)
                selfInterrupt();
        }
        // 外部类方法, 方便阅读, 放在此处
        // ㈣ 因为某线程可能重入，需要将 state 全部释放，获取state，然后把它全部减掉，以全部释放
        final int fullyRelease(Node node) { 
            boolean failed = true;
            try { 
                int savedState = getState();
                // 唤醒等待队列队列中的下一个节点
                if (release(savedState)) { 
                    failed = false;
                    return savedState;
                } else { 
                    throw new IllegalMonitorStateException();
                }
            } finally { 
                if (failed)
                    node.waitStatus = Node.CANCELLED;
            }
        }
        // 打断模式 - 在退出等待时重新设置打断状态
        private static final int REINTERRUPT = 1;
        // 打断模式 - 在退出等待时抛出异常
        private static final int THROW_IE = -1;
        // 判断打断模式
        private int checkInterruptWhileWaiting(Node node) { 
            return Thread.interrupted() ?
                    (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
                    0;
        }
        // ㈤ 应用打断模式
        private void reportInterruptAfterWait(int interruptMode)
                throws InterruptedException { 
            if (interruptMode == THROW_IE)
                throw new InterruptedException();
            else if (interruptMode == REINTERRUPT)
                selfInterrupt();
        }
        // 等待 - 直到被唤醒或打断
        public final void await() throws InterruptedException { 
            if (Thread.interrupted()) { 
                throw new InterruptedException();
            }
            // 添加一个 Node 至等待队列, 见 ㈠
            Node node = addConditionWaiter();
            // 释放节点持有的锁
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            // 如果该节点还没有转移至 AQS 队列, 阻塞
            while (!isOnSyncQueue(node)) { 
                // park 阻塞 
                LockSupport.park(this);
                // 如果被打断, 退出等待队列
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            // 退出等待队列后, 还需要获得 AQS 队列的锁
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            // 所有已取消的 Node 从队列链表删除, 见 ㈡
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            // 应用打断模式, 见 ㈤
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }
        // 等待 - 直到被唤醒或打断或超时
        public final long awaitNanos(long nanosTimeout) throws InterruptedException { 
            if (Thread.interrupted()) { 
                throw new InterruptedException();
            }
            // 添加一个 Node 至等待队列, 见 ㈠
            Node node = addConditionWaiter();
            // 释放节点持有的锁
            int savedState = fullyRelease(node);
            // 获得最后期限
            final long deadline = System.nanoTime() + nanosTimeout;
            int interruptMode = 0;
            // 如果该节点还没有转移至 AQS 队列, 阻塞
            while (!isOnSyncQueue(node)) { 
                // 已超时, 退出等待队列
                if (nanosTimeout <= 0L) { 
                    transferAfterCancelledWait(node);
                    break;
                }
                // park 阻塞一定时间, spinForTimeoutThreshold 为 1000 ns
                if (nanosTimeout >= spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                // 如果被打断, 退出等待队列
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                nanosTimeout = deadline - System.nanoTime();
            }
            // 退出等待队列后, 还需要获得 AQS 队列的锁
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            // 所有已取消的 Node 从队列链表删除, 见 ㈡
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            // 应用打断模式, 见 ㈤
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return deadline - System.nanoTime();
        }
        // 等待 - 直到被唤醒或打断或超时, 逻辑类似于 awaitNanos
        public final boolean awaitUntil(Date deadline) throws InterruptedException { 
            // ...
        }
        // 等待 - 直到被唤醒或打断或超时, 逻辑类似于 awaitNanos
        public final boolean await(long time, TimeUnit unit) throws InterruptedException { 
            // ...
        }
        // 工具方法 省略 ...
    }

[AQS]: https://blog.csdn.net/striveb/article/details/86761900
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTg5OTgw_size_16_color_FFFFFF_t_70]: /images/20221120/fec30c8ae12840e0b139fc2f32e953f5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTg5OTgw_size_16_color_FFFFFF_t_70 1]: /images/20221120/98d7950f200242479dff40bf461ee1db.png
[1595043973690]: /images/20221120/f2d7dc3b0c674016a2529f4f681d2a9e.png
[1595045253140]: /images/20221120/01bf1707932c45ac810108a23793ea99.png
[1595045270516]: /images/20221120/c1ac6052446e4e1c972a80e2198bbe22.png
[1595045451872]: /images/20221120/968225c413f14e37a6fb19295d8206c3.png
[1595046768331]: /images/20221120/ed2081fa081f4384a28ada978139471b.png
[20210105234001801.png]: /images/20221120/eb0604cc631b4626b41f0c603520b8d6.png
[1595046786213]: /images/20221120/831e9b36edfe4b5ea024970f9d80ec23.png
[1595046803755]: /images/20221120/41483b8ec9af45a4a768e3fa98b2ad08.png
[1595046828330]: /images/20221120/700e80ece3654007a4147133c9d501f3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTg5OTgw_size_16_color_FFFFFF_t_70 2]: /images/20221120/e34bdba129d24295ac4c00ed7c438fc3.png
[1595046840247]: /images/20221120/de3048eab826418e9242e8bd8a1a09a4.png
[1595046854757]: /images/20221120/10cfc01731b74972a27dc77e5f19eb54.png
[1595079373121]: /images/20221120/ebaf32702397497c877f1243b30575a4.png
[1595079397323]: /images/20221120/e9bd660c72904eec8210bad1261ce005.png
[1595079457815]: /images/20221120/d1fbd406c6d14a9f96fc68306c157cf3.png
[1595079481112]: /images/20221120/ecaad0b2f5004f3c89ddfb181ffb64ec.png
[1595079499939]: /images/20221120/9b80507387754e0fab34892b3cae1640.png
[1595079518690]: /images/20221120/91b79da7b60944428e90a8e1c236fe5c.png
[1595079772187]: /images/20221120/08f02a561b41404485a770019636a2df.png