线程条件队列ConditionObject源码解读

Posted 2022-09-10 Lovnx

小记
好久没更博，窗外光芒万丈，冬日的晚晨，多么美好，就不浪费了，循着键盘上的点点星辰，开工！

啥子是条件队列？

我们都知道，在万类之祖Object里面定义了几个监视器方法：wait()，notify
()，notifyAll()，配合synchronized语义来控制线程的一些状态，在JDK1.5之后，由Lock替代了synchronized，而这几个监视器由条件队列Condition来实现，以便在某个状态条件现在可能为 true 的另一个线程通知它之前，一直挂起该线程（即让其“等待”），以原子的方式释放锁，并挂起当前线程，所以，也可以叫它为线程的条件队列(自创的)。

来看一个应用示例

在以前刚学习Java的时候写过一个题，题干大概是这样的：开启三个线程依次轮流打印出75个数，且次序不能乱。下面是代码：

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class CountTo_75 

    public static void main(String[] args) 
        final Box box = new Box();

        new Thread(new Runnable()  // 线程1
            @Override
            public void run() 
                for (int i = 1; i <= 5; i++) 
                    box.main_1();
                
            
        ,"Thread-1").start();
        new Thread(new Runnable()  // 线程2
            @Override
            public void run() 
                for (int i = 1; i <= 5; i++) 
                    box.main_2();
                
            
        ,"Thread-2").start();
        new Thread(new Runnable()  // 线程3
            @Override
            public void run() 
                for (int i = 1; i <= 5; i++) 
                    box.main_3();
                
            
        ,"Thread-3").start();
    

    static class Box 
        Lock lock = new ReentrantLock();
        Condition condition_1 = lock.newCondition();
        Condition condition_2 = lock.newCondition();
        Condition condition_3 = lock.newCondition();
        private volatile int flag = 1;
        public static int count = 0;   //用于计数的变量

        public void main_1() 
            lock.lock();
            try 
                while(flag != 1)
                    try 
                        condition_1.await();
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                
                for (int j = 0; j < 5; j++) 
                    count++;
                    System.out.println(Thread.currentThread().getName() +" "+ count);
                
                flag = 2;
                condition_2.signal();
             finally 
                lock.unlock();
            
        

        public void main_2() 
            lock.lock();
            try 
                while(flag != 2)
                    try 
                        condition_2.await();
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                
                for (int j = 0; j < 5; j++) 
                    count++;
                    System.out.println(Thread.currentThread().getName() +" " + count);
                
                flag = 3;
                condition_3.signal();
             finally 
                lock.unlock();
            
        

        public void main_3() 
            lock.lock();
            try 
                while(flag != 3)
                    try 
                        condition_3.await();
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                
                for (int j = 0; j < 5; j++) 
                    count++;
                    System.out.println(Thread.currentThread().getName() + " " + count);
                
                flag = 1;
                condition_1.signal();
             finally 
                lock.unlock();
            
        
    

抛开当时混乱的逻辑和性能考虑不足不谈，这段丑陋的代码，终归是实现了功能，而且是运用了Lock和Condition来实现的，用在这里来说明Condition的语义再好不过了。代码中初始化了3个条件队列，分别来控制3个线程的挂起状态，flag变量则控制它们之间的关系。

与Lock之间的实现关系

Condition是一个接口，其实现类只有两个：AQS和AQLS，都以内部类的形式存在，内部类叫做ConditionObject，这里有点纳闷，既然这个类是为Lock专属定制的，为什么不在ReentrantLock里面来实现呢？放在AQS不会太臃肿吗？不知道Doug Lea上神当时是怎么考虑的。
由于在AQS中已经实现，因此在ReentrantLock里面对其操作也是很简单的，创建一个条件队列：

    public Condition newCondition() 
        return sync.newCondition();
    

sync中的实现：

    abstract static class Sync extends AbstractQueuedSynchronizer 
        private static final long serialVersionUID = -5179523762034025860L;
        final ConditionObject newCondition() 
            return new ConditionObject();
        
    

很简单吧？呵呵，下面来看ConditionObject

ConditionObject实现

首先定义了两个核心成员变量，条件队列的头节点和尾节点：

/** First node of condition queue. */
private transient Node firstWaiter;

/** Last node of condition queue. */
private transient Node lastWaiter;

1、核心方法：await() 不可中断的条件等待实现

public final void await() throws InterruptedException 
     //当前线程被中断则抛异常
     if (Thread.interrupted()) throw new InterruptedException();
     //添加进条件队列
     Node node = addConditionWaiter();
     //释放当前线程持有的锁
     int savedState = fullyRelease(node);
     int interruptMode = 0;
     //在这里一直查找创建的Node节点在不在Sync队列，不在就一直禁用当前线程
     while (!isOnSyncQueue(node)) 
         //park当前线程，直到唤醒
         LockSupport.park(this);
         //如被中断也跳出循环
         if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) 
         break;
     
     //此时已被唤醒，获取之前被释放的锁，
     if (acquireQueued(node, savedState) && interruptMode != THROW_IE)   
       interruptMode = REINTERRUPT;
     //移除节点，释放内存
     if (node.nextWaiter != null) // clean up if cancelled
       unlinkCancelledWaiters();
     //被中断后的操作
     if (interruptMode != 0)
       reportInterruptAfterWait(interruptMode);

简而言之，await()方法其实就是为当前线程创建一个Node节点，加入到Condition队列并释放锁，之后就一直查看这个节点是否在Sync队列中了(signal()方法将它移到Sync队列)，如果在的话就唤醒此线程，重新获取锁。此外，awaitNanos(long nanosTimeout) 方法和await(long time, TimeUnit unit) 方法的实现大同小异，只是在何时跳出while循环的时候加了一个超时罢了。
另外还有几个相关的方法也看一下：

/**
 *  添加一个Node节点到Condition队列中
 */
private Node addConditionWaiter() 
  Node t = lastWaiter;
  //如果尾节点被取消，就清理掉
  if (t != null && t.waitStatus != Node.CONDITION) 
      unlinkCancelledWaiters();
      t = lastWaiter;
  
  //新建状态为的CONDITION节点，并添加在尾部
  Node node = new Node(Thread.currentThread(), Node.CONDITION);
  if (t == null)
      firstWaiter = node;
  else
      t.nextWaiter = node;
  lastWaiter = node;
  return node;

    /**
     *  释放当前线程的state，实际还是调用tryRelease方法
     */
   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;
       
   

    /**
     *  检查当前节点在不在Sync队列
     */
   final boolean isOnSyncQueue(Node node) 
       //如果当前节点状态为CONDITION，一定还在Condition队列
       //如果Sync队列的前置节点为null，则表明当前节点一定还在Condition队列
       if (node.waitStatus == Node.CONDITION || node.prev == null)
           return false;
       //有后继节点，也有前置节点，那么一定在Sync队列
       if (node.next != null) // If has successor, it must be on queue
           return true;

       //倒查Node节点，前置节点不能为null，第一个if已经做了判断，其前置节点为non-null,但是当前节点也不在Sync也是可能的,因为CAS操作将其加入队列也可能失败，所以我们需要从尾部开始遍历确保其在队列
       return findNodeFromTail(node);
   

2、核心方法：signal() 将等待时间最长的线程（如果存在）从Condition队列中移动到拥有锁的Sync队列

   public final void signal() 
       //当前线程非独占线程，报非法监视器状态异常
       if (!isHeldExclusively())
           throw new IllegalMonitorStateException();
       //头节点是等待时间最长的节点
       Node first = firstWaiter;
       if (first != null)
           doSignal(first);
   

   private void doSignal(Node first) 
         do 
             //如果头节点的下一节点为null，则将Condition的lastWaiter置为null
             if ( (firstWaiter = first.nextWaiter) == null)
                 lastWaiter = null;
             //将头结点的下一个节点设为null
             first.nextWaiter = null;
             //被唤醒并且头节点不为null则结束循环
          while (!transferForSignal(first) &&
                  (first = firstWaiter) != null);
     

   final boolean transferForSignal(Node node) 
       //如果无法改变节点状态，说明节点已经被唤醒
       if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
           return false;
       //将当前节点添加到Sync队列尾部，并设置前置节点的waitStatus为SIGNAL，表明后继有节点（可能）将被唤醒，如果取消或者设置waitStatus失败，会唤醒重新同步操作，这时候waitStatus是瞬时的，出现错误也是无妨的
       Node p = enq(node);
       int ws = p.waitStatus;
       if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
           LockSupport.unpark(node.thread);
       return true;
   

3、核心方法：signalAll() 将所有线程从Condition队列移动到拥有锁的Sync队列中。

   public final void signalAll() 
       if (!isHeldExclusively())
           throw new IllegalMonitorStateException();
       Node first = firstWaiter;
       if (first != null)
           doSignalAll(first);
   

    /**
     * 遍历所有节点，加入到拥有锁的Sync队列
     */
   private void doSignalAll(Node first) 
       lastWaiter = firstWaiter = null;
       do 
           Node next = first.nextWaiter;
           first.nextWaiter = null;
           transferForSignal(first);
           first = next;
        while (first != null);