Java并发Condition原理分析
一.介绍
1.1 Condition用途
当多个线程需要访问一个共享资源时，需要给共享资源加锁。 当一个线程释放锁时，所有等待锁的线程都会尝试去获取锁。 但是如果想只让部分等待锁的线程去获取锁时，就需要用到Condition。
1.2 整体分析
Condition具体实现在AbstractQueuedSynchronizer类中。这个类中管理了一个阻塞队列和N多个条件队列。
阻塞队列记录了等待获取锁的线程，头结点记录了当前正在运行的线程。
条件队列记录了由Condition.await()阻塞的线程，一个Lock可以有多个Condition,每个Condition是一个队列。
Condition是AbstractQueuedSynchronizer的一个内部类ConditionObject,所以创建的Condition对象是可以访问整个AbstractQueuedSynchronizer对象的属性的，通过这样将Condition与Lock相关联。
1.3 Condition原理介绍
Condition是AQS的内部类。每个Condition对象都包含一个条件队列 (等待队列)。
等待队列是一个FIFO的队列，在队列中的每个节点Node都包含了一个线程引用，该线程就是在Condition对象上等待的线程。
如果一个线程调用了Condition.await()方法，那么该线程将会释放锁、构造成节点加入等待队列并进入等待状态。
等待队列的基本结构如下所示。
等待分为首节点和尾节点。当一个线程调用Condition.await()方法，将会以当前线程构造节点，并将节点从尾部加入等待队列。新增节点就是将尾部节点指向新增的节点。节点引用更新本来就是在获取锁以后的操作，所以不需要CAS保证。同时也是线程安全的操作。
signal()将条件队列中的第一个节点加入到阻塞队列的尾端，表示可以被唤醒执行。
二.Condition类
2.1 属性
public class ConditionObject implements Condition, java.io.Serializable {
private static final long serialVersionUID = 1173984872572414699L;
/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter;
//省略部分...
}
2.2 await()释放锁阻塞线程
造成当前线程在接到信号或被中断之前一直处于等待状态。
await()方法释放锁，并将当前线程封装成一个node加入到条件队列的尾部。同时阻塞当前线程，等待唤醒。在调用await()方法时，当前线程必须是获取到了锁的状态，否则会抛出异常IllegalMonitorStateException。
public final void await() throws InterruptedException {
if (Thread.interrupted()) // 当前线程如果中断抛出异常
throw new InterruptedException();
// 把当前线程封装成 Node，添加到等待队列队尾。这里的队列是单向链表;
Node node = addConditionWaiter();
int savedState = fullyRelease(node); // 释放锁,调用await之前当前线程是占有锁的
int interruptMode = 0;
// isOnSyncQueue表示线程是否在阻塞队列中，第一次判断为false,因为线程已经释放了
// 释放完毕后,遍历AQS队列,看当前节点是否在同步队列中,
//如果不在,说明它还没有竞争锁的资格,继续将自己沉睡 ;直到被加入到队列(singal)
while (!isOnSyncQueue(node)) {
LockSupport.park(this); // 继续沉睡
//线程被挂起,阻塞在这个位置
// 判断线程被打断后,退出循环。
// 该方法的返回值代表当前线程是否在 park 的时候被中断唤醒
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
//被唤醒后,重新开始正式竞争锁,如果竞争不到还是会将自己沉睡,等待唤醒重新开始竞争
//acquireQueued: 该方法的功能是循环的尝试获取锁, 如果返回false就表示拿到锁了
// interruptMode !=THROW_IE 表示这个线程没有成功将node入队，但signal执行了enq方法让其入队了 ,将这个变量设置成 REINTERRUPT
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
// 如果 node 的下一个等待者不是 null, 则进行清理,清理 Condition 队列上的节点
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
线程(获取到锁后)调用await方法,后从AQS队列中移除出,并构成新节点加入Condition等待队列中,如下图所示;
2.3 signal()唤醒线程
其实正常情况下，signal()并不会执行LockSupport.unpark来唤醒等待的线程，而是将线程节点从条件队列 (队头head) 移动到了阻塞队列 (队尾tail))。那么当运行signal()的线程释放了锁以后，那么调用await()的线程LockSupport.park阻塞的，就可以被唤醒。
public final void signal() {
//当前线程是否获取独占锁，isHeldExclusively由子类实现
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter; // 找到Condition队列第一个节点,唤醒
if (first != null)
doSignal(first);
}
// 删除和传输节点，直到碰到非取消的1或null。从上面方法分离出来是因为等待节点可能为null
private void doSignal(Node first) {
do {
// 修改头结点，移出旧头节点, 断掉节点与条件队列的联系,
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) && //将老的头节点加入到AQS的等待队列中(
(first = firstWaiter) != null);
}
/** 将一个节点从条件队列转移到同步队列。如果成功转移，返回true(否则节点是signal前取消) */
final boolean transferForSignal(Node node) {
// 如果不能修改waitStatus，表示该节点已经取消。
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
// enq(node): 自旋进入阻塞队列的队尾
// 注意，这里的返回值 p 是 node 在阻塞队列的前驱节点
Node p = enq(node); //将节点插入队列，必要时进行初始化。
int ws = p.waitStatus;
//如果该结点的状态为cancel 或者修改waitStatus失败，则直接唤醒。
// 通常这里的ws是小于0的，compareAndSetWaitStatus是会返回true的，
// 如果上一个节点的状态被取消了, 或者尝试设置上一个节点的状态为SIGNAL失败了(SIGNAL 表示: 他的 next节点需要停止阻塞),
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread); // 唤醒节点内封装的线程
return true;
}
可以看到，正常情况 ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL) 这个判断是不会为true的，所以，不会在这个时候唤醒该线程。 只有到发送signal信号的线程调用reentrantLock.unlock()后因为它已经被加到AQS的等待队列中，所以才会被唤醒。
在调用signal()方法之前必须先判断是否获取到了锁。接着获取等待队列的首节点，将其移动到同步队列并且利用LockSupport唤醒节点中的线程。节点从等待队列移动到同步队列如下图所示:
2.4 实例说明流程(重要)
public static void main(String[] args) {
Thread thread1 = new Thread(() -> {
lock.lock();
try {
System.out.println("线程1 start...");
boolean r = condition.await(5, TimeUnit.SECONDS);
System.out.println("线程1 await结果" + r);
System.out.println("线程1 end...");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}, "线程1");
thread1.start();
Thread thread2 = new Thread(() -> {
lock.lock();
try {
System.out.println("线程2 start...");
Thread.sleep(3000);
condition.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}, "线程2");
thread2.start();
}
AQS自己维护的队列是当前等待资源的队列，AQS会在资源被释放后，依次唤醒队列中从前到后的所有节点，使他们对应的线程恢复执行。直到队列为空。
而Condition自己也维护了一个队列，该队列的作用是维护一个等待signal信号的队列，两个队列的作用是不同，事实上，每个线程也仅仅会同时存在以上两个队列中的一个。
上面实例流程(重要)：
1. 线程1调用reentrantLock.lock时，线程被加入到AQS的等待队列中。
2. 线程1调用await方法被调用时，该线程从AQS中移除，对应操作是锁的释放。
3. 接着马上被加入到Condition的等待队列中，以为着该线程需要signal信号。
4. 线程2，因为线程1释放锁的关系，被唤醒，并判断可以获取锁，于是线程2获取锁，并被加入到AQS的等待队列中。
5. 线程2调用signal方法，这个时候Condition的等待队列中只有线程1一个节点，于是它被取出来，并被加入到AQS的等待队列中。 注意，这个时候，线程1 并没有被唤醒。
6. signal方法执行完毕，线程2调用reentrantLock.unLock()方法，释放锁。这个时候因为AQS中只有线程1，于是，AQS释放锁后按从头到尾的顺序唤醒线程时，线程1被唤醒，于是线程1回复执行。
7. 直到释放锁整个过程执行完毕。
可以看到，整个协作过程是靠结点在AQS的等待队列和Condition的等待队列中来回移动实现的，Condition作为一个条件类，很好的自己维护了一个等待信号的队列，并在适时的时候将结点加入到AQS的等待队列中来实现的唤醒操作。
三.Condition 总结
线程 awaitThread 先通过 lock.lock()方法获取锁成功后调用了 condition.await 方法进入等待队列，
而另一个线程 signalThread 通过lock.lock()方法获取锁成功后调用了 condition.signal 或者 signalAll 方法，使得线程awaitThread 能够有机会移入到同步队列中，
当其他线程释放 lock 后使得线程 awaitThread 能够有机会获取lock，从而使得线程 awaitThread 能够从 await 方法中退出执行后续操作。
如果 awaitThread 获取 lock 失败会直接进入到同步队列。
参考:
https://www.cnblogs.com/gemine/p/9039012.html
https://blog.csdn.net/yyzzhc999/article/details/96917878
https://blog.csdn.net/weixin_44366439/article/details/86540308
http://www.voidcn.com/article/p-pdrhqbfl-sc.html