Java高并发底层原理（十）—— Condition 如何实现线程等待与通知

上一章讨论了ReentrantLock如何实现互斥同一时刻只有一个线程能够进入临界区。但互斥只解决“谁能进去”的问题并不保证线程进去以后业务条件一定已经满足。以消费者从队列中取数据为例lock.lock();try{if(queue.isEmpty()){// 当前没有数据无法继续消费}returnqueue.remove();}finally{lock.unlock();}消费者已经获得了锁但队列可能仍然为空。它不能直接从空队列中取数据也不能一直持有锁等待因为生产者需要获得同一把锁才能向队列中添加数据。Condition解决的正是这个问题线程已经进入过临界区但业务条件暂时不满足时应该释放锁并等待等其他线程改变条件后再被通知回来重新检查。一、为什么有了锁还需要 Condition假设生产者和消费者共同操作一个队列privatefinalQueueStringqueuenewArrayDeque();privatefinalReentrantLocklocknewReentrantLock();消费者只有在队列不为空时才能取出数据publicStringtake(){lock.lock();try{returnqueue.remove();}finally{lock.unlock();}}如果队列为空queue.remove()会失败。消费者必须等待生产者放入数据。一种错误做法是在持有锁时不断检查lock.lock();try{while(queue.isEmpty()){// 一直等待}returnqueue.remove();}finally{lock.unlock();}问题在于消费者始终占着lock生产者无法进入临界区也就无法添加数据。消费者在等生产者改变队列生产者又在等消费者释放锁程序无法继续。所以等待业务条件时不能只“等”还必须先让出锁。Condition提供的就是这组动作条件不满足时释放锁并等待条件可能变化后再回来重新竞争锁。二、Condition 是什么Condition由某一把ReentrantLock创建privatefinalReentrantLocklocknewReentrantLock();privatefinalConditionnotEmptylock.newCondition();这里的notEmpty表示一个业务条件队列不为空。消费者在条件不满足时调用notEmpty.await();生产者在添加数据后调用notEmpty.signal();两者可以先这样理解方法作用await()当前线程释放锁并等待条件变化signal()通知一个等待线程条件可能已经变化可以回来重新检查Condition不是一把新的锁。真正负责互斥的仍然是ReentrantLockCondition只负责管理“已经获得过锁但业务条件暂时不满足”的线程。三、await 的基本语义释放锁并等待消费者的标准写法是publicStringtake()throwsInterruptedException{lock.lock();try{while(queue.isEmpty()){notEmpty.await();}returnqueue.remove();}finally{lock.unlock();}}当队列为空时消费者调用notEmpty.await()。这个方法可以先拆成三件事理解当前线程进入notEmpty对应的条件等待队列释放当前持有的lock然后暂停执行。这意味着消费者不会继续占着锁空转。它停在await()中以后生产者就有机会获得同一把lock向队列中添加数据。因此下面这段代码虽然写成while但它不是忙循环while(queue.isEmpty()){notEmpty.await();}每次循环只是在重新获得锁之后检查一次条件如果条件仍然不满足线程会再次释放锁并进入等待。四、await 为什么必须在持锁状态下调用await()不能脱离锁单独调用notEmpty.await();// 错误用法正确写法必须先获得与这个Condition绑定的锁lock.lock();try{while(queue.isEmpty()){notEmpty.await();}}finally{lock.unlock();}原因是检查条件和进入等待之间不能留下空隙。假设消费者没有持锁它先检查到队列为空正准备等待此时生产者添加了数据并调用signal()但消费者还没有真正进入条件队列这次通知不会被长期保存。消费者随后再执行await()就可能错过已经发生的通知。使用同一把锁后生产者无法插入到消费者“检查条件”和“进入条件队列”之间。消费者持锁检查到队列为空后await()会把“进入条件队列”和“释放锁”衔接成一个受保护的过程生产者只有在消费者释放锁之后才能添加数据并发送通知。所以await()必须在持有对应锁时调用。如果线程没有持有这把锁直接调用await()会抛出IllegalMonitorStateException。五、signal 的基本语义通知后还要重新竞争锁生产者的代码通常写成publicvoidput(Stringvalue){lock.lock();try{queue.add(value);notEmpty.signal();}finally{lock.unlock();}}signal()的含义不是把锁直接交给消费者而是通知一个正在等待的消费者条件可能已经变化可以准备回来重新检查。生产者调用signal()时仍然持有lock所以消费者被通知后不能立即继续执行。只有生产者执行unlock()之后消费者才有机会重新竞争这把锁等它重新获得锁以后await()才会真正返回。这里还有一个顺序要求应当先修改条件依赖的数据再发送通知。queue.add(value);notEmpty.signal();这个顺序表达的是先让条件从“不满足”变成“可能满足”再通知等待线程。虽然在同一个临界区里先signal()再添加数据也不会让消费者立刻越过生产者读取队列但这种写法不符合语义如果后续的数据修改抛出异常等待线程就可能收到一次无效通知。更不能把数据修改放在锁外。否则消费者可能被通知后先获得锁发现队列仍然为空于是再次等待生产者随后才添加数据却没有再次通知程序仍然可能停住。六、为什么等待条件必须用 while消费者应该使用while(queue.isEmpty()){notEmpty.await();}而不是if(queue.isEmpty()){notEmpty.await();}原因是收到通知只代表条件可能发生变化不代表线程重新获得锁时条件仍然满足。例如队列中只新增了一个元素但有两个消费者先后从等待中返回。第一个消费者获得锁后取走这个元素第二个消费者随后获得锁时队列已经再次为空。如果第二个消费者使用if它会从await()返回后直接执行queue.remove()如果使用while它会重新检查queue.isEmpty()发现条件仍然不满足后再次等待。即使只调用signal()也不能把“被通知”理解成“条件一定满足”。默认非公平的ReentrantLock还可能让其他线程先获得锁并改变队列状态。因此while表达的是一个更稳固的规则每次从await()返回后都重新检查业务条件直到条件确实满足。七、Condition 条件队列和锁等待队列有什么区别ReentrantLock和Condition会涉及两类等待队列。队列等待的事情典型来源锁等待队列等待获得ReentrantLock调用lock()但锁被占用条件队列等待业务条件变化已经持有锁但调用了await()消费者调用await()后会从持锁执行状态进入Condition条件队列并释放lock。生产者调用signal()后被通知的消费者不会直接执行临界区代码而是从条件队列转移到锁等待队列重新竞争同一把ReentrantLock。这一步是理解Condition的关键图中的Sync Queue就是 AQS 中用于竞争锁的等待队列。线程从await()返回时已经重新持有了lock所以外层finally中可以安全执行unlock()。八、为什么一把锁可以创建多个 Condition一个有界队列同时存在两个不同的业务条件条件谁在等待什么时候通知队列不为空消费者生产者添加数据后队列不满生产者消费者取走数据后因此可以从同一把锁创建两个ConditionprivatefinalReentrantLocklocknewReentrantLock();privatefinalConditionnotEmptylock.newCondition();privatefinalConditionnotFulllock.newCondition();消费者在队列为空时等待notEmpty生产者在队列已满时等待notFull。生产者添加数据后只需要通知等待数据的消费者消费者取走数据后只需要通知等待空间的生产者。如果所有线程都混在同一个条件队列中生产者添加数据后可能唤醒另一个生产者但“队列不为空”和“队列不满”是两种不同条件。多个Condition的价值就在于让等待队列按业务条件拆分从而减少无效唤醒。九、signal 和 signalAll 有什么区别signal()通知当前Condition上的一个等待线程notEmpty.signal();如果生产者只添加了一个元素通常只需要唤醒一个消费者。唤醒更多消费者只会让它们一起转入锁等待队列最终仍然只有一个线程取到数据其他线程重新检查后还要再次等待。signalAll()会通知当前Condition上的所有等待线程notEmpty.signalAll();它的成本更高因为所有等待线程都会重新竞争锁并在获得锁后重新检查条件。但signal()也有使用前提同一个Condition中的线程等待的是同一种业务条件并且一次状态变化只需要一个线程继续执行。如果条件变化可能让多个线程都能继续或者无法准确判断应该唤醒哪一个线程就应考虑使用signalAll()。可以简单记成signal()更精确成本更低signalAll()更保守成本更高但在条件复杂时更不容易漏掉应该继续执行的线程。十、await 为什么支持中断线程执行await()后可能长时间等待业务条件满足。如果程序准备关闭之后不会再有生产者添加数据等待线程就需要一种退出方式。另一个线程可以调用consumerThread.interrupt();如果消费者正在await()它会停止等待并在重新获得lock后抛出InterruptedExceptionpublicStringtake(){lock.lock();try{while(queue.isEmpty()){notEmpty.await();}returnqueue.remove();}catch(InterruptedExceptione){Thread.currentThread().interrupt();returnnull;}finally{lock.unlock();}}这里要注意两点。第一中断不是强制杀死线程而是请求等待线程结束当前等待线程自己的代码决定如何处理这个请求。第二无论await()是因为signal()正常返回还是因为中断抛出异常它离开await()前都会重新获得lock因此finally中的unlock()仍然是安全的。十一、一个简单的阻塞队列把前面的规则组合起来可以实现一个简化版有界阻塞队列publicclassSimpleBlockingQueueE{privatefinalQueueEqueuenewArrayDeque();privatefinalintcapacity;privatefinalReentrantLocklocknewReentrantLock();privatefinalConditionnotEmptylock.newCondition();privatefinalConditionnotFulllock.newCondition();publicSimpleBlockingQueue(intcapacity){if(capacity0){thrownewIllegalArgumentException(capacity must be greater than 0);}this.capacitycapacity;}publicvoidput(Evalue)throwsInterruptedException{lock.lock();try{while(queue.size()capacity){notFull.await();}queue.add(value);notEmpty.signal();}finally{lock.unlock();}}publicEtake()throwsInterruptedException{lock.lock();try{while(queue.isEmpty()){notEmpty.await();}Evaluequeue.remove();notFull.signal();returnvalue;}finally{lock.unlock();}}}这段代码里ReentrantLock负责保护queue的互斥访问notEmpty管理等待数据的消费者notFull管理等待空间的生产者。每次等待都使用while重新检查条件每次状态变化后只通知对应条件队列中的线程。本章总结Condition的完整因果链可以从“线程不能长期占着锁等待业务条件”开始理解消费者拿到锁后发现队列为空如果继续持锁等待生产者就无法进入临界区添加数据因此等待条件必须同时包含释放锁这一动作。await()把线程从“持锁执行”转入“条件等待”并把锁让给能够改变条件的线程。生产者获得锁后修改共享状态再通过signal()把等待线程从条件队列转入锁等待队列。被通知的线程并不直接继续执行而是重新竞争同一把锁只有重新持有锁后await()才返回。这条链路决定了几个写法上的约束await()必须在持锁状态下调用因为检查条件和进入等待不能被其他线程插入signal()应该发生在共享状态改变之后因为通知必须对应一次真实的条件变化等待条件必须写成while因为通知只表示“可能满足”不保证线程重新获得锁时条件仍然成立。所以Condition不是独立于锁之外的通知器而是ReentrantLock的条件等待层。它把“释放锁、等待条件、收到通知、重新竞争锁、再次检查条件”串成一个闭环让线程能够在不长期占用锁的情况下等待业务状态变化。