Condition实现等待唤醒的原理
1、引言
任何一个Java对象，都拥有一组监视器方法（定义在java.lang.Object上），主要包括了wait、notify、notifyAll方法，他们与synchronized关键字配合可以实现等待/通知模型。那么如果我们不适用内置锁而是通过Lock想要实现等待/通知模型的话就需要使用到Condition接口。这两种实现等待通知模型的方式在性能上还是有一些不同的，这里引用《Java并发编程的艺术》里的对比图片：
对比.png
2、使用
Condition配合Lock的实现等待唤醒的方式与内置锁等待唤醒的使用方式类似，都需要获取锁（这里的是调用Lock方法），调用ReentrantLock的newCondition创建ConditionObject对象，通过conditionObject对象调用await实现等待，调用signal实现唤醒，signalAll唤醒全部。
下面通过一个通过ReentrantLock与Condition实现的生产者消费者模型看一下使用流程：
public class ProducterAndConsumerTest {
private static Queue<Integer> queue = new ArrayDeque<>();
//创建ReentrantLock
private static ReentrantLock lock = new ReentrantLock();
//通过newCondition创建Condition的实例
private static Condition condition = lock.newCondition();
public static void main(String[] args) {
new Thread(new ProducterRunnable()).start();
new Thread(new ConsumerRunnable()).start();
}
private static class ProducterRunnable implements Runnable {
@Override
public void run() {
try {
lock.lock();
while (true) {
if (queue.size() >= 10) {
System.out.println("生产者车间已满");
condition.await();
}
queue.offer(1);
condition.signal();
System.out.println("生产者生产了一个苹果");
Thread.sleep(1000);
}
} catch (Exception e) {
} finally {
lock.unlock();
}
}
}
private static class ConsumerRunnable implements Runnable {
@Override
public void run() {
try {
lock.lock();
while (true) {
if (queue.size() <= 0) {
System.out.println("消费者都吃完了");
condition.await();
}
System.out.println("消费者吃了一个苹果");
condition.signal();
queue.poll();
Thread.sleep(1000);
}
} catch (Exception e) {
} finally {
lock.unlock();
}
}
}
}
使用总结：
1、创建ReentrantLock对象
2、通过ReentrantLock对象调用newCondition方法从而创建ConditionObject对象
3、根据需求调用condition.await实现等待；通过condition.signal或者condition.signalAll实现唤醒
3、源码分析
对于ReentrantLock中同步队列的介绍，加锁解锁的原理参考之前的文章：
3.1、newCondition/await
private static Condition condition = lock.newCondition();
//ReentrantLock中的newCondition方法
public Condition newCondition() {
return sync.newCondition();
}
//ReentrantLock中内部类Sync的newCondition方法
final ConditionObject newCondition() {
//分析1
return new ConditionObject();
}
===>分析1
//ConditionObject是AQS（AbstractQueuedSynchronizer）中的内部类，实现
//了Condition接口；通过ConditionObject类（firstWaiter，lastWaiter）可以分析出
//Condition中也包含了一个等待队列拥有收尾节点，在调用await方法时会将线程构造成Node节点（这里的Node是AQS的内部类NODE）
//，那么我们看一下ConditionObject中的await方法
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
//将当前的线程构造成Node节点未插入等待队列
Node node = addConditionWaiter();
//释放当前线程所占用的锁，在释放过程中会唤醒下一个节点
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
//当前线程等待
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
//自旋获取到同步状态（即获取到LOCK）
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
//处理中断
reportInterruptAfterWait(interruptMode);
}
当调用newCondition时会将AQS中的内部类ConditionObject进行实例化；当调用await时会将当前线程构造成Node节点并通过尾插入 插入到等待队列末尾处（注意：这里的等待队列和AQS的同步队列不是一个）
同步队列与等待队列.png
当await执行后，实际上就是从AQS的同步队列当中把头结点处的Node拿下来尾插入到等待队列末端
当前线程加入到等待队列中.png
在await后，同步队列后面的节点会被唤醒获取锁执行，之前的节点被放入到等待队列中自旋等待唤醒。
3.2、signal
//唤醒方法
public final void signal() {
//判断当前线程是否已经获取了锁
//分析1
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
//拿到等待队列中的头结点 并执行doSignal
Node first = firstWaiter;
if (first != null)
//分析2
doSignal(first);
}
===>分析1
//ReentrantLock中的isHeldExclusively方法
//判断当前获取独占锁的线程是否是当前线程
protected final boolean isHeldExclusively() {
// While we must in general read state before owner,
// we don't need to do so to check if current thread is owner
return getExclusiveOwnerThread() == Thread.currentThread();
}
===>分析2
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
//将头部节点从队列中移除
first.nextWaiter = null;
//分析3
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
===>分析3
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
//将当前node更新状态为0
if (!node.compareAndSetWaitStatus(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 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
唤醒操作就是将等待队列中的头结点放入到同步队列中等待获取锁后继续执行
等待队列到同步队列.png
4、总结
通过调用ReentrantLock中的newCondition方法最终会创建一个ConditionObject（AQS下的内部类实现了Condition接口）；await方法就是将要阻塞的线程从同步队列中拿出尾插入到等待队列中，signal时会将等待时间最久的（等待队列中的头结点）从等待队列中移出放入到同步队列中，等待获取锁后执行。
参考：
《Java并发编程的艺术》
详解Condition
网友评论