文章部分代码和照片来自参考资料
问题 :
- ConditionObject 的 await 和 signal 方法是如何实现的
ConditonObject
ConditionObject 继承 Condition 这个接口， 看一下这个接口的注解说明 ：
Condition factors out the Object monitor methods (wait, notify and notifyAll) into distinct objects to give the effect of having multiple wait-sets per object, by combining them with the use of arbitrary Lock implementations. Where a Lock replaces the use of synchronized methods and statements, a Condition replaces the use of the Object monitor methods.
如果Lock替换了synchronized方法和语句的使用，则Condition将替换Object监视方法的使用。
Condition 经常可以用在生产者-消费者的场景中,ArrayBlockingQueue 采用这种方式实现了生产者-消费者.
ConditionObject 是 AQS里的一个对象，继承Condition 接口，上一节我们提到AQS 通过同步队列（sync queue ）和 等待队列（wait queue ）还有 状态变量（statue）进行并发控制。这节我们要讲的就是在等待队列的操作。
下面是 wait queue 和 sync queue 的图例。
await 和 signal 方法
这两个方法可以用下面两种两张图来描述。其中await 是释放所有的锁，然后将节点加入到 wait queue ,然后等待唤醒。 signal 方法是从wait queue 移动到 sync queue 中，然后唤醒。
图一. await 方法
图二. signal 方法
Conditon 的方法实现 基于 ReetranLock 。下面源码分析会涉及到。
Condition 的await 方法 包括的操作有 :
- If current thread is interrupted, throw InterruptedException.
- Save lock state returned by getState.
- Invoke release with saved state as argument, throwing IllegalMonitorStateException if it fails.
- Block until signalled or interrupted.
- Reacquire by invoking specialized version of acquire with saved state as argument.
If interrupted while blocked in step 4, throw InterruptedException.
1 // 首先，这个方法是可被中断的，不可被中断的是另一个方法 awaitUninterruptibly() 2 // 这个方法会阻塞，直到调用 signal 方法（指 signal() 和 signalAll()，下同），或被中断 3 public final void await() throws InterruptedException { 4 if (Thread.interrupted()) 5 throw new InterruptedException(); 6 // 添加到 condition 的条件队列中 7 Node node = addConditionWaiter(); 8 // 释放锁，返回值是释放锁之前的 state 值 9 int savedState = fullyRelease(node); 10 int interruptMode = 0; 11 // 这里退出循环有两种情况，之后再仔细分析 12 // 1. isOnSyncQueue(node) 返回 true，即当前 node 已经转移到阻塞队列了 13 // 2. checkInterruptWhileWaiting(node) != 0 会到 break，然后退出循环，代表的是线程中断 14 while (!isOnSyncQueue(node)) { 15 LockSupport.park(this); 16 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) 17 break; 18 } 19 // 被唤醒后，抢锁，抢不到将进入阻塞队列，等待获取锁 20 if (acquireQueued(node, savedState) && interruptMode != THROW_IE) 21 interruptMode = REINTERRUPT; 22 if (node.nextWaiter != null) // clean up if cancelled 23 unlinkCancelledWaiters(); 24 if (interruptMode != 0) 25 reportInterruptAfterWait(interruptMode); 26 }
1 // 将当前线程对应的节点入队，插入队尾 2 private Node addConditionWaiter() { 3 Node t = lastWaiter; 4 // 如果条件队列的最后一个节点取消了，将其清除出去 5 if (t != null && t.waitStatus != Node.CONDITION) { 6 // 这个方法会遍历整个条件队列，然后会将已取消的所有节点清除出队列 7 unlinkCancelledWaiters(); 8 t = lastWaiter; 9 } 10 Node node = new Node(Thread.currentThread(), Node.CONDITION); 11 // 如果队列为空 12 if (t == null) 13 firstWaiter = node; 14 else 15 t.nextWaiter = node; 16 lastWaiter = node; 17 return node; 18 } 19 在addWaiter 方法中，有一个 unlinkCancelledWaiters() 方法，该方法用于清除队列中已经取消等待的节点。 20 21 当 await 的时候如果发生了取消操作（这点之后会说），或者是在节点入队的时候，发现最后一个节点是被取消的，会调用一次这个方法。 22 23 // 等待队列是一个单向链表，遍历链表将已经取消等待的节点清除出去 24 // 纯属链表操作，很好理解，看不懂多看几遍就可以了 25 private void unlinkCancelledWaiters() { 26 Node t = firstWaiter; 27 Node trail = null; 28 while (t != null) { 29 Node next = t.nextWaiter; 30 // 如果节点的状态不是 Node.CONDITION 的话，这个节点就是被取消的 31 if (t.waitStatus != Node.CONDITION) { 32 t.nextWaiter = null; 33 if (trail == null) 34 firstWaiter = next; 35 else 36 trail.nextWaiter = next; 37 if (next == null) 38 lastWaiter = trail; 39 } 40 else 41 trail = t; 42 t = next; 43 } 44 }
ReentranLock 是可重入的，所以释放所有的锁。
1 // 首先，我们要先观察到返回值 savedState 代表 release 之前的 state 值 2 // 对于最简单的操作：先 lock.lock()，然后 condition1.await()。 3 // 那么 state 经过这个方法由 1 变为 0，锁释放，此方法返回 1 4 // 相应的，如果 lock 重入了 n 次，savedState == n 5 // 如果这个方法失败，会将节点设置为"取消"状态，并抛出异常 IllegalMonitorStateException 6 final int fullyRelease(Node node) { 7 boolean failed = true; 8 try { 9 int savedState = getState(); 10 // 这里使用了当前的 state 作为 release 的参数，也就是完全释放掉锁，将 state 置为 0 11 if (release(savedState)) { 12 failed = false; 13 return savedState; 14 } else { 15 throw new IllegalMonitorStateException(); 16 } 17 } finally { 18 if (failed) 19 node.waitStatus = Node.CANCELLED; 20 } 21 }
1 // 在节点入条件队列的时候，初始化时设置了 waitStatus = Node.CONDITION 2 // 前面我提到，signal 的时候需要将节点从条件队列移到阻塞队列， 3 // 这个方法就是判断 node 是否已经移动到阻塞队列了 4 final boolean isOnSyncQueue(Node node) { 5 // 移动过去的时候，node 的 waitStatus 会置为 0，这个之后在说 signal 方法的时候会说到 6 // 如果 waitStatus 还是 Node.CONDITION，也就是 -2，那肯定就是还在条件队列中 7 // 如果 node 的前驱 prev 指向还是 null，说明肯定没有在 阻塞队列 8 if (node.waitStatus == Node.CONDITION || node.prev == null) 9 return false; 10 // 如果 node 已经有后继节点 next 的时候，那肯定是在阻塞队列了 11 if (node.next != null) 12 return true; 13 14 // 这个方法从阻塞队列的队尾开始从后往前遍历找，如果找到相等的，说明在阻塞队列，否则就是不在阻塞队列 15 16 // 可以通过判断 node.prev() != null 来推断出 node 在阻塞队列吗？答案是：不能。 17 // 这个可以看上篇 AQS 的入队方法，首先设置的是 node.prev 指向 tail， 18 // 然后是 CAS 操作将自己设置为新的 tail，可是这次的 CAS 是可能失败的。 19 20 // 调用这个方法的时候，往往我们需要的就在队尾的部分，所以一般都不需要完全遍历整个队列的 21 return findNodeFromTail(node); 22 } 23 24 // 从同步队列的队尾往前遍历，如果找到，返回 true 25 private boolean findNodeFromTail(Node node) { 26 Node t = tail; 27 for (;;) { 28 if (t == node) 29 return true; 30 if (t == null) 31 return false; 32 t = t.prev; 33 } 34 } 35 }
1 int interruptMode = 0; 2 while (!isOnSyncQueue(node)) { 3 // 线程挂起 4 LockSupport.park(this); 5 6 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) 7 break; 8 }
回到前面的循环，isOnSyncQueue(node) 返回 false 的话，那么进到 LockSupport.park(this); 这里线程挂起。
接下来就是 signal 移动元素到同步队列，释放锁的时候唤醒线程，转移到阻塞队列为了大家理解，这里我们先看唤醒操作，因为刚刚到 LockSupport.park(this); 把线程挂起了，等待唤醒。唤醒操作通常由另一个线程来操作，就像生产者-消费者模式中，如果线程因为等待消费而挂起，那么当生产者生产了一个东西后，会调用 signal 移动元素到同步队列，释放锁的时候唤醒线程来消费。
1 2 // 唤醒等待了最久的线程 3 // 其实就是，将这个线程对应的 node 从条件队列转移到阻塞队列 4 public final void signal() { 5 // 调用 signal 方法的线程必须持有当前的独占锁 6 if (!isHeldExclusively()) 7 throw new IllegalMonitorStateException(); 8 Node first = firstWaiter; 9 if (first != null) 10 doSignal(first); 11 } 12 13 // 从条件队列队头往后遍历，找出第一个需要转移的 node 14 // 因为前面我们说过，有些线程会取消排队，但是还在队列中 15 private void doSignal(Node first) { 16 do { 17 // 将 firstWaiter 指向 first 节点后面的第一个 18 // 如果将队头移除后，后面没有节点在等待了，那么需要将 lastWaiter 置为 null 19 if ( (firstWaiter = first.nextWaiter) == null) 20 lastWaiter = null; 21 // 因为 first 马上要被移到阻塞队列了，和条件队列的链接关系在这里断掉 22 first.nextWaiter = null; 23 } while (!transferForSignal(first) && 24 (first = firstWaiter) != null); 25 // 这里 while 循环，如果 first 转移不成功，那么选择 first 后面的第一个节点进行转移，依此类推 26 } 27 28 // 将节点从条件队列转移到阻塞队列 29 // true 代表成功转移 30 // false 代表在 signal 之前，节点已经取消了 31 final boolean transferForSignal(Node node) { 32 33 // CAS 如果失败，说明此 node 的 waitStatus 已不是 Node.CONDITION，说明节点已经取消， 34 // 既然已经取消，也就不需要转移了，方法返回，转移后面一个节点 35 // 否则，将 waitStatus 置为 0 36 if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) 37 return false; 38 39 // enq(node): 自旋进入阻塞队列的队尾 40 // 注意，这里的返回值 p 是 node 在阻塞队列的前驱节点 41 Node p = enq(node); 42 int ws = p.waitStatus; 43 // ws > 0 说明 node 在阻塞队列中的前驱节点取消了等待锁，直接唤醒 node 对应的线程。唤醒之后会怎么样，后面再解释 44 // 如果 ws <= 0, 那么 compareAndSetWaitStatus 将会被调用，上篇介绍的时候说过，节点入队后，需要把前驱节点的状态设为 Node.SIGNAL(-1) 45 if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) 46 // 如果前驱节点取消或者 CAS 失败，会进到这里唤醒线程，之后的操作看下一节 47 LockSupport.unpark(node.thread); 48 return true; 49
注意 ：
- 调用 signal 前要先获得锁
- 正常情况下，signal 方法只是对元素进行了移动（从wait queue 到 sync queue），真正的唤醒操作在释放锁的代码里。
那么阻塞的线程现在继续执行，我们可以看到，即使被唤醒依旧要继续抢。
1 /** 2 * Acquires in exclusive uninterruptible mode for thread already in 3 * queue. Used by condition wait methods as well as acquire. 4 * 5 * @param node the node 6 * @param arg the acquire argument 7 * @return {@code true} if interrupted while waiting 8 */ 9 final boolean acquireQueued(final Node node, int arg) { 10 boolean failed = true; 11 try { 12 boolean interrupted = false; 13 for (;;) { 14 final Node p = node.predecessor(); 15 if (p == head && tryAcquire(arg)) { 16 setHead(node); 17 p.next = null; // help GC 18 failed = false; 19 return interrupted; 20 } 21 if (shouldParkAfterFailedAcquire(p, node) && 22 parkAndCheckInterrupt()) 23 interrupted = true; 24 } 25 } finally { 26 if (failed) 27 cancelAcquire(node); 28 } 29 }
注意 ：
- await 方法被唤醒后依旧要抢，要是抢不到就继续阻塞。
要是CAS 成功，检查中断状态
1 int interruptMode = 0; 2 while (!isOnSyncQueue(node)) { 3 // 线程挂起 4 LockSupport.park(this); 5 6 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) 7 break; 8 }
1 // 1. 如果在 signal 之前已经中断，返回 THROW_IE 2 // 2. 如果是 signal 之后中断，返回 REINTERRUPT 3 // 3. 没有发生中断，返回 0 4 private int checkInterruptWhileWaiting(Node node) { 5 return Thread.interrupted() ? 6 (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 7 0; 8 }
1 // 只有线程处于中断状态，才会调用此方法 2 // 如果需要的话，将这个已经取消等待的节点转移到阻塞队列 3 // 返回 true：如果此线程在 signal 之前被取消， 4 final boolean transferAfterCancelledWait(Node node) { 5 // 用 CAS 将节点状态设置为 0 6 // 如果这步 CAS 成功，说明是 signal 方法之前发生的中断，因为如果 signal 先发生的话，signal 中会将 waitStatus 设置为 0 7 if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) { 8 // 将节点放入阻塞队列 9 // 这里我们看到，即使中断了，依然会转移到阻塞队列 10 enq(node); 11 return true; 12 } 13 14 // 到这里是因为 CAS 失败，肯定是因为 signal 方法已经将 waitStatus 设置为了 0 15 // signal 方法会将节点转移到阻塞队列，但是可能还没完成，这边自旋等待其完成 16 // 当然，这种事情还是比较少的吧：signal 调用之后，没完成转移之前，发生了中断 17 while (!isOnSyncQueue(node)) 18 Thread.yield(); 19 return false; 20 }
小提示
在AQS 的子类实现中，Conditon 常常会配合 ReentranLock 使用，例如 LinkBlockingQueue 。 要注意的是ConditionObject的 await 方法会释放掉所有的锁。show me code !
1 public class LinkBlockingQueueTest { 2 private ReentrantLock mLock = new ReentrantLock(); 3 private Condition mConditionObject = mLock.newCondition(); 4 5 public void testMethod() throws InterruptedException { 6 7 final ReentrantLock methodLock = this.mLock; 8 methodLock.lockInterruptibly(); 9 try { 10 mConditionObject.await(); 11 System.out.println("sss"); 12 } finally { 13 methodLock.unlock(); 14 } 15 } 16 17 public void testMethod1() throws InterruptedException { 18 19 final ReentrantLock methodLock = this.mLock; 20 methodLock.lockInterruptibly(); 21 try { 22 System.out.println("s1"); 23 } finally { 24 methodLock.unlock(); 25 } 26 } 27 28 }
1 public static void main(String[] args) throws InterruptedException { 2 3 LinkBlockingQueueTest obj = new LinkBlockingQueueTest(); 4 5 new Thread(() -> { 6 try { 7 obj.testMethod(); 8 } catch (InterruptedException e) { 9 e.printStackTrace(); 10 } 11 }).start(); 12 13 Thread.sleep(1000L * 2); 14 new Thread(() -> { 15 try { 16 obj.testMethod1(); 17 } catch (InterruptedException e) { 18 e.printStackTrace(); 19 } 20 }).start(); 21 22 }
会输出 s1 。
为什么有了sync queue 还需要 wait queue ?
ConditionObject 里的await方法会（假如这个节点已经存在sync queue）释放锁移入 wait queue , signal 方法则是重新移入到 sync queue ，那么我们就可以知道了 wait queue 的作用是临时存放节点，移除在 sync queue 的节点（假如存在），再插入到 sync queue 的队尾。它的作用我们可以在阅读ArrayBlockingQueue源码时就可以感受到了。