lesson5:Condition的原理分析及demo展示

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

原文地址:https://www.cnblogs.com/mantu/p/5761501.html