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()线程。