AQS源码探究_06 Conditon条件队列(await方法、线程入队与挂起、signal方法)

叁歲伎倆 2022-10-21 11:48 9阅读 0赞

# AQS源码探究\_06 Conditon条件队列(await方法、线程入队与挂起、signal方法) #

**提示**：读源码的时候尽量自己点开源码跟着博客注释一起看，不然容易迷路~

## 1、条件队列流程图 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzU5MTk4MA_size_16_color_FFFFFF_t_70]

## 2、Condition接口 ##

public interface Condition { 
    	// 线程等待，可抛出中断异常(可以响应中断)
    	void await() throws InterruptedException;
    	
    	// 线程等待，但是不可响应中断
    	void awaitUninterruptibly();
    	
    	// 线程超时等待
    	boolean await(long time, TimeUnit unit) throws InterruptedException;
    	
    	// 线程等待，直到收到被唤醒、或收到中断信号、或到了指定的截止时间日期自动唤醒
    	boolean awaitUntil(Date deadline) throws InterruptedException;
    	
    	// 线程唤醒
    	void signal();
    	
    	// 唤醒所有线程
    	void signalAll();
    }

## 3、AQS中实现Condition接口的内部类ConditionObject ##

// 位于AQS中的内部类：
    public class ConditionObject implements Condition, java.io.Serializable { 
        private static final long serialVersionUID = 1173984872572414699L;
        // 队列的第一个等待者节点：头等待者
        private transient Node firstWaiter;
        // 队列的最后一个等待者节点：尾等待者
        private transient Node lastWaiter;
        
        /** * 无参构造函数： * 在ReentrantLock中的newCondition()方法内，借助ReentrantLock的静态内部类Sync，去调用 * newCondition方法实例化ConditionObject */
        public ConditionObject() {  }
    
        ...
    }

接下来分析一下ConditionObject的内部方法：

### 3.1 await线程等待方法 ###

// 线程等待，可抛出中断异常(可以响应中断)
    public final void await() throws InterruptedException { 
        // 判断当前线程是否是中断状态：
        if (Thread.interrupted())
            // 如果是中断状态，则直接抛出中断异常
            throw new InterruptedException();
        
        // 把当前线程包装成一个node，放入条件队列中，并返回封装当前线程的node
        Node node = addConditionWaiter();
        // 完全释放当前线程对应的锁（将state置为0）:
        // 为什么要释放锁呢？ 因为加着锁挂起的时候，除了本线程能唤醒（本线程已挂起），其他线程都没办法唤醒啊~
        int savedState = fullyRelease(node);
        // Condition队列中断状态：
        // 0: 在Condition队列挂起期间，未接收过中断信号~
        // -1: 在Condition队列挂起期间，接收到中断信号~
        // 1: 在Condition队列挂起期间，未接收到中断信号，但是迁移到阻塞队列之后，接收到过中断信号~
        int interruptMode = 0;
        
        // while循环条件：判断当前node节点是否在阻塞队列中: 
        // isOnSyncQueue返回true：表示当前线程对应的node已经迁移到了阻塞队列中了
        // isOnSyncQueue返回false：说明当前node仍还在条件队列中，需要park挂起~
        while (!isOnSyncQueue(node)) { 
            // 挂起当前线程，使其处于等待状态~
            LockSupport.park(this);
            
            //什么时候会被唤醒？都有几种情况呢？
            // 1.常规路径：外部线程获取到lock之后，调用了 signal()方法 转移条件队列的头节点到 阻塞队列， 当这个节点获取到锁后，会唤醒。
            // 2.转移至阻塞队列后，发现阻塞队列中的前驱节点状态 是 取消状态，此时会唤醒当前节点
            // 3.当前节点挂起期间，被外部线程使用中断唤醒..
    
            // checkInterruptWhileWaiting: 即使在Condition队列挂起期间，线程发生中断了，对应的node仍然会被迁移到阻塞队列中（等待去获取锁）~
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        
        // 执行到这里，说明当前node已经迁移到阻塞队列了
        
        // acquireQueued：竞争队列的逻辑，线程节点竞争资源
        // 条件一：返回true 表示在阻塞队列中 被外部线程中断唤醒过..
        // 条件二：interruptMode != THROW_IE 成立，说明当前node在条件队列内 未发生过中断
        // 设置interruptMode = REINTERRUPT
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        // 考虑下 node.nextWaiter != null 条件什么时候成立呢？
        // 其实是node在条件队列内时 如果被外部线程 中断唤醒时，会加入到阻塞队列，但是并未设置nextWaiter = null。
        if (node.nextWaiter != null) // clean up if cancelled
            // 清理条件队列内取消状态的节点..
            unlinkCancelledWaiters();
    
        // 条件成立：说明挂起期间 发生过中断（1.条件队列内的挂起 2.条件队列之外的挂起）
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }
    
    // 把当前线程包装成一个node，放入条件队列中，并返回封装当前线程的node
    // 注意：调用await方法的线程都是持锁状态的线程，也就是说：在await方法中调用的addConditionWaiter方法不存在并发问题~
    private Node addConditionWaiter() { 
        // 获取当前条件队列的尾节点的引用，保存到局部变量t中
        Node t = lastWaiter;
        // 条件1：t != null 成立：说明当前条件队列中，已经有node元素了
        // 条件2：(node在队列中时，它的状态是 CONDITION(-2))
        // t.waitStatus != Node.CONDITION 成立：说明当前node发生中断了...
        if (t != null && t.waitStatus != Node.CONDITION) { 
            // 清理条件队列中，所有CONDITION(-2)取消状态的node节点
            unlinkCancelledWaiters();
            // 更新局部变量t 为最新的尾结点的引用~ 
            // 因为unlinkCancelledWaiters方法可能会更改lastWaiter的引用
            t = lastWaiter;
        }
        // 为当前线程创建node节点，设置状态为 CONDITION(-2)
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        
        // 如果t == null 条件成立，说明条件队列中没有任何元素，当前线程是进入队列的第一个元素
        if (t == null)
            firstWaiter = node;
        else
            t.nextWaiter = node;
        
        // 更新队尾节点引用指向node
        lastWaiter = node;
        
        // 返回包装当前线程的node
        return node;
    }
    
    // 即使在Condition队列挂起期间，线程发生中断了，对应的node仍然会被迁移到阻塞队列中（等待去获取锁）~
    private int checkInterruptWhileWaiting(Node node) { 
        // Thread.interrupted() 返回当前线程中断标记位，并且重置当前标记位为false
        return Thread.interrupted() ?
                // transferAfterCancelledWait 这个方法只有在线程是被中断唤醒时才会调用！
            (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 0;
    }
    
    final boolean transferAfterCancelledWait(Node node) { 
        // 条件成立：说明当前node一定是在 条件队列内，因为signal 迁移节点到阻塞队列时，会将节点的状态修改为0
        if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) { 
            // 中断唤醒的node也会被加入到 阻塞队列中！！
            enq(node);
            // true：表示是在条件队列内被中断的.
            return true;
        }
    
        // 执行到这里有几种情况？
        // 1.当前node已经被外部线程调用 signal 方法将其迁移到 阻塞队列内了。
        // 2.当前node正在被外部线程调用 signal 方法将其迁移至 阻塞队列中 进行中状态..
    
        while (!isOnSyncQueue(node))
            Thread.yield();
    
        // false:表示当前节点被中断唤醒时 不在 条件队列了..
        return false;
    }
    
    // 清理条件队列内取消状态的节点..(纯链表操作的方法)
    private void unlinkCancelledWaiters() { 
        // 表示循环当前节点，从链表的第一个节点开始 向后迭代处理.
        Node t = firstWaiter;
        // 当前链表上一个正常状态的node节点
        Node trail = null;
    
        while (t != null) { 
            // 当前节点的下一个节点.
            Node next = t.nextWaiter;
            // 条件成立：说明当前节点状态为 取消状态
            if (t.waitStatus != Node.CONDITION) { 
                // 更新nextWaiter为null
                t.nextWaiter = null;
                // 条件成立：说明遍历到的节点还未碰到过正常节点..
                if (trail == null)
                    // 更新firstWaiter指针为下个节点就可以
                    firstWaiter = next;
                else
                    // 让上一个正常节点指向 取消节点的 下一个节点..中间有问题的节点 被跳过去了..
                    trail.nextWaiter = next;
    
                // 条件成立：当前节点为队尾节点了，更新lastWaiter 指向最后一个正常节点 就Ok了
                if (next == null)
                    lastWaiter = trail;
            }
            else// 条件不成立执行到else，说明当前节点是正常节点
                trail = t;
    
            t = next;
        }
    }
    
    private void reportInterruptAfterWait(int interruptMode)
        throws InterruptedException { 
        // 条件成立：说明在条件队列内发生过中断，此时await方法抛出中断异常
        if (interruptMode == THROW_IE)
            throw new InterruptedException();
    
        // 条件成立：说明在条件队列外发生的中断，此时设置当前线程的中断标记位 为true
        // 中断处理 交给 你的业务处理。 如果你不处理，那什么事 也不会发生了...
        else if (interruptMode == REINTERRUPT)
            selfInterrupt();
    }

### 3.2 fullyRelease完全释放当前线程锁的方法 ###

// 位于AQS中：
    // 完全释放当前线程对应的锁（将state置为0）:
    final int fullyRelease(Node node) { 
        // 完全释放锁是否成功，当failed失败时，说明当前线程是未持有锁调用await方法的线程(错误写法...)
        // 假设失败，在finally代码块中，会将刚刚加入到条件队列的，当前线程对应的node节点状态修改为取消状态
        // 后继线程就会将取消状态的节点给清理出去~ 
        boolean failed = true;
        try { 
            // 获取当前线程所持有的state值。
            int savedState = getState();
            
            // 绝大部分情况下：release这里会返回true
            if (release(savedState)) { 
                // 失败标记设置为false
                failed = false;
                // 返回当前线程释放的state值
                return savedState;
            } else { 
                throw new IllegalMonitorStateException();
            }
        } finally { 
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }

### 3.3 isOnSyncQueue判断当前node节点是否在阻塞队列中 ###

// 位于AQS中：
    // 判断当前node节点是否在阻塞队列中
    final boolean isOnSyncQueue(Node node) { 
        // 条件一：node.waitStatus == Node.CONDITION 条件成立：说明当前node一定是在条件队列中，
        // 因为signal方法迁移节点到阻塞队列前，会将node的状态设置为0
        // 条件二：前置条件node.waitStatus != Node.CONDITION ===>然后再细分以下几种情况：
        // 1.node.waitStatus == 0 (表示当前节点已经被singal)
        // 2.node.waitStatus == 1 (表示当前线程未持有锁就调用了await方法，最终会将node的状态修改为取消状态...)
        // node.waitStatus == 0 为什么还要判断node.prev == null呢？
        // 因为：signal方法是先修改状态，再迁移
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        
        // 执行到这里，会是哪种情况？
        // node.waitStatus != CONDITION 且 node.prev != null ===> 可以排除掉 node.waitStatus == 1 取消状态..
        // 为什么可以排除取消状态？ 因为signal方法是不会把 取消状态的node迁移走的
        // 设置prev引用的逻辑 是 迁移 阻塞队列 逻辑的设置的（enq()）
        // 入队的逻辑：1.设置node.prev = tail; 2. cas当前node为 阻塞队列的 tail 尾节点 成功才算是真正进入到 阻塞队列！ 3.pred.next = node;
        // 可以推算出，就算prev不是null，也不能说明当前node 已经成功入队到 阻塞队列了。
        
        // 条件成立：说明当前节点已经成功入队到阻塞队列，且当前节点后面已经有其它node了...
        if (node.next != null)
            return true;
        
        /** * 执行到这里，说明当前节点的状态为：node.prev != null 且 node.waitStatus == 0 * findNodeFromTail 从阻塞队列的尾巴开始向前遍历查找node，如果查找到 返回true,查找不到返回false * 当前node有可能正在signal过程中，正在迁移中...还未完成... */
        return findNodeFromTail(node);
    }

### 3.4 signal线程唤醒方法 ###

// 位于AQS中：线程唤醒方法
    public final void signal() { 
        // 判断调用signal方法的线程是否是独占锁持有线程，如果不是，直接抛出异常..
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
    
        // 获取条件队列的一个node
        Node first = firstWaiter;
        // 第一个节点不为null，则将第一个节点 进行迁移到阻塞队列的逻辑..
        if (first != null)
            doSignal(first);
    }
    
    // 位于AQS中：将第一个节点 进行迁移到阻塞队列 
    private void doSignal(Node first) { 
        do { 
            // firstWaiter = first.nextWaiter因为当前first马上要出条件队列了，
            // 所以更新firstWaiter为 当前节点的下一个节点..
            // 如果当前节点的下一个节点 是 null，说明条件队列只有当前一个节点了...
            // 当前出队后，整个队列就空了..所以需要更新lastWaiter = null
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
    
            // 当前first节点 出 条件队列。断开和下一个节点的关系
            first.nextWaiter = null;
    
            // transferForSignal(first) 返回boolean类型
            // 返回true 表示当前first节点迁移到阻塞队列成功 返回false 表示迁移失败...
            // while循环 ：(first = firstWaiter) != null 
            // 当前first迁移失败，则将first更新为 first.next 继续尝试迁移..
            // 直至迁移某个节点成功，或者 条件队列为null为止。
        } while (!transferForSignal(first) &&
                 (first = firstWaiter) != null);
    }

### 3.4 transferForSignal当前节点迁移到阻塞队列 ###

// 位于AQS：当前节点迁移到阻塞队列：
    // 返回true 表示当前first节点迁移到阻塞队列成功 返回false 表示迁移失败...
    final boolean transferForSignal(Node node) { 
        // cas修改当前节点的状态，修改为0，因为当前节点马上要迁移到 阻塞队列了
        // 成功：当前节点在条件队列中状态正常。
        // 失败： 1.取消状态 （线程await时 未持有锁，最终线程对应的node会设置为 取消状态）
        // 2.node对应的线程 挂起期间，被其它线程使用 中断信号 唤醒过...
        // （就会主队进入到 阻塞队列，这时也会修改状态为0）
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
    
        // enq最终会将当前 node 入队到 阻塞队列，p 是当前节点在阻塞队列的 前驱节点.
        Node p = enq(node);
    
        // ws 前驱节点的状态..
        int ws = p.waitStatus;
        // 条件一：ws > 0 成立：说明前驱节点的状态在阻塞队列中是 取消状态,唤醒当前节点。
        // 条件二：前置条件(ws <= 0)，
        // compareAndSetWaitStatus(p, ws, Node.SIGNAL) 返回true 表示设置前驱节点状态为 SIGNAl状态成功
        // compareAndSetWaitStatus(p, ws, Node.SIGNAL) 返回false ===> 什么时候会false?
        // 当前驱node对应的线程 是 lockInterrupt入队的node时，是会响应中断的，外部线程给前驱线程中断信号之后，前驱node会将
        // 状态修改为 取消状态，并且执行 出队逻辑..
        // 前驱节点状态 只要不是 0 或者 -1 那么，就唤醒当前线程。
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            // 唤醒当前node对应的线程...回头再说。
            LockSupport.unpark(node.thread);
    
        return true;
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzU5MTk4MA_size_16_color_FFFFFF_t_70]: /images/20221021/9b56fc4f86dc487893f461ffda4a34bf.png