一、概述
任何一个Java对象，都拥有一组监视器方法，主要包括wait()、notify()、notifyAll()方法，这些方法与synchronized关键字配合使用可以实现等待/通知机制。使用这种方式实现了生产者-消费者模式。
类似地，Condition接口也提供类似的Object的监视器的方法，主要包括await()、signal()、signalAll()方法，这些方法与Lock锁配合使用也可以实现等待/通知机制。
1.1、对比object监视器
| 对比项 | Object监视器 | Condition |
|---|---|---|
| 前置条件 | 获取对象的锁 | 调用Lock.lock获取锁，调用Lock.newCondition获取Condition对象 |
| 调用方式 | 直接调用，比如object.notify() | 直接调用，比如condition.await() |
| 等待队列的个数 | 一个 | 多个 |
| 当前线程释放锁进入等待状态 | 支持 | 支持 |
| 当前线程释放锁进入等待状态，在等待状态中不断响中断 | 不支持 | 支持 |
| 当前线程释放锁并进入超时等待状态 | 支持 | 支持 |
| 当前线程释放锁并进入等待状态直到将来的某个时间 | 不支持 | 支持 |
| 唤醒等待队列中的一个线程 | 支持 | 支持 |
| 唤醒等待队列中的全部线程 | 支持 | 支持 |
1.2、Condition接口定义
public interface Condition {
// 使当前线程处于等待状态，释放与Condtion绑定的lock锁
// 直到 singal()方法被调用后，被唤醒（若中断，就game over了）
// 唤醒后，该线程会再次获取与条件绑定的 lock锁
void await() throws InterruptedException;
// 相比较await()而言，不响应中断
void awaitUninterruptibly();
// 在wait()的返回条件基础上增加了超时响应，返回值表示当前剩余的时间
// < 0 ，则表示超时
long awaitNanos(long nanosTimeout) throws InterruptedException;
// 同上，只是时间参数不同而已
boolean await(long time, TimeUnit unit) throws InterruptedException;
// 同上，只是时间参数不同而已
boolean awaitUntil(Date deadline) throws InterruptedException;
// 表示条件达成，唤醒一个被条件阻塞的线程
void signal();
// 唤醒所有被条件阻塞的线程。
void signalAll();
}
Condition的使用小结:
- Condition与Lock配套使用，通过
Lock#newConditin()进行实例化 -
Condition#await()会释放lock，线程阻塞；直到线程中断orCondition#singal()被执行，唤醒阻塞线程，并重新获取lock - 经典case可以参考jdk的阻塞队列实现（ArrayBlockingQueue, LinkedBlockingQueue）
后续可以分析：ArrayBlockingQueue, LinkedBlockingQueue实现
1.3、使用Condition接口配合Lock锁的使用实例如下
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition(); public void conditionWait() throws InterruptedException {
lock.lock();
try {
//....
condition.await();
}finally {
lock.unlock();
}
} public void conditionSignal(){
lock.lock();
try {
//...
condition.signal();
}finally {
lock.unlock();
}
}
一般而言，都会将Condition变量作为成员变量。当调用await方法后，当前线程会释放锁并进入Condition变量的等待队列，而其他线程调用signal方法后，通知正在Condition变量等待队列的线程从await方法返回，并且在返回前已经获得了锁。
1.4、示例
package com.lhx.cloud.threadlock; import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock; /**
* @author lihongxu
* @desc @link(https://github.com/bjlhx15/common-study)
* @since 2019/3/26 下午4:23
*/
public class SampleQueue<T> {
private Object[] elements;
private Lock lock=new ReentrantLock();
//队列是否为空
private Condition notEmpty = lock.newCondition();
//队列是否已满
private Condition notFull = lock.newCondition();
private int length=0,addIndex=0,removeIndex=0; public SampleQueue(int size) {
this.elements = new Object[size];
} public int getLength() {
return length;
} public void add(T object) throws InterruptedException {
lock.lock();
try {
while (length==elements.length){
System.out.println("队列已满，等待~~~");
notFull.await();
}
elements[addIndex]=object;
if(addIndex++ ==elements.length){
addIndex = 0;
}
length++;
notEmpty.signal();
} finally {
lock.unlock();
}
}
public T remove() throws InterruptedException {
lock.lock();
try {
while (length==0){
System.out.println("队列为空，等待~~~");
notEmpty.await();
}
Object element = elements[removeIndex];
if(removeIndex++ ==elements.length){
removeIndex = 0;
}
length--;
notFull.signal();
return (T) element;
} finally {
lock.unlock();
}
}
}
测试用例
@Test
public void add() throws InterruptedException {
SampleQueue<String> queue=new SampleQueue<>(5);
queue.add("111");
queue.add("222");
queue.add("333");
queue.add("444");
queue.add("555");
queue.add("666");
} @Test
public void remove() throws InterruptedException {
SampleQueue<String> queue=new SampleQueue<>(5);
queue.add("111");
queue.add("222");
queue.add("333");
queue.add("444");
for (int i = 0; i <=4; i++) {
String remove = queue.remove();
System.out.println(remove);
}
String remove = queue.remove();
System.out.println(remove); }
Condition的使用方式是比较简单的，需要注意的是使用Condition的等待/通知需要提前获取到与Condition对象关联的锁，Condition对象由Lock对象创建。
以上述示例中的add(T object)为例，详细描述一下Condition等待/通知的整个过程：
- 获取锁，确保对数据数据修改的安全性；
- 数组元素的个数等于数组的长度时，调用notFull.await()，插入线程释放锁进入等待；
- 数组未满，添加元素到数组中，调用notEmpty.signal()通知等待在notEmpty上的线程，数组中有新的元素可以操作。
总的来说，Condition的等待/通知使用方式大体上跟经典的Object监视器上的等待/通知是非常类似的。
二、Condition具体实现分析
通过：Condition condition = lock.newCondition();查看ReentrantLock的实现
public Condition newCondition() {
return sync.newCondition();
}
进入到Sync实现查看newCondition
final ConditionObject newCondition() {
return new ConditionObject();
}
继续查看ConditionObject
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;
/**
* Creates a new {@code ConditionObject} instance.
*/
public ConditionObject() { }
// Internal methods
/**
* Adds a new waiter to wait queue.
* @return its new wait node
*/
private Node addConditionWaiter() {
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
/**
* Removes and transfers nodes until hit non-cancelled one or
* null. Split out from signal in part to encourage compilers
* to inline the case of no waiters.
* @param first (non-null) the first node on condition queue
*/
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
/**
* Removes and transfers all nodes.
* @param first (non-null) the first node on condition queue
*/
private void doSignalAll(Node first) {
lastWaiter = firstWaiter = null;
do {
Node next = first.nextWaiter;
first.nextWaiter = null;
transferForSignal(first);
first = next;
} while (first != null);
}
/**
* Unlinks cancelled waiter nodes from condition queue.
* Called only while holding lock. This is called when
* cancellation occurred during condition wait, and upon
* insertion of a new waiter when lastWaiter is seen to have
* been cancelled. This method is needed to avoid garbage
* retention in the absence of signals. So even though it may
* require a full traversal, it comes into play only when
* timeouts or cancellations occur in the absence of
* signals. It traverses all nodes rather than stopping at a
* particular target to unlink all pointers to garbage nodes
* without requiring many re-traversals during cancellation
* storms.
*/
private void unlinkCancelledWaiters() {
Node t = firstWaiter;
Node trail = null;
while (t != null) {
Node next = t.nextWaiter;
if (t.waitStatus != Node.CONDITION) {
t.nextWaiter = null;
if (trail == null)
firstWaiter = next;
else
trail.nextWaiter = next;
if (next == null)
lastWaiter = trail;
}
else
trail = t;
t = next;
}
}
// public methods
/**
* Moves the longest-waiting thread, if one exists, from the
* wait queue for this condition to the wait queue for the
* owning lock.
*
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code false}
*/
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
/**
* Moves all threads from the wait queue for this condition to
* the wait queue for the owning lock.
*
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code false}
*/
public final void signalAll() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignalAll(first);
}
/**
* Implements uninterruptible condition wait.
* <ol>
* <li> Save lock state returned by {@link #getState}.
* <li> Invoke {@link #release} with saved state as argument,
* throwing IllegalMonitorStateException if it fails.
* <li> Block until signalled.
* <li> Reacquire by invoking specialized version of
* {@link #acquire} with saved state as argument.
* </ol>
*/
public final void awaitUninterruptibly() {
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
boolean interrupted = false;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if (Thread.interrupted())
interrupted = true;
}
if (acquireQueued(node, savedState) || interrupted)
selfInterrupt();
}
/*
* For interruptible waits, we need to track whether to throw
* InterruptedException, if interrupted while blocked on
* condition, versus reinterrupt current thread, if
* interrupted while blocked waiting to re-acquire.
*/
/** Mode meaning to reinterrupt on exit from wait */
private static final int REINTERRUPT = 1;
/** Mode meaning to throw InterruptedException on exit from wait */
private static final int THROW_IE = -1;
/**
* Checks for interrupt, returning THROW_IE if interrupted
* before signalled, REINTERRUPT if after signalled, or
* 0 if not interrupted.
*/
private int checkInterruptWhileWaiting(Node node) {
return Thread.interrupted() ?
(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
0;
}
/**
* Throws InterruptedException, reinterrupts current thread, or
* does nothing, depending on mode.
*/
private void reportInterruptAfterWait(int interruptMode)
throws InterruptedException {
if (interruptMode == THROW_IE)
throw new InterruptedException();
else if (interruptMode == REINTERRUPT)
selfInterrupt();
}
/**
* Implements interruptible condition wait.
* <ol>
* <li> If current thread is interrupted, throw InterruptedException.
* <li> Save lock state returned by {@link #getState}.
* <li> Invoke {@link #release} with saved state as argument,
* throwing IllegalMonitorStateException if it fails.
* <li> Block until signalled or interrupted.
* <li> Reacquire by invoking specialized version of
* {@link #acquire} with saved state as argument.
* <li> If interrupted while blocked in step 4, throw InterruptedException.
* </ol>
*/
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
/**
* Implements timed condition wait.
* <ol>
* <li> If current thread is interrupted, throw InterruptedException.
* <li> Save lock state returned by {@link #getState}.
* <li> Invoke {@link #release} with saved state as argument,
* throwing IllegalMonitorStateException if it fails.
* <li> Block until signalled, interrupted, or timed out.
* <li> Reacquire by invoking specialized version of
* {@link #acquire} with saved state as argument.
* <li> If interrupted while blocked in step 4, throw InterruptedException.
* </ol>
*/
public final long awaitNanos(long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
final long deadline = System.nanoTime() + nanosTimeout;
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
if (nanosTimeout <= 0L) {
transferAfterCancelledWait(node);
break;
}
if (nanosTimeout >= spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
nanosTimeout = deadline - System.nanoTime();
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
return deadline - System.nanoTime();
}
/**
* Implements absolute timed condition wait.
* <ol>
* <li> If current thread is interrupted, throw InterruptedException.
* <li> Save lock state returned by {@link #getState}.
* <li> Invoke {@link #release} with saved state as argument,
* throwing IllegalMonitorStateException if it fails.
* <li> Block until signalled, interrupted, or timed out.
* <li> Reacquire by invoking specialized version of
* {@link #acquire} with saved state as argument.
* <li> If interrupted while blocked in step 4, throw InterruptedException.
* <li> If timed out while blocked in step 4, return false, else true.
* </ol>
*/
public final boolean awaitUntil(Date deadline)
throws InterruptedException {
long abstime = deadline.getTime();
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
boolean timedout = false;
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
if (System.currentTimeMillis() > abstime) {
timedout = transferAfterCancelledWait(node);
break;
}
LockSupport.parkUntil(this, abstime);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
return !timedout;
}
/**
* Implements timed condition wait.
* <ol>
* <li> If current thread is interrupted, throw InterruptedException.
* <li> Save lock state returned by {@link #getState}.
* <li> Invoke {@link #release} with saved state as argument,
* throwing IllegalMonitorStateException if it fails.
* <li> Block until signalled, interrupted, or timed out.
* <li> Reacquire by invoking specialized version of
* {@link #acquire} with saved state as argument.
* <li> If interrupted while blocked in step 4, throw InterruptedException.
* <li> If timed out while blocked in step 4, return false, else true.
* </ol>
*/
public final boolean await(long time, TimeUnit unit)
throws InterruptedException {
long nanosTimeout = unit.toNanos(time);
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
final long deadline = System.nanoTime() + nanosTimeout;
boolean timedout = false;
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
if (nanosTimeout <= 0L) {
timedout = transferAfterCancelledWait(node);
break;
}
if (nanosTimeout >= spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
nanosTimeout = deadline - System.nanoTime();
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
return !timedout;
}
// support for instrumentation
/**
* Returns true if this condition was created by the given
* synchronization object.
*
* @return {@code true} if owned
*/
final boolean isOwnedBy(AbstractQueuedSynchronizer sync) {
return sync == AbstractQueuedSynchronizer.this;
}
/**
* Queries whether any threads are waiting on this condition.
* Implements {@link AbstractQueuedSynchronizer#hasWaiters(ConditionObject)}.
*
* @return {@code true} if there are any waiting threads
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code false}
*/
protected final boolean hasWaiters() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
if (w.waitStatus == Node.CONDITION)
return true;
}
return false;
}
/**
* Returns an estimate of the number of threads waiting on
* this condition.
* Implements {@link AbstractQueuedSynchronizer#getWaitQueueLength(ConditionObject)}.
*
* @return the estimated number of waiting threads
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code false}
*/
protected final int getWaitQueueLength() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
int n = 0;
for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
if (w.waitStatus == Node.CONDITION)
++n;
}
return n;
}
/**
* Returns a collection containing those threads that may be
* waiting on this Condition.
* Implements {@link AbstractQueuedSynchronizer#getWaitingThreads(ConditionObject)}.
*
* @return the collection of threads
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code false}
*/
protected final Collection<Thread> getWaitingThreads() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
ArrayList<Thread> list = new ArrayList<Thread>();
for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
if (w.waitStatus == Node.CONDITION) {
Thread t = w.thread;
if (t != null)
list.add(t);
}
}
return list;
}
}
ConditionObject是Condition在java并发中的具体的实现，它是AQS的内部类。因为Condition相关操作都需要获取锁，所以作为AQS的内部类也比较合理。接下来就以ConditionObject的等待队列、等待、通知为切入点分析ConditionObject的具体实现。
2.1、等待队列
ConditionObject的等待队列是一个FIFO队列，队列的每个节点都是等待在Condition对象上的线程的引用，在调用Condition的await()方法之后，线程释放锁，构造成相应的节点进入等待队列等待。其中节点的定义复用AQS的Node定义。
等待队列的对应实现：在上述源码中实现
/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter; private Node addConditionWaiter() {}
- ConditionObject包含等待队列的首节点firstWaiter和尾节点lastWaiter；
- 线程调用await()方法时，调用addConditionWaiter()方法入队：
- step1：将线程构造成Node；
- step2：将Node加入到等待队列中。
从队列相关操作的具体实现可以知道等待队列的基本结构如下图所示：
插入节点只需要将原有尾节点的nextWaiter指向当前节点，并且更新尾节点。更新节点并没有像AQS更新同步队列使用CAS是因为调用await()方法的线程必定是获取了锁的线程，锁保证了操作的线程安全。
注：AQS实质上拥有一个同步队列和多个等待队列，具体对应关系如下图所示：
2.2、等待
调用Condition的await开头的系列方法，当前线程进入等待队列等待，那么Condition的等待实质是await系列方法的具体实现。
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
具体执行流程如下：
- 调用addConditionWaiter将当前线程加入等待队列；
- 调用fullRelease释放当前线程节点的同步状态，唤醒后继节点；
- 线程进入等待状态；
- 线程被唤醒后，从while循环中退出，调用acquireQueued尝试获取同步状态；
- 同步状态获取成功后，线程从await方法返回。
其他以await开头的方法具体实现与await基本一致，只是在它的基础上增加了超时限制，不管有没有被唤醒，到达指定时间，等待结束，从await返回。整个await系列方法将线程加入等待队列的流程可以总结为下图：
2.3、唤醒
调用Condition的signal()方法将会唤醒再等待队列中的首节点，该节点也是到目前为止等待时间最长的节点。
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
- step1：前置检查，判断当前线程是否是获取了锁的线程，如果不是抛出异常IllegalMonitorStateException，否则，执行step2；
- step2：取得等待队列的头结点，头结点不为空执行doSignal，否则，signal结束。
可以看出，doSignal方法是整个signal方法实现的核心，它完成了将线程从唤醒的所有操作。
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
整个doSignal完成了这两个操作：调用transferForSignal将节点从等待队列移动到同步队列，并且，将该节点从等待队列删除。
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
- step1：将节点waitStatus设置为0，设置成功执行step2，否则返回false；
- step2：调用enq方法将该节点加入同步队列；
- step3：使用LockSuppor.unpark()方法唤醒该节点的线程。
Condition的signalAll()方法，将等待队列中的所有节点全部唤醒，相当于将等待队列中的每一个节点都执行一次signal()。整个signal系列方法将线程从等待队列移动到同步队列可以总结为下图：
参看地址：
https://www.jianshu.com/p/be2dc7c878dc
https://blog.csdn.net/u011116672/article/details/51064752
https://my.oschina.net/u/566591/blog/1557978
https://www.cnblogs.com/yulinfeng/p/6921900.html