AQS同步队列器之二：等待通知机制

一、简介

　　　　Condition是在java 1.5中才出现的，它用来替代传统的Object的wait()、notify()实现线程间的协作，相比使用Object的wait()、notify()，使用Condition的await()、signal()这种方式实现线程间协作更加安全和高效。简单说，他的作用是使得某些线程一起等待某个条件（Condition），只有当该条件具备(signal 或者 signalAll方法被调用)时，这些等待线程才会被唤醒，从而重新争夺锁。wait()、notify()这些都更倾向于底层的实现开发，而Condition接口更倾向于代码实现的等待通知效果。两者之间的区别与共通点也可以了解一下：

对比项	Object监视器	Condition
前置条件	获取对象的锁	调用Lock.lock()获取锁 Lock.newCondition获取Condition对象
调用方式	直接调用Object.notify()	直接调用condition.await()
等待队列的个数	一个	多个
当前线程释放锁进入等待状态	支持	支持
当前线程释放锁进入等待状态在等待状态中不断响应中断	不支持	支持
当前线程释放锁并进入等待超时状态	支持	支持
当前线程释放锁并进入等待状态直到将来的某个时间	不支持	支持
唤醒等待队列中的一个线程	支持notify()	支持condition.signal()
唤醒等待队列中的全部线程	支持notifyAll()	支持condition.signalAll()

相比之下Condition提供了比Object监视器更方便更全面的处理方式，而且使用起来也依旧很简单。

二、简单使用示例

package cn.memedai;  2 
import java.util.concurrent.locks.Condition;  4 import java.util.concurrent.locks.Lock;  5 import java.util.concurrent.locks.ReentrantLock;  6 
/**
 * Lock与Condition接口示例  9  */
public class LockConditionDemo {  11 
    //存储地方
    class Depot {  14         private int capacity;  15         private int size;  16         private Lock lock;  17         private Condition fullCondition;  18         private Condition emptyCondition;  19 
        public Depot(int capacity) {  21             this.capacity = capacity;  22             this.size = 0;  23             this.lock = new ReentrantLock();  24             this.fullCondition = lock.newCondition();  25             this.emptyCondition = lock.newCondition();  26  }  27 
        //生产操作
        public void produce(int newSize) throws InterruptedException {  30  lock.lock();  31             int left = newSize;  32             try {  33                 while (left > 0) {  34                     //代表超过了容量就不能再生产了
                    while (size >= capacity) {  36                         fullCondition.await();//进行等待处理
 }  38                     //获取实际生产的数量(及库存中新增的数量)  39                     //如果库存+要生产的大于了总的容量那么新增的就是总容量的数量相减
                    int inc = (size + left) > capacity ? (capacity - size) : left;  41                     size += inc;  42                     left -= inc;  43                     System.out.println(Thread.currentThread().getName() + "------left剩余" + left + "------size容量" + size + "-------inc增长" + inc);  44  emptyCondition.signal();  45  }  46             } finally {  47                 lock.unlock();//解锁
 }  49  }  50 
        //消费操作
        public void consume(int newSize) throws InterruptedException {  53  lock.lock();  54             try {  55                 int left = newSize;  56                 while (left > 0) {  57                     //库存为0等待生产者进行生产的操作
                    while (size <= 0) {  59  emptyCondition.await();  60  }  61                     int dec = (size < left) ? size : left;  62                     size -= dec;  63                     left -= dec;  64                     System.out.println(Thread.currentThread().getName() + "-------left剩余" + left + "-------size容量" + size + "--------减少量dec" + dec);  65  fullCondition.signal();  66  }  67             } finally {  68  lock.unlock();  69  }  70  }  71  }  72 
    //生产者
    class Producer{  75         private Depot depot;  76 
        public Producer(Depot depot) {  78             this.depot = depot;  79  }  80 
        //往存储地方生产
        public void produce(final int newSize){  83             new Thread(){  84  @Override  85                 public void run() {  86                     try {  87  depot.produce(newSize);  88                     } catch (InterruptedException e) {  89  e.printStackTrace();  90  }  91  }  92  }.start();  93  }  94  }  95 　　　//消费者
    class Customer{  97         private Depot depot;  98 
        public Customer(Depot depot) { 100             this.depot = depot; 101  } 102 　　　　 //进行消费
        public void consume(final int newSize){ 104             new Thread(){ 105  @Override 106                 public void run() { 107                     try { 108  depot.consume(newSize); 109                     } catch (InterruptedException e) { 110  e.printStackTrace(); 111  } 112  } 113  }.start(); 114  } 115  } 116 
    public static void main(String[] args) { 118         Depot depot = new LockConditionDemo().new Depot(100); 119         Producer producer = new LockConditionDemo().new Producer(depot); 120         Customer customer = new LockConditionDemo().new Customer(depot); 121         producer.produce(60); 122         producer.produce(120); 123         customer.consume(90); 124         customer.consume(150); 125         producer.produce(110); 126  } 127 }

下面是这段代码运行的一种结果：

Thread-1------left剩余20------size容量100-------inc增长100 Thread-2-------left剩余0-------size容量10--------减少量dec90 Thread-3-------left剩余140-------size容量0--------减少量dec10 Thread-4------left剩余10------size容量100-------inc增长100 Thread-3-------left剩余40-------size容量0--------减少量dec100 Thread-4------left剩余0------size容量10-------inc增长10 Thread-3-------left剩余30-------size容量0--------减少量dec10 Thread-1------left剩余0------size容量20-------inc增长20 Thread-3-------left剩余10-------size容量0--------减少量dec20 Thread-0------left剩余0------size容量60-------inc增长60 Thread-3-------left剩余0-------size容量50--------减少量dec10 

通过简单的示例，使用Condition具备两个条件，首先线程一定需要获取到当前的同步状态，其次必须从锁中获取到Condition对象，而condition.await()方法就对应了Object.wait()方法使得当前线程在满足某种条件的时候就进行等待，condition.signal()就是在某种条件下唤醒当前线程。其配合lock接口的使用非常方便。

三、Condition等待/通知机制的实现原理

　　　  首先可以看一下Condition接口的定义的相关方法：

　　　 

　　　 await()：使当前线程进入等待状态直到被signal()、signalAll()方法唤醒或者被中断

　　　 signal()：唤醒等待中的一个线程

　　　 signalAll()：唤醒等待中的全部线程

Condition接口只是定义了相关的处理等待通知的方法，真正实现其等待通知效果的在AQS中的ConditionObject类，在了解源码之前先讲一下同步队列和等待队列：

　　　　前面的文章讲过当线程未获取到同步状态的时候，会创建一个Node节点并把这个节点放入同步队列的尾部，进入同步队列的中的线程都是阻塞的。

　　　　在AQS中同步队列和等待队列都复用了Node这个节点类，一个同步状态可以含有多个等待队列，同时等待队列只是一个单向的队列。

 

一个AQS可以存在一个同步对列多个等待队列

接下来可以看一些重点方法的源码了解一下原理

await()：使当前线程进入等待状态

 　　　　 public final void await() throws InterruptedException {  2 if (Thread.interrupted())//响应中断  3 throw new InterruptedException();  4 Node node = addConditionWaiter();//放入到等待队列中  5 int savedState = fullyRelease(node);//释放同步状态(同步队列头节点释放状态唤醒后继节点获取同步状态)  6 int interruptMode = 0;
　　　　　　　　　//判断是否在同步队列中  7 while (!isOnSyncQueue(node)) {  8 LockSupport.park(this);//存在等待队列中就阻塞该线程  9 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)//判断等待过程中是否被中断过 10 break; 11  }
　　　　　　　　 //自旋去获取同步状态【在AQS中了解】获取成功并且在退出等待时不抛出中断异常（抛出了异常就会立马被中断） 12 if (acquireQueued(node, savedState) && interruptMode != THROW_IE) 13 interruptMode = REINTERRUPT;//在退出等待时重新中断 14 if (node.nextWaiter != null) //如果存在其他节点 15  unlinkCancelledWaiters();//移除所有不是等待状态的节点 16 if (interruptMode != 0) 17  reportInterruptAfterWait(interruptMode);//如果在等待过程中发现被中断，就执行中断的操作 18 }

 

addConditionWaiter()：往等待队列中添加元素

 　　　　 private Node addConditionWaiter() {  2 Node t = lastWaiter;//等待队列中的最后一个元素  4 if (t != null && t.waitStatus != Node.CONDITION) {//如果尾节点部位null，并且尾节点不是等待状态中说明这个节点不应该待在等待队列中  5  unlinkCancelledWaiters();//从等待队列中移除  6 t = lastWaiter;  7  }  8 Node node = new Node(Thread.currentThread(), Node.CONDITION);//创建一个等待状态的节点  9 if (t == null) 10 firstWaiter = node; 11 else 12 t.nextWaiter = node; 13 lastWaiter = node;//加入等待队列的尾部 14 return node; 15 }

 

unlinkCancelledWaiters()：将不是等待状态的节点从等待队列中移除

　　　　private void unlinkCancelledWaiters() {  2             Node t = firstWaiter;//头节点  3             Node trail = null;  4             while (t != null) {//存在节点  5                 Node next = t.nextWaiter;//下一个节点  6                 if (t.waitStatus != Node.CONDITION) {//如果不是出于等待中的状态  7                     t.nextWaiter = null;//t的后指针引用清除  8                     if (trail == null)//前面是否存在节点  9                         firstWaiter = next;//下一个节点就是头节点 10                     else
                        trail.nextWaiter = next;//赋值给前节点的后指针引用 12                     if (next == null)//代表不存在元素了 13                         lastWaiter = trail; 14  } 15                 else
                    trail = t;//将t赋值给trail 17                 t = next;//next赋值给t 18  } 19         }

 

fullyRelease(Node node)：释放当前状态值，返回同步状态

　　 final int fullyRelease(Node node) {  2         boolean failed = true;//失败状态  3         try {  4             int savedState = getState();//获取当前同步状态值  5             if (release(savedState)) {//独占模式下释放同步状态，AQS独占式释放锁、前面文章讲过  6                 failed = false;//失败状态为false  7                 return savedState;//返回同步状态  8             } else {  9                 throw new IllegalMonitorStateException(); 10  } 11         } finally { 12             if (failed) 13                 node.waitStatus = Node.CANCELLED;//取消等待状态 14  } 15     }

 

isOnSyncQueue：判断线程是否在同步队列中

    final boolean isOnSyncQueue(Node node) { if (node.waitStatus == Node.CONDITION || node.prev == null)//如果等待状态为等待中，或者前继节点为null代表第一种情况该节点出于等待状态，第二种情况可能已经被唤醒不在等待队列中了 return false; if (node.next != null) //如果后继节点不为null代表肯定在等待队列中
            return true;
        return findNodeFromTail(node);//从后往前找判断是否在等待队列中 }

总结一下等待操作：

　　　　首先等待操作没有进行CAS或者任何的同步操作，因为调用await()方法的是获取当前lock锁对象的线程，也就是同步队列中的首节点，当调用await()方法后，将同步队列的首节点创建一个等待节点放入等待队列的尾部，然后释放出同步状态（不释放同步状态就会造成死锁），唤醒同步队列中的后继节点，然后当前线程进入等待的状态

 

调用await()方法过程　　　

signal()：唤醒等待队列中的一个线程

　　　　public final void signal() { 2             if (!isHeldExclusively())//判断当前线程是否已经获取同步状态
                throw new IllegalMonitorStateException(); 4             Node first = firstWaiter;//等待队列头节点
            if (first != null) 6                 doSignal(first);//具体实现方法唤醒第一个node
        }

 

doSignal(Node node)：具体处理唤醒节点的操作

 　　　　private void doSignal(Node first) { do { if((firstWaiter = first.nextWaiter) == null)//执行移除头节点的操作 lastWaiter = null; first.nextWaiter = null; } while (!transferForSignal(first) && (first = firstWaiter) != null); }

 

transferForSignal(Node node)：唤醒的具体实现方式

　　　final boolean transferForSignal(Node node) {
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))//将节点的等待状态设置更改为初始状态如果改变失败就会被取消 return false; Node p = enq(node);//往同步队列中添加节点【死循环方式】 int ws = p.waitStatus;//获取节点的等待状态 if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))//如果该结点的状态为cancel 或者修改waitStatus失败，则直接唤醒（这一步判断是为了不立刻唤醒脱离等待中的线程，因为他要等同步队列中的头节点释放同步状态再去竞争） LockSupport.unpark(node.thread);//具体的唤醒操作 return true; }

 总结一下唤醒操作的流程：当调用signal()方法时，将等待队列中的首节点拿出来，加入到同步队列中，此时该节点不会立刻被唤醒因为就算被唤醒也是需要重新去获取同步状态的，而是在调用lock.unlock()方法释放锁以后将其唤醒获取同步状态。

 调用signal()唤醒方法过程

 

到现在为止，基本的Condition的等待通知机制已经讲解完毕，至于附加功能的比如超时等待或者唤醒全部的功能在源码上都差不了多少稍微新增一些功能需要，在原有的await()方法上增加了一些处理逻辑，真正的原理还是相差无几的。

 

 

 

================================================================================== 

不管岁月里经历多少辛酸和艰难，告诉自己风雨本身就是一种内涵，努力的面对，不过就是一场命运的漂流，既然在路上，那么目的地必然也就是前方。

==================================================================================