Java多线程系列 JUC锁07 ConditionObject分析

ConditionObject　

　ConditionObject是AQS中的内部类，提供了条件锁的同步实现，实现了Condition接口，并且实现了其中的await(),signal(),signalALL()等方法。

ConditionObject主要是为并发编程中的同步提供了等待通知的实现方式，可以在不满足某个条件的时候挂起线程等待。直到满足某个条件的时候在唤醒线程。

　　在一个AQS同步器中，可以定义多个Condition，只需要多次lock.newCondition(),每次都会返回一个新的ConditionObject对象。在ConditionObject中，通过一个等待队列来维护条线等待的线程。所以在一个同步器中可以有多个等待队列，他们等待的条件是不一样的。

条件队列

　　条件队列是一个FIFO的队列，在队列的每个节点都包含了一个线程引用。该线程就是在Condition对象上等待的线程。这里的节点和AQS中的同步队列中的节点一样，使用的都是AbstractQueuedSynchronizer.Node类。每个调用了condition.await()的线程都会进入到条件队列中去。在Condition中包含了firstWaiter和lastWaiter，每次加入到条件队列中的线程都会加入到条件队列的尾部，来构成一个FIFO的条件队列。

方法分析

await()方法的具体实现

public final void await() throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        Node node = addConditionWaiter();　　//把当前线程的节点加入到CONDITION队列中
        int savedState = fullyRelease(node); //由于调用await()方法的线程是已经获取了锁的，所以在加入到CONDITION队列之后，需要去释放锁，并且唤醒后继节点线程
        int interruptMode = 0;
        while (!isOnSyncQueue(node)) { //判断该节点是否在CLH队列中，不在 说明它还没有竞争锁的资格，所以继续将自己沉睡,线程进入自旋。
            LockSupport.park(this);　　//挂起当前线程，当别的线程调用了signal（），并且是当前线程被唤醒的时候才从park()方法返回
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        //出了while循环，代表线程被唤醒，并且已经将该node从CONDITION队列transfer到了CLH队列中
        //acquireQueued在队列中获取锁，如果竞争不到还是会阻塞当前线程，并且在上面while循环等待的过程中没有发生异常，则修改interruptMode状态为REINTERRUPT
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE) //当被唤醒后，该线程会尝试去获取锁，只有获取到了才会从await()方法返回，否则的话，会挂起自己
            interruptMode = REINTERRUPT;

        //该节点调用transferAfterCancelledWait添加到CLH队列中的，此时该节点的nextWaiter不为null，需要调用unlinkCancelledWaiters将该节点从CONDITION队列中删除，该节点的状态为0
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();

        //如果interruptMode不为0，则代表该线程在上面过程中发生了中断或者抛出了异常，则调用reportInterruptAfterWait方法在此处抛出异常
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }

 addConditionWaiter( )

 private Node addConditionWaiter() {
        Node t = lastWaiter;
         if (t != null && t.waitStatus != Node.CONDITION) { //首先判断lastWaiter节点是否为空，或者是否是处于条件等待，如果不是的话则把它从等待队列中删除。
            unlinkCancelledWaiters();
            t = lastWaiter;
        }
        Node node = new Node(Thread.currentThread(), Node.CONDITION); //创建一个node节点，状态为CONDITION
        if (t == null)   //把当前线程构建的节点加入到CONDITION队列中去，并且返回当前节点
        firstWaiter = node;
            else
        t.nextWaiter = node;
        lastWaiter = node;
        return node;
 }

fullyRelease( ) 

//完全释放锁,释放成功则返回，失败则将当前节点的状态设置成cancelled表示当前节点失效
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; //释放锁失败，则当前节点的状态变为cancelled（此时该节点在CONDITION队列中）
       }
}

 signal()

//对CONDITION队列中从首部开始的第一个CONDITION状态的节点，执行transferForSignal操作，将node从CONDITION队列中转换到CLH队列中，同时修改CLH队列中原先尾节点的状态
private void doSignal(Node first) {
        do {
            //当前循环将first节点从CONDITION队列transfer到CLH队列
            //从CONDITION队列中删除first节点，调用transferForSignal将该节点添加到CLH队列中，成功则跳出循环  
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            first.nextWaiter = null;
        } while (!transferForSignal(first) && 
        (first = firstWaiter) != null);
}

//唤醒CONDITION队列中首部的第一个CONDITION状态的节点
public final void signal() {
        if (!isHeldExclusively()) //判断当前线程是不是独占的持有锁，如果不是，则当前线程不能signal其他线程 。
            throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)  //CONDITION队列不为null，则doSignal方法将唤醒CONDITION队列中所有的节点线程
            doSignal(first);
}

transferForSignal( )

//两步操作，首先enq将该node添加到CLH队列中，其次若CLH队列原先尾节点为CANCELLED或者对原先尾节点CAS设置成SIGNAL失败，则唤醒node节点；
//否则该节点在CLH队列总前驱节点已经是signal状态了，唤醒工作交给前驱节点（节省了一次park和unpark操作）
final boolean transferForSignal(Node node) {

　　 //如果CAS失败，则当前节点的状态为CANCELLED
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;
    Node p = enq(node);  //enq将node添加到CLH队列队尾，返回node的前一个节点

    //如果p是一个取消了的节点，或者对p进行CAS设置失败，则唤醒node节点，让node所在线程进入到acquireQueue方法中，重新进行相关操作
    //否则，由于该节点的前驱节点已经是signal状态了，不用在此处唤醒await中的线程，唤醒工作留给CLH队列中前驱节点
     int ws = p.waitStatus;
     if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);　　//唤醒刚加入到同步队列的线程，被唤醒之后，该线程才能从await()方法的park()中返回。
     return true;
 }

signalAll( ) 

public final void signalAll() {
    //查看当前线程是否独占锁，若不是，则当前线程没有权限执行signalAll操作，抛出异常
    if (!isHeldExclusively())
         throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    //唤醒CONDITION队列中所有节点，同时transfer到CLH队列中
    if (first != null)
        doSignalAll(first);
}
private void doSignalAll(Node first) {
        lastWaiter = firstWaiter = null;
        do {
            //将first节点从CONDITION队列中出队
            Node next = first.nextWaiter;
            first.nextWaiter = null;
            //将first节点在CLH队列中入队，同时可能需要执行unpark操作
             transferForSignal(first);
            //更新first的指向
            first = next;
        } while (first != null);
}

条件队列的节点状态

 调用条件队列的等待操作，会设置节点的waitingStatus为Condition，标识当前节点正处于条件队列中。条件队列的节点状态转换图如下：

   Node的各个状态的主要作用：Cancelled主要是解决线程在持有锁时被外部中断的逻辑，AQS的可中断锁获取方法lockInterrutible()是基于该状态实现的。显式锁必须手动释放锁，尤其是有中断的环境中，一个线程被中断可能仍然持有锁，所以必须注意在finally中unlock。Condition则是支持条件队列的等待操作，是Lock与条件队列关联的基础。Signal是阻塞后继线程的标识，一个等待线程只有在其前驱节点的状态是SIGNAL时才会被阻塞，否则一直执行自旋尝试操作，以减少线程调度的开销。

等待和唤醒操作

     条件队列上的等待和唤醒操作，本质上是节点在AQS线程等待队列和条件队列之间相互转移的过程，当需要等待某个条件时，线程会将当前节点添加到条件队列中，并释放持有锁；当某个线程执行条件队列的唤醒操作，则会将条件队列的节点转移到AQS等待队列。每个Condition就是一个条件队列，可以通过Lock的newCondition创建多个等待条件。AQS的条件队列，它的等待和唤起操作流程如下：

　　显式条件队列弥补了内置条件队列只能关联一个条件的缺陷，同时继承了Lock对象的公平性。在Condition对象中，与Object的wait/notify/notifyAll对应的扩展方法是await/signal/signallAll，同时它也具有Object的这三个方法，所以使用的时候需要注意使用版本的正确。另外，显式锁必须遵从特定的使用规范，先lock，然后再在finally中unlock，以确保锁必然会被正确释放。

Condition示例

package concurrent;

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

/**
 * @Description: Condition实例
 * @Author: lizhouwei
 * @CreateDate: 2018/5/27 16:39
*/
public class ConditionDemo {
     private static Lock reentrantLock = new ReentrantLock();
     private static Condition condition = reentrantLock.newCondition();

     public static void main(String[] args) {

          ThreadA threadA = new ThreadA("threadA");

          reentrantLock.lock();
          try {
               System.out.println(Thread.currentThread().getName()+" 我拿到锁了");
               threadA.start();
               System.out.println(Thread.currentThread().getName()+" 我释放锁了");
               condition.await();
               System.out.println(Thread.currentThread().getName()+" 我又拿到锁了");
          } catch (InterruptedException e) {
               e.printStackTrace();
          } finally {
               reentrantLock.unlock();
          }
     }

     static class ThreadA extends Thread{

          public ThreadA(String name) {
               super(name);
          }

          public void run() {
               reentrantLock.lock();
               try {
                    System.out.println(Thread.currentThread().getName()+"我拿到锁了");
                    condition.signal();
                    System.out.println(Thread.currentThread().getName()+"我唤醒其他线程了");
               } finally {
                    reentrantLock.unlock();
               }
          }
     }
}

运行结果

可以看到Condition的执行方式，

1. 线程1调用reentrantLock.lock时，线程被加入到AQS的等待队列中。
2. 线程1调用wait方法被调用时，该线程从AQS中移除，对应操作是锁的释放。
3. 接着线程1马上被加入到Condition的等待队列中，意味着该线程需要signal信号。
4. 线程2，因为线程1释放锁的关系，被唤醒，并判断可以获取锁，于是线程2获取锁。
5. 线程2调用signal方法，这个时候Condition的等待队列中只有线程1一个节点，于是它被取出来，并被加入到AQS的等待队列中。 注意，这个时候，线程1 并没有被唤醒。
6. signal方法执行完毕，线程2调用reentrantLock.unLock()方法，释放锁。这个时候因为AQS中只有线程1，于是，AQS释放锁后按从头到尾的顺序唤醒线程时，线程1被唤醒，于是线程1回复执行。
7. 直到释放所整个过程执行完毕。

可以看到，整个协作过程是靠结点在AQS的等待队列和Condition的等待队列中来回移动实现的，Condition作为一个条件类，很好的自己维护了一个等待信号的队列，并在适时的时候将结点加入到AQS的等待队列中来实现的唤醒操作。