Condition 将 Object 监视器方法（wait，notify，和notifyAll）分解成截然不同的对象，以便通过将这些对象与任意Lock实现组合使用，为每个对象提供多个等待 set（wait-set）。其中，Lock 替代了 synchronized 方法和语句的使用，Condition 替代了 Object 监视器方法的使用。每个condition都是与一个锁关联的，一个锁可以创造一个或多个condition。
关于condition的描述，参考资料：http://download.oracle.com/technetwork/java/javase/6/docs/zh/api/java/util/concurrent/locks/Condition.html
demo源码：https://github.com/mantuliu/javaAdvance 中的Lesson5相关的类
我们来看一下Condition接口的实现类ConditionObject
public class ConditionObject implements Condition, java.io.Serializable {
private static final long serialVersionUID = 1173984872572414699L;
/** First node of condition queue. */
private transient Node firstWaiter;//第一个等待节点，后面我们会看到每个wait的线程都是一个等待节点
/** Last node of condition queue. */
private transient Node lastWaiter;//最后一个等待节点
/**
* Creates a new <tt>ConditionObject</tt> instance.
*/
public ConditionObject() { }
// Internal methods
/** * 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)) {//判断node是否被唤醒 LockSupport.park(this);//等待unpark许可 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); }
/**
* Adds a new waiter to wait queue.
* @return its new wait node
*/
private Node addConditionWaiter() {//在wait队列中创建一个新的节点
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() {//释放wait队列中所有非等待状态的节点
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);//通知第一个节点，它调用的方法调用了LockSupport.unpark(node.thread);unpark与await中的park对应
}
/**
* 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 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);
long lastTime = System.nanoTime();
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
if (nanosTimeout <= 0L) {
transferAfterCancelledWait(node);
break;
}
LockSupport.parkNanos(this, nanosTimeout);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
long now = System.nanoTime();
nanosTimeout -= now - lastTime;
lastTime = now;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
return nanosTimeout - (System.nanoTime() - lastTime);
}
/**
* 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 {
if (deadline == null)
throw new NullPointerException();
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 {
if (unit == null)
throw new NullPointerException();
long nanosTimeout = unit.toNanos(time);
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
long lastTime = System.nanoTime();
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;
long now = System.nanoTime();
nanosTimeout -= now - lastTime;
lastTime = now;
}
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}.
*
* @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}.
*
* @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}.
*
* @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的代码后，我产生了以下几点推论：
1.await()方法是可以响应线程中断命令的；
2.await()和signal()方法在当前线程锁定了Condition对应的锁时才能使用；
3.一个锁对应多个Condition对象，每个Condition的signal()方法只通知相同Condition的await()方法，condition之间不会互相通知；
4.signal()被调用时，wait队列中的第一个对象会被唤醒，signalAll()时，wait队列中元素的全部按先入先出的顺序被唤醒；
5.如果既存在await的线程，又存在一直等待lock()的线程，当signal()的线程完成时，lock()的线程优先级比await()的线程优先级高，当所有lock()线程获取到锁并释放后，才会轮到await()线程。
下面我将用代码来证明这几个推论:
下面的例子展示了await()方法是响应中断的，而awaitUninterruptibly()是无视外部中断的:
package com.mantu.advance;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
* blog http://www.cnblogs.com/mantu/
* github https://github.com/mantuliu/
* @author mantu
*
*/
public class Lesson5ConditionAwaitInterrupt implements Runnable {
public static ReentrantLock lock = new ReentrantLock();
public static Condition con = lock.newCondition();
public static void main(String [] args) throws InterruptedException{
Thread thread = new Thread(new Lesson5ConditionAwaitInterrupt());
thread.start();
Thread.sleep(5000L);//当前线程睡5秒
System.out.println("开始中断线程");
thread.interrupt();//中断线程
}
@Override
public void run() {
try{
lock.lock();
while(true){
System.out.println("开始等待signal的通知");
con.await();
//con.awaitUninterruptibly();
}
}
catch(Exception ex){
ex.printStackTrace();
}
finally{
lock.unlock();
}
}
}
下面的例子展示了当前线程一定是在获得了condition对应的锁的情况下，才能调用await和signal等方法：
package com.mantu.advance;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
* blog http://www.cnblogs.com/mantu/
* github https://github.com/mantuliu/
* @author mantu
*
*/
public class Lesson5ConditionAwaitWithLock implements Runnable {
public static ReentrantLock lock = new ReentrantLock();
public static Condition con = lock.newCondition();
public static void main(String [] args) throws InterruptedException {
Thread thread = new Thread(new Lesson5ConditionAwaitWithLock());
thread.start();
try {
Thread.sleep(5000L);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
con.signal();//抛出异常
con.await();
}
@Override
public void run() {
try {
lock.lock();
con.signal();
System.out.println("通知信号发送完毕");
con.await();
System.out.println("接收到通知信号");
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
}
}
下面的例子展示了一个锁对应多个Condition对象，每个Condition的signal()方法只通知相同Condition的await()方法，condition之间不会互相通知；signal()被调用时，wait队列中的第一个对象会被唤醒:
package com.mantu.advance;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* blog http://www.cnblogs.com/mantu/
* github https://github.com/mantuliu/
* @author mantu
*
*/
public class Lesson5ConditonsRelation{
public static ReentrantLock lock = new ReentrantLock();
public static Condition conOne = lock.newCondition();
public static Condition conTwo = lock.newCondition();
public static void main(String [] args){
for(int i=0;i<3;i++){
new Thread(new UsedConditionOne()).start();
new Thread(new UsedConditionTwo()).start();
}
try {
Thread.sleep(5000L);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
try {
lock.lock();
Lesson5ConditonsRelation.conOne.signal();
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
try {
Thread.sleep(5000L);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
try {
lock.lock();
Lesson5ConditonsRelation.conTwo.signal();
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
}
}
class UsedConditionOne implements Runnable{
@Override
public void run() {
Lock lock = Lesson5ConditonsRelation.lock;
try {
lock.lock();
System.out.println("UsedConditionOne ‘s thread "+ Thread.currentThread().getId() + " 开始等待");
Lesson5ConditonsRelation.conOne.await();
System.out.println("UsedConditionOne ‘s thread "+ Thread.currentThread().getId() + " 等待完毕");
Lesson5ConditonsRelation.conOne.signal();
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
}
}
class UsedConditionTwo implements Runnable{
@Override
public void run() {
Lock lock = Lesson5ConditonsRelation.lock;
try {
lock.lock();
System.out.println("UsedConditionTwo ‘s thread "+ Thread.currentThread().getId() + " 开始等待");
Lesson5ConditonsRelation.conTwo.await();
System.out.println("UsedConditionTwo ‘s thread "+ Thread.currentThread().getId() + " 等待完毕");
Lesson5ConditonsRelation.conTwo.signal();
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
}
}
下面的demo展示了signalAll()方法会通知所有的await()线程：
package com.mantu.advance;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
* blog http://www.cnblogs.com/mantu/
* github https://github.com/mantuliu/
* @author mantu
*
*/
public class Lesson5ConditionSignalAll implements Runnable{
public static ReentrantLock lock = new ReentrantLock();
public static Condition con = lock.newCondition();
public static void main(String [] args){
for(int i=0;i<30;i++){
Thread thread = new Thread(new Lesson5ConditionSignalAll());
thread.start();
}
try {
Thread.sleep(5000L);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
try {
lock.lock();
con.signalAll();
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
}
@Override
public void run() {
try {
lock.lock();
System.out.println("thread "+ Thread.currentThread().getId() + " 开始等待");
con.await();
System.out.println("thread "+ Thread.currentThread().getId() + " 等待完毕");
Thread.sleep(1000L);
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
}
}
下面的demo展示了在同等条件下去获取锁，lock()线程的优先级会比await()线程的优先级高:
package com.mantu.advance;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* blog http://www.cnblogs.com/mantu/
* github https://github.com/mantuliu/
* @author mantu
*
*/
public class Lesson5ConditionSignalAwaitLock {
public static ReentrantLock lock = new ReentrantLock();
public static Condition conOne = lock.newCondition();
public static void main(String [] args) throws InterruptedException{
for(int i=0;i<3;i++){
new Thread(new ThreadAwait()).start();
}
Thread.currentThread().sleep(1000L);
new Thread(new ThreadSignal()).start();
Thread.currentThread().sleep(1000L);
for(int i=0;i<3;i++){
new Thread(new ThreadLock()).start();
}
}
}
class ThreadAwait implements Runnable{
@Override
public void run() {
Lock lock = Lesson5ConditionSignalAwaitLock.lock;
try {
lock.lock();
System.out.println("ThreadAwait ‘s thread "+ Thread.currentThread().getId() + " 开始等待");
Lesson5ConditionSignalAwaitLock.conOne.await();
System.out.println("ThreadAwait ‘s thread "+ Thread.currentThread().getId() + " 等待完毕");
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
}
}
class ThreadSignal implements Runnable{
@Override
public void run() {
Lock lock = Lesson5ConditionSignalAwaitLock.lock;
try {
lock.lock();
System.out.println("ThreadSignal ‘s thread "+ Thread.currentThread().getId() + " 拿到了锁");
Thread.currentThread().sleep(2000L);
new Thread(new ThreadLock()).start();
System.out.println("ThreadSignal ‘s thread "+ Thread.currentThread().getId() + " 开始通知");
Lesson5ConditionSignalAwaitLock.conOne.signal();
Lesson5ConditionSignalAwaitLock.conOne.signal();
Lesson5ConditionSignalAwaitLock.conOne.signal();
System.out.println("ThreadSignal ‘s thread "+ Thread.currentThread().getId() + " 通知完毕");
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
}
}
class ThreadLock implements Runnable{
@Override
public void run() {
Lock lock = Lesson5ConditionSignalAwaitLock.lock;
try {
System.out.println("ThreadLock ‘s thread "+ Thread.currentThread().getId() + " 进入线程开始拿锁");
lock.lock();
System.out.println("ThreadLock ‘s thread "+ Thread.currentThread().getId() + " 拿到了锁");
} catch (Exception e) {
e.printStackTrace();
}
finally{
lock.unlock();
}
}
}
到此为止，本文已经分析完毕：最后再强调一下重点：
1.await()方法是可以响应线程中断命令的；
2.await()和signal()方法在当前线程锁定了Condition对应的锁时才能使用；
3.一个锁对应多个Condition对象，每个Condition的signal()方法只通知相同Condition的await()方法，condition之间不会互相通知；
4.signal()被调用时，wait队列中的第一个对象会被唤醒，signalAll()时，wait队列中元素的全部按先入先出的顺序被唤醒；
5.如果既存在await的线程，又存在一直等待lock()的线程，当signal()的线程完成时，lock()的线程优先级比await()的线程优先级高，当所有lock()线程获取到锁并释放后，才会轮到await()线程。
时间： 2024-10-19 17:17:18