先来回顾一下java中的等待/通知机制
我们有时会遇到这样的场景：线程A执行到某个点的时候，因为某个条件condition不满足，需要线程A暂停；等到线程B修改了条件condition，使condition满足了线程A的要求时，A再继续执行。
自旋实现的等待通知
最简单的实现方法就是将condition设为一个volatile的变量，当A线程检测到条件不满足时就自旋，类似下面：
1public class Test { 2 private static volatile int condition = 0; 3 4 public static void main(String[] args) throws InterruptedException { 5 Thread A = new Thread(new Runnable() { 6 @Override 7 public void run() { 8 while (!(condition == 1)) { 9 // 条件不满足，自旋 10 } 11 System.out.println("a executed"); 12 } 13 }); 14 15 A.start(); 16 Thread.sleep(2000); 17 condition = 1; 18 } 19 20}
这种方式的问题在于自旋非常耗费CPU资源，当然如果在自旋的代码块里加入Thread.sleep(time)将会减轻CPU资源的消耗，但是如果time设的太大，A线程就不能及时响应condition的变化，如果设的太小，依然会造成CPU的消耗。
Object提供的等待通知
因此，java在Object类里提供了wait()和notify()方法，使用方法如下：
1class Test1 { 2 private static volatile int condition = 0; 3 private static final Object lock = new Object(); 4 5 public static void main(String[] args) throws InterruptedException { 6 Thread A = new Thread(new Runnable() { 7 @Override 8 public void run() { 9 synchronized (lock) { 10 while (!(condition == 1)) { 11 try { 12 lock.wait(); 13 } catch (InterruptedException e) { 14 Thread.currentThread().interrupt(); 15 } 16 } 17 System.out.println("a executed by notify"); 18 } 19 } 20 }); 21 A.start(); 22 Thread.sleep(2000); 23 condition = 1; 24 synchronized (lock) { 25 lock.notify(); 26 } 27 } 28}
通过代码可以看出，在使用一个对象的wait()、notify()方法前必须要获取这个对象的锁。
当线程A调用了lock对象的wait()方法后，线程A将释放持有的lock对象的锁，然后将自己挂起，直到有其他线程调用notify()/notifyAll()方法或被中断。可以看到在lock.wait()前面检测condition条件的时候使用了一个while循环而不是if，那是因为当有其他线程把condition修改为满足A线程的要求并调用notify()后，A线程会重新等待获取锁，获取到锁后才从lock.wait()方法返回，而在A线程等待锁的过程中，condition是有可能再次变化的。
因为wait()、notify()是和synchronized配合使用的，因此如果使用了显示锁Lock，就不能用了。所以显示锁要提供自己的等待/通知机制，Condition应运而生。
显示锁提供的等待通知
我们用Condition实现上面的例子：
1class Test2 { 2 private static volatile int condition = 0; 3 private static Lock lock = new ReentrantLock(); 4 private static Condition lockCondition = lock.newCondition(); 5 6 public static void main(String[] args) throws InterruptedException { 7 Thread A = new Thread(new Runnable() { 8 @Override 9 public void run() { 10 lock.lock(); 11 try { 12 while (!(condition == 1)) { 13 lockCondition.await(); 14 } 15 } catch (InterruptedException e) { 16 Thread.currentThread().interrupt(); 17 } finally { 18 lock.unlock(); 19 } 20 System.out.println("a executed by condition"); 21 } 22 }); 23 A.start(); 24 Thread.sleep(2000); 25 condition = 1; 26 lock.lock(); 27 try { 28 lockCondition.signal(); 29 } finally { 30 lock.unlock(); 31 } 32 } 33}
可以看到通过 lock.newCondition() 可以获得到 lock 对应的一个Condition对象lockCondition ，lockCondition的await()、signal()方法分别对应之前的Object的wait()和notify()方法。整体上和Object的等待通知是类似的。
应用举例
上面我们看到了Condition实现的等待通知和Object的等待通知是非常类似的，而Condition提供的等待通知功能更强大，最重要的一点是，一个lock对象可以通过多次调用 lock.newCondition() 获取多个Condition对象，也就是说，在一个lock对象上，可以有多个等待队列，而Object的等待通知在一个Object上，只能有一个等待队列。用下面的例子说明，下面的代码实现了一个阻塞队列，当队列已满时，add操作被阻塞有其他线程通过remove方法删除元素；当队列已空时，remove操作被阻塞直到有其他线程通过add方法添加元素。
1public class BoundedQueue1<T> { 2 public List<T> q; //这个列表用来存队列的元素 3 private int maxSize; //队列的最大长度 4 private Lock lock = new ReentrantLock(); 5 private Condition addConditoin = lock.newCondition(); 6 private Condition removeConditoin = lock.newCondition(); 7 8 public BoundedQueue1(int size) { 9 q = new ArrayList<>(size); 10 maxSize = size; 11 } 12 13 public void add(T e) { 14 lock.lock(); 15 try { 16 while (q.size() == maxSize) { 17 addConditoin.await(); 18 } 19 q.add(e); 20 removeConditoin.signal(); //执行了添加操作后唤醒因队列空被阻塞的删除操作 21 } catch (InterruptedException e1) { 22 Thread.currentThread().interrupt(); 23 } finally { 24 lock.unlock(); 25 } 26 } 27 28 public T remove() { 29 lock.lock(); 30 try { 31 while (q.size() == 0) { 32 removeConditoin.await(); 33 } 34 T e = q.remove(0); 35 addConditoin.signal(); //执行删除操作后唤醒因队列满而被阻塞的添加操作 36 return e; 37 } catch (InterruptedException e) { 38 Thread.currentThread().interrupt(); 39 return null; 40 } finally { 41 lock.unlock(); 42 } 43 } 44 45}
源码分析
下面来分析Condition源码
概述
之前我们介绍AQS的时候说过，AQS的同步排队用了一个隐式的双向队列，同步队列的每个节点是一个AbstractQueuedSynchronizer.Node实例。
Node的主要字段有：
- waitStatus：等待状态，所有的状态见下面的表格。
- prev：前驱节点
- next：后继节点
- thread：当前节点代表的线程
- nextWaiter：Node既可以作为同步队列节点使用，也可以作为Condition的等待队列节点使用(将会在后面讲Condition时讲到)。在作为同步队列节点时，nextWaiter可能有两个值：EXCLUSIVE、SHARED标识当前节点是独占模式还是共享模式；在作为等待队列节点使用时，nextWaiter保存后继节点。
状态
值
含义
CANCELLED
1
当前节点因为超时或中断被取消同步状态获取，该节点进入该状态不会再变化
SIGNAL
-1
标识后继的节点处于阻塞状态，当前节点在释放同步状态或被取消时，需要通知后继节点继续运行。每个节点在阻塞前，需要标记其前驱节点的状态为SIGNAL。
CONDITION
-2
标识当前节点是作为等待队列节点使用的。
PROPAGATE
-3
0
0
初始状态
Condition实现等待的时候内部也有一个等待队列，等待队列是一个隐式的单向队列，等待队列中的每一个节点也是一个AbstractQueuedSynchronizer.Node实例。
每个Condition对象中保存了firstWaiter和lastWaiter作为队列首节点和尾节点，每个节点使用Node.nextWaiter保存下一个节点的引用，因此等待队列是一个单向队列。
每当一个线程调用Condition.await()方法，那么该线程会释放锁，构造成一个Node节点加入到等待队列的队尾。
等待
Condition.await()方法的源码如下：
1public final void await() throws InterruptedException { 2 if (Thread.interrupted()) 3 throw new InterruptedException(); 4 Node node = addConditionWaiter(); //构造一个新的等待队列Node加入到队尾 5 int savedState = fullyRelease(node); //释放当前线程的独占锁，不管重入几次，都把state释放为0 6 int interruptMode = 0; //如果当前节点没有在同步队列上，即还没有被signal，则将当前线程阻塞 7 while (!isOnSyncQueue(node)) { 8 LockSupport.park(this); //后面的蓝色代码都是和中断相关的，主要是区分两种中断：是在被signal前中断还是在被signal后中断，如果是被signal前就被中断则抛出 InterruptedException，否则执行 Thread.currentThread().interrupt(); 9 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) //被中断则直接退出自旋 10 break; 11 } //退出了上面自旋说明当前节点已经在同步队列上，但是当前节点不一定在同步队列队首。acquireQueued将阻塞直到当前节点成为队首，即当前线程获得了锁。然后await()方法就可以退出了，让线程继续执行await()后的代码。 12 if (acquireQueued(node, savedState) && interruptMode != THROW_IE) 13 interruptMode = REINTERRUPT; 14 if (node.nextWaiter != null) // clean up if cancelled 15 unlinkCancelledWaiters(); 16 if (interruptMode != 0) 17 reportInterruptAfterWait(interruptMode); 18 } final int fullyRelease(Node node) { 19 boolean failed = true; 20 try { 21 int savedState = getState(); 22 if (release(savedState)) { 23 failed = false; 24 return savedState; 25 } else { 26 throw new IllegalMonitorStateException(); 27 } 28 } finally { 29 if (failed) 30 node.waitStatus = Node.CANCELLED; 31 } 32 } 33 34 final boolean isOnSyncQueue(Node node) { //如果当前节点状态是CONDITION或node.prev是null，则证明当前节点在等待队列上而不是同步队列上。之所以可以用node.prev来判断，是因为一个节点如果要加入同步队列，在加入前就会设置好prev字段。 35 if (node.waitStatus == Node.CONDITION || node.prev == null) 36 return false; //如果node.next不为null，则一定在同步队列上，因为node.next是在节点加入同步队列后设置的 37 if (node.next != null) // If has successor, it must be on queue 38 return true; 39 return findNodeFromTail(node); //前面的两个判断没有返回的话，就从同步队列队尾遍历一个一个看是不是当前节点。 40 } 41 42 private boolean findNodeFromTail(Node node) { 43 Node t = tail; 44 for (;;) { 45 if (t == node) 46 return true; 47 if (t == null) 48 return false; 49 t = t.prev; 50 } 51 }
1final boolean acquireQueued(final Node node, int arg) { 2 boolean failed = true; 3 try { 4 boolean interrupted = false; 5 for (;;) { 6 final Node p = node.predecessor(); 7 if (p == head && tryAcquire(arg)) { 8 setHead(node); 9 p.next = null; // help GC 10 failed = false; 11 return interrupted; 12 } 13 if (shouldParkAfterFailedAcquire(p, node) && 14 parkAndCheckInterrupt()) 15 interrupted = true; 16 } 17 } finally { 18 if (failed) 19 cancelAcquire(node); 20 } 21 }
通知
Condition.signal() 方法的源码如下：
1public final void signal() { 2 if (!isHeldExclusively()) 3 throw new IllegalMonitorStateException(); //如果同步状态不是被当前线程独占，直接抛出异常。从这里也能看出来，Condition只能配合独占类同步组件使用。 4 Node first = firstWaiter; 5 if (first != null) 6 doSignal(first); //通知等待队列队首的节点。 7 }
1private void doSignal(Node first) { 2 do { 3 if ( (firstWaiter = first.nextWaiter) == null) 4 lastWaiter = null; 5 first.nextWaiter = null; 6 } while (!transferForSignal(first) && //transferForSignal方法尝试唤醒当前节点，如果唤醒失败，则继续尝试唤醒当前节点的后继节点。 7 (first = firstWaiter) != null); 8 } 9 10 final boolean transferForSignal(Node node) { 11 //如果当前节点状态为CONDITION，则将状态改为0准备加入同步队列；如果当前状态不为CONDITION，说明该节点等待已被中断，则该方法返回false，doSignal()方法会继续尝试唤醒当前节点的后继节点 12 if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) 13 return false; 14 15 /* 16 * Splice onto queue and try to set waitStatus of predecessor to 17 * indicate that thread is (probably) waiting. If cancelled or 18 * attempt to set waitStatus fails, wake up to resync (in which 19 * case the waitStatus can be transiently and harmlessly wrong). 20 */ 21 Node p = enq(node); //将节点加入同步队列，返回的p是节点在同步队列中的先驱节点 22 int ws = p.waitStatus; //如果先驱节点的状态为CANCELLED(>0) 或设置先驱节点的状态为SIGNAL失败，那么就立即唤醒当前节点对应的线程，线程被唤醒后会执行acquireQueued方法，该方法会重新尝试将节点的先驱状态设为SIGNAL并再次park线程；如果当前设置前驱节点状态为SIGNAL成功，那么就不需要马上唤醒线程了，当它的前驱节点成为同步队列的首节点且释放同步状态后，会自动唤醒它。 //其实笔者认为这里不加这个判断条件应该也是可以的。只是对于CAS修改前驱节点状态为SIGNAL成功这种情况来说，如果不加这个判断条件，提前唤醒了线程，等进入acquireQueued方法了节点发现自己的前驱不是首节点，还要再阻塞，等到其前驱节点成为首节点并释放锁时再唤醒一次；而如果加了这个条件，线程被唤醒的时候它的前驱节点肯定是首节点了，线程就有机会直接获取同步状态从而避免二次阻塞，节省了硬件资源。 23 if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) 24 LockSupport.unpark(node.thread); 25 return true; 26 }
Condition等待通知的本质
总的来说，Condition的本质就是等待队列和同步队列的交互：
当一个持有锁的线程调用Condition.await()时，它会执行以下步骤：
- 构造一个新的等待队列节点加入到等待队列队尾
- 释放锁，也就是将它的同步队列节点从同步队列队首移除
- 自旋，直到它在等待队列上的节点移动到了同步队列（通过其他线程调用signal()）或被中断
- 阻塞当前节点，直到它获取到了锁，也就是它在同步队列上的节点排队排到了队首。
当一个持有锁的线程调用Condition.signal()时，它会执行以下操作：
从等待队列的队首开始，尝试对队首节点执行唤醒操作；如果节点CANCELLED，就尝试唤醒下一个节点；如果再CANCELLED则继续迭代。
对每个节点执行唤醒操作时，首先将节点加入同步队列，此时await()操作的步骤3的解锁条件就已经开启了。然后分两种情况讨论：
- 如果先驱节点的状态为CANCELLED(>0) 或设置先驱节点的状态为SIGNAL失败，那么就立即唤醒当前节点对应的线程，此时await()方法就会完成步骤3，进入步骤4.
- 如果成功把先驱节点的状态设置为了SIGNAL，那么就不立即唤醒了。等到先驱节点成为同步队列首节点并释放了同步状态后，会自动唤醒当前节点对应线程的，这时候await()的步骤3才执行完成，而且有很大概率快速完成步骤4.
总结
如果知道Object的等待通知机制，Condition的使用是比较容易掌握的，因为和Object等待通知的使用基本一致。
对Condition的源码理解，主要就是理解等待队列，等待队列可以类比同步队列，而且等待队列比同步队列要简单，因为等待队列是单向队列，同步队列是双向队列。
以下是笔者对等待队列是单向队列、同步队列是双向队列的一些思考，欢迎提出不同意见：
之所以同步队列要设计成双向的，是因为在同步队列中，节点唤醒是接力式的，由每一个节点唤醒它的下一个节点，如果是由next指针获取下一个节点，是有可能获取失败的，因为虚拟队列每添加一个节点，是先用CAS把tail设置为新节点，然后才修改原tail的next指针到新节点的。因此用next向后遍历是不安全的，但是如果在设置新节点为tail前，为新节点设置prev，则可以保证从tail往前遍历是安全的。因此要安全的获取一个节点Node的下一个节点，先要看next是不是null，如果是null，还要从tail往前遍历看看能不能遍历到Node。
而等待队列就简单多了，等待的线程就是等待者，只负责等待，唤醒的线程就是唤醒者，只负责唤醒，因此每次要执行唤醒操作的时候，直接唤醒等待队列的首节点就行了。等待队列的实现中不需要遍历队列，因此也不需要prev指针。
推荐一篇文章，感觉分析的也挺好的：
https://blog.csdn.net/yanyan19880509/article/details/52502882