Java多线程与高并发专题——关于Condition

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.
Conditions (also known as condition queues or condition variables) provide a means for one thread to suspend execution (to "wait") until notified by another thread that some state condition may now be true. Because access to this shared state information occurs in different threads, it must be protected, so a lock of some form is associated with the condition. The key property that waiting for a condition provides is that it atomically releases the associated lock and suspends the current thread, just like Object.wait.
A Condition instance is intrinsically bound to a lock. To obtain a Condition instance for a particular Lock instance use its newCondition() method.
As an example, suppose we have a bounded buffer which supports put and take methods. If a take is attempted on an empty buffer, then the thread will block until an item becomes available; if a put is attempted on a full buffer, then the thread will block until a space becomes available. We would like to keep waiting put threads and take threads in separate wait-sets so that we can use the optimization of only notifying a single thread at a time when items or spaces become available in the buffer. This can be achieved using two Condition instances.
class BoundedBuffer {final Lock lock = new ReentrantLock();final Condition notFull  = lock.newCondition();final Condition notEmpty = lock.newCondition();final Object[] items = new Object[100];int putptr, takeptr, count;public void put(Object x) throws InterruptedException {lock.lock();try {while (count == items.length)notFull.await();items[putptr] = x;if (++putptr == items.length) putptr = 0;++count;notEmpty.signal();} finally {lock.unlock();}}public Object take() throws InterruptedException {lock.lock();try {while (count == 0)notEmpty.await();Object x = items[takeptr];if (++takeptr == items.length) takeptr = 0;--count;notFull.signal();return x;} finally {lock.unlock();}}
}
(The java.util.concurrent.ArrayBlockingQueue class provides this functionality, so there is no reason to implement this sample usage class.)
A Condition implementation can provide behavior and semantics that is different from that of the Object monitor methods, such as guaranteed ordering for notifications, or not requiring a lock to be held when performing notifications. If an implementation provides such specialized semantics then the implementation must document those semantics.
Note that Condition instances are just normal objects and can themselves be used as the target in a synchronized statement, and can have their own monitor wait and notification methods invoked. Acquiring the monitor lock of a Condition instance, or using its monitor methods, has no specified relationship with acquiring the Lock associated with that Condition or the use of its waiting and signalling methods. It is recommended that to avoid confusion you never use Condition instances in this way, except perhaps within their own implementation.
Except where noted, passing a null value for any parameter will result in a NullPointerException being thrown.
Implementation Considerations
When waiting upon a Condition, a "spurious wakeup" is permitted to occur, in general, as a concession to the underlying platform semantics. This has little practical impact on most application programs as a Condition should always be waited upon in a loop, testing the state predicate that is being waited for. An implementation is free to remove the possibility of spurious wakeups but it is recommended that applications programmers always assume that they can occur and so always wait in a loop.
The three forms of condition waiting (interruptible, non-interruptible, and timed) may differ in their ease of implementation on some platforms and in their performance characteristics. In particular, it may be difficult to provide these features and maintain specific semantics such as ordering guarantees. Further, the ability to interrupt the actual suspension of the thread may not always be feasible to implement on all platforms.
Consequently, an implementation is not required to define exactly the same guarantees or semantics for all three forms of waiting, nor is it required to support interruption of the actual suspension of the thread.
An implementation is required to clearly document the semantics and guarantees provided by each of the waiting methods, and when an implementation does support interruption of thread suspension then it must obey the interruption semantics as defined in this interface.
As interruption generally implies cancellation, and checks for interruption are often infrequent, an implementation can favor responding to an interrupt over normal method return. This is true even if it can be shown that the interrupt occurred after another action that may have unblocked the thread. An implementation should document this behavior.

翻译：

Condition 将 Object 的监视器方法（wait、notify 和 notifyAll）分离成不同的对象，通过与任意 Lock 实现结合使用，实现每个对象有多个等待集的效果。当 Lock 替代了 synchronized 方法和语句的使用时，Condition 就替代了 Object 监视器方法的使用。
条件（也称为条件队列或条件变量）为一个线程提供了一种暂停执行（即 “等待”）的方式，直到另一个线程通知某些状态条件现在可能为真。因为对这种共享状态信息的访问发生在不同线程中，所以必须加以保护，因此某种形式的锁与该条件相关联。等待条件所具备的关键特性是，它会原子性地释放相关联的锁并挂起当前线程，就如同 Object.wait 一样。
一个 Condition 实例本质上与一个锁绑定。要为特定的 Lock 实例获取 Condition 实例，需使用其 newCondition () 方法。
例如，假设我们有一个有界缓冲区，支持 put 和 take 方法。如果在空缓冲区上尝试 take 操作，那么线程将阻塞，直到有项目可用；如果在满缓冲区上尝试 put 操作，那么线程将阻塞，直到有空间可用。我们希望将等待的 put 线程和 take 线程分别放在不同的等待集中，这样当缓冲区中有项目或空间可用时，我们可以采用每次仅通知单个线程的优化方式。这可以通过使用两个 Condition 实例来实现。
/*** BoundedBuffer类实现了一个线程安全的有界缓冲区。* 使用ReentrantLock和Condition实现线程同步。*/
class BoundedBuffer {// 用于同步的锁final Lock lock = new ReentrantLock();// 缓冲区不满的条件final Condition notFull  = lock.newCondition();// 缓冲区非空的条件final Condition notEmpty = lock.newCondition();// 存储元素的数组final Object[] items = new Object[100];// 下一个写入位置、下一个读取位置和当前元素数量int putptr, takeptr, count;/*** 将元素放入缓冲区。* 如果缓冲区已满，则等待直到有空间可用。** @param x 要放入缓冲区的元素* @throws InterruptedException 如果等待过程中线程被中断*/public void put(Object x) throws InterruptedException {lock.lock();try {while (count == items.length)notFull.await();items[putptr] = x;if (++putptr == items.length) putptr = 0;++count;notEmpty.signal();} finally {lock.unlock();}}/*** 从缓冲区中取出一个元素。* 如果缓冲区为空，则等待直到有元素可用。** @return 从缓冲区中取出的元素* @throws InterruptedException 如果等待过程中线程被中断*/public Object take() throws InterruptedException {lock.lock();try {while (count == 0)notEmpty.await();Object x = items[takeptr];if (++takeptr == items.length) takeptr = 0;--count;notFull.signal();return x;} finally {lock.unlock();}}
}
（java.util.concurrent.ArrayBlockingQueue 类已经提供了此功能，所以没有理由实现这个示例使用类。）
一个 Condition 实现可以提供与 Object 监视器方法不同的行为和语义，比如保证通知的顺序，或者在执行通知时不要求持有锁。如果一个实现提供了此类特殊语义，那么该实现必须对这些语义进行文档说明。
请注意，Condition 实例只是普通对象，它们自身可以在 synchronized 语句中用作目标，并且可以调用它们自己的监视器等待和通知方法。获取 Condition 实例的监视器锁，或者使用其监视器方法，与获取与该 Condition 关联的 Lock 或者使用其等待和信号方法之间没有特定关系。为避免混淆，建议除了在其自身实现内部，不要以这种方式使用 Condition 实例。
除非另有说明，任何参数传入 null 值都将导致抛出 NullPointerException。
实现注意事项
一般来说，在等待 Condition 时，允许发生 “虚假唤醒”，这是对底层平台语义的一种妥协。这对大多数应用程序实际影响不大，因为应该始终在循环中等待 Condition，同时测试正在等待的状态谓词。实现可以自由消除虚假唤醒的可能性，但建议应用程序程序员始终假定虚假唤醒可能发生，因此始终在循环中等待。
条件等待的三种形式（可中断、不可中断和定时）在某些平台上的实现难度和性能特征可能有所不同。特别是，要提供这些特性并维持特定语义（如顺序保证）可能很困难。此外，在所有平台上，中断线程实际挂起状态的能力并非总是可行的。
因此，一种实现不要求为所有三种等待形式定义完全相同的保证或语义，也不要求支持中断线程的实际挂起状态。
一种实现必须清楚地记录每个等待方法所提供的语义和保证，并且当一种实现确实支持中断线程挂起时，它必须遵守此接口中定义的中断语义。
由于中断通常意味着取消，并且对中断的检查往往不频繁，因此一种实现可以优先响应中断而不是正常的方法返回。即使可以证明中断发生在可能已解除线程阻塞的其他操作之后，情况也是如此。实现应该记录这种行为。

源码实现

可以看到它的源码注释超级详细，我们再看看它的源码实现：

/*** Condition接口提供了一种线程间协调的机制,类似于Object的监视器方法。* 它允许线程在等待某个条件时释放锁,并在条件满足时被唤醒。* * Condition实例与Lock实例绑定,通过Lock.newCondition()方法获得。* * 主要方法包括:* - await(): 使当前线程等待,直到被唤醒或中断* - signal(): 唤醒一个等待的线程* - signalAll(): 唤醒所有等待的线程*/
public interface Condition {/*** 使当前线程等待,直到被唤醒或中断。* * 调用此方法时,当前线程会:* 1. 原子性地释放与此Condition关联的锁* 2. 禁用线程调度,进入休眠状态* 3. 在被唤醒、中断或发生虚假唤醒时,重新获取锁并返回* * 如果线程在进入此方法时已被中断,或等待时被中断,* 则抛出InterruptedException并清除中断状态。** @throws InterruptedException 如果当前线程在等待时被中断*/void await() throws InterruptedException;/*** 使当前线程等待,直到被唤醒。* * 与await()类似,但忽略中断。即使线程被中断,也会继续等待。* 当方法返回时,线程的中断状态仍然保持设置状态。*/void awaitUninterruptibly();/*** 使当前线程等待,直到被唤醒、中断或超时。* * 行为类似于await(),但增加了超时机制。* * @param nanosTimeout 最大等待时间,以纳秒为单位* @return 剩余等待时间的估计值,如果由于超时而返回则小于等于零* @throws InterruptedException 如果当前线程在等待时被中断*/long awaitNanos(long nanosTimeout) throws InterruptedException;/*** 使当前线程等待,直到被唤醒、中断或超时。* * 此方法在行为上等同于:* awaitNanos(unit.toNanos(time)) > 0** @param time 最大等待时间* @param unit time参数的时间单位* @return 如果在方法返回前等待时间已经超时则返回false,否则返回true* @throws InterruptedException 如果当前线程在等待时被中断*/boolean await(long time, TimeUnit unit) throws InterruptedException;/*** 使当前线程等待,直到被唤醒、中断或到达指定的最后期限。** @param deadline 等待的绝对时间期限* @return 如果在方法返回时最后期限已过则返回false,否则返回true* @throws InterruptedException 如果当前线程在等待时被中断*/boolean awaitUntil(Date deadline) throws InterruptedException;/*** 唤醒一个等待的线程。* * 如果有线程正在等待此条件,则选择其中一个唤醒。* 被唤醒的线程在从await()返回之前必须重新获取锁。*/void signal();/*** 唤醒所有等待的线程。* * 如果有线程正在等待此条件,则全部唤醒。* 每个被唤醒的线程在从await()返回之前必须重新获取锁。*/void signalAll();
}

方法

接下来我们深入看看里面的方法

await()

导致当前线程等待，直到收到信号或被中断。
与此 Condition 关联的锁会被原子性地释放，并且当前线程会因线程调度目的而被禁用，进入休眠状态，直到以下四种情况之一发生：

其他某个线程调用此 Condition 的 signal 方法，且当前线程恰好被选中作为要唤醒的线程
其他某个线程调用此 Condition 的 signalAll 方法；
其他某个线程中断当前线程，并且支持中断线程的挂起状态；
发生 “虚假唤醒”。

在所有这些情况下，在该方法返回之前，当前线程必须重新获取与此条件关联的锁。当线程返回时，保证其持有此锁。

如果当前线程：

在进入此方法时已设置其中断状态；
在等待时被中断，并且支持中断线程的挂起状态，

那么将抛出 InterruptedException，并且当前线程的中断状态将被清除。在第一种情况下，未指定在释放锁之前是否进行中断测试。

实现注意事项

调用此方法时，假定当前线程持有与此 Condition 关联的锁。由实现来确定是否是这种情况，如果不是，确定如何响应。通常，会抛出一个异常（如 IllegalMonitorStateException），并且实现必须记录这一事实。

一种实现可以优先响应中断，而不是响应信号后的正常方法返回。在这种情况下，如果存在其他等待线程，实现必须确保将信号重定向到另一个等待线程。

awaitUninterruptibly()

使当前线程等待，直至收到信号。

与此条件关联的锁会被原子性地释放，当前线程会因线程调度目的而停用并进入休眠状态，直到以下三种情况之一发生：

其他某个线程调用此 Condition 的 signal 方法，且当前线程恰好被选为要唤醒的线程；
其他某个线程调用此 Condition 的 signalAll 方法；
发生 “虚假唤醒”。

在所有情况下，在该方法返回前，当前线程必须重新获取与此条件关联的锁。当线程返回时，可确保其持有该锁。

如果当前线程在进入此方法时中断状态已被设置，或者在等待时被中断，它将继续等待，直至收到信号。当它最终从此方法返回时，其中断状态仍将保持设置。

实现注意事项

调用此方法时，假定当前线程持有与此 Condition 关联的锁。由实现来判断是否确实如此，若不是，则决定如何响应。通常会抛出一个异常（例如 IllegalMonitorStateException），并且实现必须记录这一情况。

awaitNanos(long nanosTimeout)

使当前线程等待，直到它收到信号、被中断或者指定的等待时间过去。

与此条件关联的锁会被原子性地释放，当前线程会因线程调度目的而停用，并进入休眠状态，直到以下五种情况之一发生：

其他某个线程调用此 Condition 的 signal 方法，并且当前线程恰好被选为要唤醒的线程；
其他某个线程调用此 Condition 的 signalAll 方法；
其他某个线程中断当前线程，并且支持对线程挂起状态的中断；
指定的等待时间过去；
发生 “虚假唤醒”。

在所有情况下，在该方法返回之前，当前线程必须重新获取与此条件关联的锁。当线程返回时，保证其持有该锁。

如果当前线程：

在进入此方法时其中断状态已被设置；或者
在等待时被中断，并且支持对线程挂起状态的中断，

那么将抛出 InterruptedException，并且当前线程的中断状态将被清除。在第一种情况下，未指定在释放锁之前是否进行中断测试。

该方法返回时，会给出基于传入的 nanosTimeout 值所估计的剩余等待纳秒数，如果超时，则返回小于或等于零的值。此值可用于确定在等待返回但预期条件仍不满足的情况下，是否需要再次等待以及等待多长时间。此方法的典型使用形式如下：
/*** 在指定的超时时间内等待条件满足* @param timeout 等待的超时时间* @param unit 超时时间的单位* @return 如果条件满足返回true,如果超时返回false*/
boolean aMethod(long timeout, TimeUnit unit) {    // 将超时时间转换为纳秒long nanos = unit.toNanos(timeout);    // 获取锁lock.lock();   try {     // 当条件不满足时,继续等待while (!conditionBeingWaitedFor()) {     // 如果等待时间已经用完,返回falseif (nanos <= 0L)         return false;       // 等待指定的时间,并更新剩余等待时间nanos = theCondition.awaitNanos(nanos);    }     // 条件满足,执行后续操作// ...   } finally {    // 确保在方法结束时释放锁lock.unlock();    }  // 注意:这里缺少return语句,应该添加return true;
}
设计说明：此方法需要一个以纳秒为单位的参数，以避免在报告剩余时间时出现截断误差。这种精度损失会使程序员难以确保在需要重新等待的情况下，总等待时间不会系统性地短于指定时间。

实现注意事项

调用此方法时，假定当前线程持有与此 Condition 关联的锁。由实现来判断是否确实如此，若不是，则决定如何响应。通常会抛出一个异常（例如 IllegalMonitorStateException），并且实现必须记录这一情况。

一种实现可以优先响应中断，而非响应信号后的正常方法返回，或者优先响应中断而非指示指定等待时间已过。在这两种情况下，如果存在其他等待线程，实现都必须确保将信号重定向到另一个等待线程。

await(long time, TimeUnit unit)

使当前线程等待，直到它收到信号、被中断，或者指定的等待时间过去。此方法在行为上等同于：awaitNanos(unit.toNanos(time)) > 0

awaitUntil(Date deadline)

使当前线程等待，直到它收到信号、被中断，或者指定的截止时间过去。

与此条件关联的锁会被原子性地释放，当前线程会因线程调度目的而停用，并进入休眠状态，直到以下五种情况之一发生：

其他某个线程调用此 Condition 的 signal 方法，且当前线程恰好被选为要唤醒的线程；
其他某个线程调用此 Condition 的 signalAll 方法；
其他某个线程中断当前线程，并且支持对线程挂起状态的中断；
指定的截止时间过去；
发生 “虚假唤醒”。

在所有情况下，在该方法返回之前，当前线程必须重新获取与此条件关联的锁。当线程返回时，保证其持有该锁。

如果当前线程：

在进入此方法时其中断状态已被设置；或者
在等待时被中断，并且支持对线程挂起状态的中断，

那么将抛出 InterruptedException，并且当前线程的中断状态将被清除。在第一种情况下，未指定在释放锁之前是否进行中断测试。

返回值表示截止时间是否已过，可如下使用：
/*** 在指定的截止时间之前等待某个条件满足。* * @param deadline 等待的截止时间* @return 如果条件在截止时间之前满足则返回true，否则返回false*/
boolean aMethod(Date deadline) {boolean stillWaiting = true;lock.lock();  // 获取锁try {while (!conditionBeingWaitedFor()) {  // 当条件不满足时循环if (!stillWaiting)return false;  // 如果已经不再等待，则返回falsestillWaiting = theCondition.awaitUntil(deadline);  // 等待直到截止时间，返回是否仍在等待}     // 条件满足，可以继续执行其他操作// ...} finally {   lock.unlock();  // 确保在方法结束时释放锁} // 注意：这里缺少return语句，应该添加一个返回值
}
实现注意事项

调用此方法时，假定当前线程持有与此 Condition 关联的锁。由具体实现来判断是否确实如此，若不是，则决定如何响应。通常会抛出一个异常（例如IllegalMonitorStateException），并且实现必须记录这一情况。

一种实现可以优先响应中断，而非响应信号后的正常方法返回，或者优先响应中断而非指示指定截止时间已过。在这两种情况下，如果存在其他等待线程，实现都必须确保将信号重定向到另一个等待线程。

signal()

唤醒一个等待的线程。

如果有任何线程在等待此条件，那么会选择其中一个唤醒。被唤醒的线程在从 await 返回之前，必须重新获取锁。

实现注意事项

一种实现可能（并且通常会）要求在调用此方法时，当前线程持有与此 Condition 关联的锁。实现必须记录此前置条件，以及如果未持有锁时所采取的任何操作。通常，会抛出诸如 IllegalMonitorStateException 之类的异常。

signalAll()

唤醒所有等待的线程。

如果有任何线程在等待此条件，那么所有这些线程都会被唤醒。每个线程在从await返回之前，都必须重新获取锁。

实现注意事项

一种实现可能（并且通常会）要求在调用此方法时，当前线程持有与此Condition关联的锁。实现必须记录这个前置条件，以及在未持有锁时所采取的任何措施。通常情况下，会抛出诸如IllegalMonitorStateException这样的异常。

作用

我们假设线程 1 需要等待某些条件满足后，才能继续运行，这个条件会根据业务场景不同，有不同的可能性，比如等待某个时间点到达或者等待某些任务处理完毕。在这种情况下，我们就可以执行Condition 的 await 方法，一旦执行了该方法，这个线程就会进入 WAITING 状态。

通常会有另外一个线程，我们把它称作线程 2，它去达成对应的条件，直到这个条件达成之后，那么，线程 2 调用 Condition 的 signal 方法（或 signalAll 方法），代表“这个条件已经达成了，之前等待这个条件的线程现在可以苏醒了”。这个时候，JVM 就会找到等待该 Condition 的线程，并予以唤醒，根据调用的是 signal 方法或 signalAll 方法，会唤醒 1 个或所有的线程。于是，线程 1 在此时就会被唤醒，然后它的线程状态又会回到 Runnable 可执行状态。

代码案例

我们用一个代码来说明这个问题，如下所示：

/*** ConditionDemo类演示了如何使用ReentrantLock和Condition进行线程同步。* 该类包含两个方法,用于展示线程等待和唤醒的机制。*/
public class ConditionDemo {/** 可重入锁,用于同步访问共享资源 */private ReentrantLock lock = new ReentrantLock();/** 与锁相关联的条件变量,用于线程间的通信 */private Condition condition = lock.newCondition();/*** 演示等待条件的方法。* 该方法获取锁,然后等待条件满足。* @throws InterruptedException 如果线程在等待时被中断*/void method1() throws InterruptedException {lock.lock();try{System.out.println(Thread.currentThread().getName()+":条件不满足，开始 await");condition.await();System.out.println(Thread.currentThread().getName()+":条件满足了，开始执行后续的任务");}finally {lock.unlock();}}/*** 演示唤醒等待线程的方法。* 该方法获取锁,模拟一些准备工作,然后唤醒等待的线程。* @throws InterruptedException 如果线程在睡眠时被中断*/void method2() throws InterruptedException {lock.lock();try {System.out.println(Thread.currentThread().getName() + ":需要5秒钟的准备时间");Thread.sleep(5000);System.out.println(Thread.currentThread().getName() + ":准备工作完成，唤醒其他的线程");condition.signal();} finally {lock.unlock();}}/*** 主方法,用于演示ConditionDemo的使用。* 创建两个线程,一个执行method2,另一个执行method1。* @param args 命令行参数(未使用)* @throws InterruptedException 如果主线程在等待时被中断*/public static void main(String[] args) throws InterruptedException {ConditionDemo conditionDemo = new ConditionDemo();new Thread(new Runnable() {@Overridepublic void run() {try {conditionDemo.method2();} catch (InterruptedException e) {e.printStackTrace();}}}).start();conditionDemo.method1();}
}

在这个代码中，有以下三个方法。

method1，它代表主线程将要执行的内容，首先获取到锁，打印出“条件不满足，开始await”，然后调用 condition.await() 方法，直到条件满足之后，则代表这个语句可以继续向下执行了，于是打印出“条件满足了，开始执行后续的任务”，最后会在 finally 中解锁。
method2，它同样也需要先获得锁，然后打印出“需要 5 秒钟的准备时间”，接着用 sleep 来模拟准备时间；在时间到了之后，则打印出“准备工作完成”，最后调用 condition.signal() 方法，把之前已经等待的线程唤醒。
main 方法，它的主要作用是执行上面这两个方法，它先去实例化我们这个类，然后再用子线程去调用这个类的 method2 方法，接着用主线程去调用 method1 方法。

最终这个代码程序运行结果如下所示：

同时也可以看到，打印这行语句它所运行的线程，第一行语句和第四行语句打印的是在 main 线程中，也就是在主线程中去打印的，而第二、第三行是在子线程中打印的。这个代码就模拟了我们前面所描述的场景。

注意点

下面我们来看一下，在使用 Condition 的时候有哪些注意点。线程 2 解锁后，线程 1 才能获得锁并继续执行线程 2 对应刚才代码中的子线程，而线程 1 对应主线程。这里需要额外注意，并不是说子线程调用了 signal 之后，主线程就可以立刻被唤醒去执行下面的代码了，而是说在调用了 signal 之后，还需要等待子线程完全退出这个锁，即执行 unlock 之后，这个主线程才有可能去获取到这把锁，并且当获取锁成功之后才能继续执行后面的任务。刚被唤醒的时候主线程还没有拿到锁，是没有办法继续往下执行的。