Java锁——Condition使用示例及讲解

admin管理员组

文章数量:1584643

Condition简介

任何一个java对象都天然继承于Object类，在线程间实现通信的往往会应用到Object的几个方法，比如wait(),wait(long timeout),wait(long timeout, int nanos)与notify(),notifyAll()几个方法实现等待/通知机制，同样的， 在java Lock体系下依然会有同样的方法实现等待/通知机制。

从整体上来看Object的wait和notify/notify是与对象监视器配合完成线程间的等待/通知机制，而Condition与Lock配合完成等待通知机制，前者是java底层级别的，后者是语言级别的，具有更高的可控制性和扩展性。两者除了在使用方式上不同外，在功能特性上还是有很多的不同：

Condition能够支持不响应中断，而通过使用Object方式不支持；

Condition能够支持多个等待队列（new 多个Condition对象），而Object方式只能支持一个；

Condition能够支持超时时间的设置，而Object不支持

Condition是在java 1.5中才出现的，它用来替代传统的Object的wait()、notify()实现线程间的协作，相比使用Object的wait()、notify()，使用Condition的await()、signal()这种方式实现线程间协作更加安全和高效。因此通常来说比较推荐使用Condition，阻塞队列实际上是使用了Condition来模拟线程间协作。

Condition是个接口，基本的方法就是await()和signal()方法；
Condition依赖于Lock接口，生成一个Condition的基本代码是lock.newCondition()
调用Condition的await()和signal()方法，都必须在lock保护之内，就是说必须在lock.lock()和lock.unlock之间才可以使用
　　Conditon中的await()对应Object的wait()；

Condition中的signal()对应Object的notify()；

Condition中的signalAll()对应Object的notifyAll()。

Demo

condition常见例子arrayblockingqueue。下面是demo：

package thread;
 
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
 * 
 * @author zhangliang
 *
 * 2016年4月8日 下午5:48:54
 */
public class ConTest {
	
	 final Lock lock = new ReentrantLock();
	 final Condition condition = lock.newCondition();
 
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		ConTest test = new ConTest();
	    Producer producer = test.new Producer();
	    Consumer consumer = test.new Consumer();
	          
	    
	    consumer.start(); 
	    producer.start();
	}
	
	 class Consumer extends Thread{
         
	        @Override
	        public void run() {
	            consume();
	        }
	          
	        private void consume() {
	           	             
	                try {
	                	   lock.lock();
	                    System.out.println("我在等一个新信号"+this.currentThread().getName());
	                    condition.await();
	                    
	                } catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} finally{
						System.out.println("拿到一个信号"+this.currentThread().getName());
	                    lock.unlock();
	                }
	            
	        }
	    }
	 
	 class Producer extends Thread{
         
	        @Override
	        public void run() {
	            produce();
	        }
	          
	        private void produce() {	             
	                try {
	                	   lock.lock();
	                       System.out.println("我拿到锁"+this.currentThread().getName());
	                        condition.signalAll();	                         
	                    System.out.println("我发出了一个信号："+this.currentThread().getName());
	                } finally{
	                    lock.unlock();
	                }
	            }
	 }
	    
}

运行结果：

Condition的执行方式，是当在线程Consumer中调用await方法后，线程Consumer将释放锁，并且将自己沉睡，等待唤醒，线程Producer获取到锁后，开始做事，完毕后，调用Condition的signalall方法，唤醒线程Consumer，线程Consumer恢复执行。

以上说明Condition是一个多线程间协调通信的工具类，使得某个，或者某些线程一起等待某个条件（Condition）,只有当该条件具备( signal 或者 signalAll方法被带调用)时 ，这些等待线程才会被唤醒，从而重新争夺锁。

Condition实现生产者、消费者模式 Demo：

package thread;
 
import java.util.PriorityQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
 
public class ConTest2 {
	    private int queueSize = 10;
	    private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);
	    private Lock lock = new ReentrantLock();
	    private Condition notFull = lock.newCondition();
	    private Condition notEmpty = lock.newCondition();
	     
	    public static void main(String[] args) throws InterruptedException  {
	    	ConTest2 test = new ConTest2();
	        Producer producer = test.new Producer();
	        Consumer consumer = test.new Consumer();	          
	        producer.start();
	        consumer.start();
	        Thread.sleep(0);
	        producer.interrupt();
	        consumer.interrupt();
	    }
	      
	    class Consumer extends Thread{	          
	        @Override
	        public void run() {
	            consume();
	        }
	        volatile boolean flag=true;  
	        private void consume() {
	            while(flag){
	                lock.lock();
	                try {
	                    while(queue.isEmpty()){
	                        try {
	                            System.out.println("队列空，等待数据");
	                            notEmpty.await();
	                        } catch (InterruptedException e) {	                          
	                            flag =false;
	                        }
	                    }
	                    queue.poll();                //每次移走队首元素
	                    notFull.signal();
	                    System.out.println("从队列取走一个元素，队列剩余"+queue.size()+"个元素");
	                } finally{
	                    lock.unlock();
	                }
	            }
	        }
	    }
	      
	    class Producer extends Thread{	          
	        @Override
	        public void run() {
	            produce();
	        }
	        volatile boolean flag=true;  
	        private void produce() {
	            while(flag){
	                lock.lock();
	                try {
	                    while(queue.size() == queueSize){
	                        try {
	                            System.out.println("队列满，等待有空余空间");
	                            notFull.await();
	                        } catch (InterruptedException e) {
	                            
	                            flag =false;
	                        }
	                    }
	                    queue.offer(1);        //每次插入一个元素
	                    notEmpty.signal();
	                    System.out.println("向队列取中插入一个元素，队列剩余空间："+(queueSize-queue.size()));
	                } finally{
	                    lock.unlock();
	                }
	            }
	        }
	    }
	}

运行结果如下：

Condition接口

  condition可以通俗的理解为条件队列。当一个线程在调用了await方法以后，直到线程等待的某个条件为真的时候才会被唤醒。这种方式为线程提供了更加简单的等待/通知模式。Condition必须要配合锁一起使用，因为对共享状态变量的访问发生在多线程环境下。一个Condition的实例必须与一个Lock绑定，因此Condition一般都是作为Lock的内部实现。

await() ：造成当前线程在接到信号或被中断之前一直处于等待状态。
await(long time, TimeUnit unit) ：造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态
awaitNanos(long nanosTimeout) ：造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。返回值表示剩余时间，如果在nanosTimesout之前唤醒，那么返回值 = nanosTimeout - 消耗时间，如果返回值 <= 0 ,则可以认定它已经超时了。
awaitUninterruptibly() ：造成当前线程在接到信号之前一直处于等待状态。【注意：该方法对中断不敏感】。
awaitUntil(Date deadline) ：造成当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态。如果没有到指定时间就被通知，则返回true，否则表示到了指定时间，返回返回false。
signal() ：唤醒一个等待线程。该线程从等待方法返回前必须获得与Condition相关的锁。
signal()All ：唤醒所有等待线程。能够从等待方法返回的线程必须获得与Condition相关的锁。

condition实现分析

• Condition接口包含了多种await方式和两个通知方法
• ConditionObject实现了Condition接口，是AbstractQueuedSynchronizer的内部类（因为Condition的操作都需要获取想关联的锁）
• Reentrantlock的newCondition方法返回与某个lock实例相关的Condition对象

public abstract class AbstractQueuedLongSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {

结合上面的类图，我们看到condition实现是依赖于aqs，而aqs是个抽象类。里面定义了同步器的基本框架，实现了基本的结构功能。只留有状态条件的维护由具体同步器根据具体场景来定制，如常见的 ReentrantLock 、 RetrantReadWriteLock和CountDownLatch 等等，

等待队列

要想能够深入的掌握condition还是应该知道它的实现原理，现在我们一起来看看condiiton的源码。创建一个condition对象是通过lock.newCondition(),而这个方法实际上是会new出一个ConditionObject对象，该类是AQS（AQS的实现原理的文章）的一个内部类，有兴趣可以去看看。

前面我们说过，condition是要和lock配合使用的也就是condition和Lock是绑定在一起的，而lock的实现原理又依赖于AQS，自然而然ConditionObject作为AQS的一个内部类无可厚非。

我们知道在锁机制的实现上，AQS内部维护了一个同步队列，如果是独占式锁的话，所有获取锁失败的线程的尾插入到同步队列，同样的，condition内部也是使用同样的方式，内部维护了一个 等待队列，所有调用condition.await方法的线程会加入到等待队列中，并且线程状态转换为等待状态。另外注意到ConditionObject中有两个成员变量：

/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter;

这样我们就可以看出来ConditionObject通过持有等待队列的头尾指针来管理等待队列。主要注意的是Node类复用了在AQS中的Node类，其节点状态和相关属性可以去看AQS的实现原理的文章，如果您仔细看完这篇文章对condition的理解易如反掌，对lock体系的实现也会有一个质的提升。Node类有这样一个属性：

//后继节点
Node nextWaiter;

进一步说明，等待队列是一个单向队列，而在之前说AQS时知道同步队列是一个双向队列。接下来我们用一个demo，通过debug进去看是不是符合我们的猜想：

public static void main(String[] args) {
   for (int i = 0; i < 10; i++) {
       Thread thread = new Thread(() -> {
           lock.lock();
           try {
               condition.await();
           } catch (InterruptedException e) {
               e.printStackTrace();
           }finally {
               lock.unlock();
           }
       });
       thread.start();
   }
}

这段代码没有任何实际意义，甚至很臭，只是想说明下我们刚才所想的。新建了10个线程，没有线程先获取锁，然后调用condition.await方法释放锁将当前线程加入到等待队列中，通过debug控制当走到第10个线程的时候查看firstWaiter即等待队列中的头结点，debug模式下情景图如下：

从这个图我们可以很清楚的看到这样几点：

1、调用condition.await方法后线程依次尾插入到等待队列中，如图队列中的线程引用依次为Thread-0,Thread-1,Thread-2…Thread-8；

2、等待队列是一个单向队列。通过我们的猜想然后进行实验验证，我们可以得出等待队列的示意图如下图所示：

同时还有一点需要注意的是：我们可以多次调用lock.newCondition()方法创建多个condition对象，也就是一个lock可以持有多个等待队列。而在之前利用Object的方式实际上是指在对象Object对象监视器上只能拥有一个同步队列和一个等待队列，而并发包中的Lock拥有一个同步队列和多个等待队列。示意图如下：

如图所示，ConditionObject是AQS的内部类，因此每个ConditionObject能够访问到AQS提供的方法，相当于每个Condition都拥有所属同步器的引用。

await实现原理

当调用condition.await()方法后会使得当前获取lock的线程进入到等待队列，如果该线程能够从await()方法返回的话一定是该线程获取了与condition相关联的lock。接下来，我们还是从源码的角度去看，只有熟悉了源码的逻辑我们的理解才是最深的。await()方法源码为：

public final void await() throws InterruptedException {
   if (Thread.interrupted())
       throw new InterruptedException();
 // 1. 将当前线程包装成Node，尾插入到等待队列中
   Node node = addConditionWaiter();
 // 2. 释放当前线程所占用的lock，在释放的过程中会唤醒同步队列中的下一个节点
   int savedState = fullyRelease(node);
   int interruptMode = 0;
   while (!isOnSyncQueue(node)) {
   // 3. 当前线程进入到等待状态
       LockSupport.park(this);
       if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
           break;
   }
 // 4. 自旋等待获取到同步状态（即获取到lock）
   if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
       interruptMode = REINTERRUPT;
   if (node.nextWaiter != null) // clean up if cancelled
       unlinkCancelledWaiters();
 // 5. 处理被中断的情况
   if (interruptMode != 0)
       reportInterruptAfterWait(interruptMode);
}

代码的主要逻辑请看注释，我们都知道当当前线程调用condition.await()方法后，会使得当前线程释放lock然后加入到等待队列中，直至被signal/signalAll后会使得当前线程从等待队列中移至到同步队列中去，直到获得了lock后才会从await方法返回，或者在等待时被中断会做中断处理。

那么关于这个实现过程我们会有这样几个问题：1. 是怎样将当前线程添加到等待队列中去的？2.释放锁的过程？3.怎样才能从await方法退出？而这段代码的逻辑就是告诉我们这三个问题的答案。

具体请看注释，在第1步中调用addConditionWaiter将当前线程添加到等待队列中，该方法源码为：

private Node addConditionWaiter() {
   Node t = lastWaiter;
   // If lastWaiter is cancelled, clean out.
   if (t != null && t.waitStatus != Node.CONDITION) {
       unlinkCancelledWaiters();
       t = lastWaiter;
   }
 //将当前线程包装成Node
   Node node = new Node(Thread.currentThread(), Node.CONDITION);
   if (t == null)
       firstWaiter = node;
   else
   //尾插入
       t.nextWaiter = node;
 //更新lastWaiter
   lastWaiter = node;
   return node;
}

这段代码就很容易理解了，将当前节点包装成Node，如果等待队列的firstWaiter为null的话（等待队列为空队列），则将firstWaiter指向当前的Node,否则，更新lastWaiter(尾节点)即可。

就是通过尾插入的方式将当前线程封装的Node插入到等待队列中即可，同时可以看出等待队列是一个不带头结点的链式队列，之前我们学习AQS时知道同步队列是一个带头结点的链式队列，这是两者的一个区别。

将当前节点插入到等待对列之后，会使当前线程释放lock，由fullyRelease方法实现，fullyRelease源码为：

final int fullyRelease(Node node) {
   boolean failed = true;
   try {
       int savedState = getState();
       if (release(savedState)) {
     //成功释放同步状态
           failed = false;
           return savedState;
       } else {
     //不成功释放同步状态抛出异常
           throw new IllegalMonitorStateException();
       }
   } finally {
       if (failed)
           node.waitStatus = Node.CANCELLED;
   }
}

这段代码就很容易理解了，调用AQS的模板方法release方法释放AQS的同步状态并且唤醒在同步队列中头结点的后继节点引用的线程，如果释放成功则正常返回，若失败的话就抛出异常。

到目前为止，这两段代码已经解决了前面的两个问题的答案了，还剩下第三个问题，怎样从await方法退出？现在回过头再来看await方法有这样一段逻辑：

while (!isOnSyncQueue(node)) {
 // 3. 当前线程进入到等待状态
   LockSupport.park(this);
   if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
       break;
}

很显然，当线程第一次调用condition.await()方法时，会进入到这个while()循环中，然后通过LockSupport.park(this)方法使得当前线程进入等待状态，那么要想退出这个await方法第一个前提条件自然而然的是要先退出这个while循环，出口就只剩下两个地方：1. 逻辑走到break退出while循环；2. while循环中的逻辑判断为false。

再看代码出现第1种情况的条件是当前等待的线程被中断后代码会走到break退出，第二种情况是当前节点被移动到了同步队列中（即另外线程调用的condition的signal或者signalAll方法），while中逻辑判断为false后结束while循环。

总结下，就是当前线程被中断或者调用condition.signal/condition.signalAll方法当前节点移动到了同步队列后 ，这是当前线程退出await方法的前提条件。当退出while循环后就会调用acquireQueued(node, savedState)，这个方法在介绍AQS的底层实现时说过了，若感兴趣的话可以去看这篇文章，该方法的作用是在自旋过程中线程不断尝试获取同步状态，直至成功（线程获取到lock）。

这样也说明了退出await方法必须是已经获得了condition引用（关联）的lock。到目前为止，开头的三个问题我们通过阅读源码的方式已经完全找到了答案，也对await方法的理解加深。await方法示意图如下图：

如图，调用condition.await方法的线程必须是已经获得了lock，也就是当前线程是同步队列中的头结点。调用该方法后会使得当前线程所封装的Node尾插入到等待队列中。
condition还额外支持了超时机制，使用者可调用方法awaitNanos,awaitUtil。这两个方法的实现原理，基本上与AQS中的tryAcquire方法如出一辙

要想不响应中断可以调用condition.awaitUninterruptibly()方法，该方法的源码为：

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();
}

这段方法与上面的await方法基本一致，只不过减少了对中断的处理，并省略了reportInterruptAfterWait方法抛被中断的异常。

signal/signalAll实现原理

调用condition的signal或者signalAll方法可以将等待队列中等待时间最长的节点移动到同步队列中，使得该节点能够有机会获得lock。

按照等待队列是先进先出（FIFO）的，所以等待队列的头节点必然会是等待时间最长的节点，也就是每次调用condition的signal方法是将头节点移动到同步队列中。我们来通过看源码的方式来看这样的猜想是不是对的，signal方法源码为：

public final void signal() {
   //1. 先检测当前线程是否已经获取lock
   if (!isHeldExclusively())
       throw new IllegalMonitorStateException();
   //2. 获取等待队列中第一个节点，之后的操作都是针对这个节点
 Node first = firstWaiter;
   if (first != null)
       doSignal(first);
}

signal方法首先会检测当前线程是否已经获取lock，如果没有获取lock会直接抛出异常，如果获取的话再得到等待队列的头指针引用的节点，之后的操作的doSignal方法也是基于该节点。下面我们来看看doSignal方法做了些什么事情，doSignal方法源码为：

private void doSignal(Node first) {
   do {
       if ( (firstWaiter = first.nextWaiter) == null)
           lastWaiter = null;
   //1. 将头结点从等待队列中移除
       first.nextWaiter = null;
   //2. while中transferForSignal方法对头结点做真正的处理
   } while (!transferForSignal(first) &&
            (first = firstWaiter) != null);
}

具体逻辑请看注释，真正对头节点做处理的逻辑在transferForSignal放，该方法源码为：

final boolean transferForSignal(Node node) {
   /*
    * If cannot change waitStatus, the node has been cancelled.
    */
 //1. 更新状态为0
   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).
    */
 //2.将该节点移入到同步队列中去
   Node p = enq(node);
   int ws = p.waitStatus;
   if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
       LockSupport.unpark(node.thread);
   return true;
}

关键逻辑请看注释，这段代码主要做了两件事情1.将头结点的状态更改为CONDITION；2.调用enq方法，将该节点尾插入到同步队列中，关于enq方法请看AQS的底层实现这篇文章。

现在我们可以得出结论：调用condition的signal的前提条件是当前线程已经获取了lock，该方法会使得等待队列中的头节点即等待时间最长的那个节点移入到同步队列，而移入到同步队列后才有机会使得等待线程被唤醒，即从await方法中的LockSupport.park(this)方法中返回，从而才有机会使得调用await方法的线程成功退出。signal执行示意图如下图：

sigllAll与sigal方法的区别体现在doSignalAll方法上，前面我们已经知道doSignal方法只会对等待队列的头节点进行操作，，而doSignalAll的源码为：

private void doSignalAll(Node first) {
   lastWaiter = firstWaiter = null;
   do {
       Node next = first.nextWaiter;
       first.nextWaiter = null;
       transferForSignal(first);
       first = next;
   } while (first != null);
}

该方法只不过时间等待队列中的每一个节点都移入到同步队列中，即“通知”当前调用condition.await()方法的每一个线程。

await与signal/signalAll的结合思考

文章开篇提到等待/通知机制，通过使用condition提供的await和signal/signalAll方法就可以实现这种机制，而这种机制能够解决最经典的问题就是“生产者与消费者问题”，关于“生产者消费者问题”之后会用单独的一篇文章进行讲解，这也是面试的高频考点。

await和signal和signalAll方法就像一个开关控制着线程A（等待方）和线程B（通知方）。它们之间的关系可以用下面一个图来表现得更加贴切：

如图，线程awaitThread先通过lock.lock()方法获取锁成功后调用了condition.await方法进入等待队列，而另一个线程signalThread通过lock.lock()方法获取锁成功后调用了condition.signal或者signalAll方法，使得线程awaitThread能够有机会移入到同步队列中，当其他线程释放lock后使得线程awaitThread能够有机会获取lock，从而使得线程awaitThread能够从await方法中退出执行后续操作。如果awaitThread获取lock失败会直接进入到同步队列。

简单的Demo

public class AwaitSignal {
   private static ReentrantLock lock = new ReentrantLock();
   private static Condition condition = lock.newCondition();
   private static volatile boolean flag = false;

   public static void main(String[] args) {
       Thread waiter = new Thread(new waiter());
       waiter.start();
       Thread signaler = new Thread(new signaler());
       signaler.start();
   }

   static class waiter implements Runnable {

       @Override
       public void run() {
           lock.lock();
           try {
               while (!flag) {
                   System.out.println(Thread.currentThread().getName() + "当前条件不满足等待");
                   try {
                       condition.await();
                   } catch (InterruptedException e) {
                       e.printStackTrace();
                   }
               }
               System.out.println(Thread.currentThread().getName() + "接收到通知条件满足");
           } finally {
               lock.unlock();
           }
       }
   }

   static class signaler implements Runnable {

       @Override
       public void run() {
           lock.lock();
           try {
               flag = true;
               condition.signalAll();
           } finally {
               lock.unlock();
           }
       }
   }
}

输出结果为：

Thread-0当前条件不满足等待
Thread-0接收到通知，条件满足

开启了两个线程waiter和signaler，waiter线程开始执行的时候由于条件不满足，执行condition.await方法使该线程进入等待状态同时释放锁，signaler线程获取到锁之后更改条件，并通知所有的等待线程后释放锁。这时，waiter线程获取到锁，并由于signaler线程更改了条件此时相对于waiter来说条件满足，继续执行。

参考文献：
https://blog.csdn/bohu83/article/details/51098106
https://www.javazhiyin/6274.html

本文标签： 示例Javacondition