Condition.signal 之后还要争锁：用可关闭通道验证等待条件

signal已经执行，等待线程仍然没有继续，这不一定是信号丢失。等待者还需要重新取得同一把锁；发信号的线程若仍然持锁，接收者就不能进入受保护区。把signal理解成“立即把执行权交给另一线程”，会让并发日志顺序看起来十分反常。

更关键的是，线程被唤醒不代表它等待的业务条件已经成立。signal只是一次通知，真正决定是否继续的是共享状态。所以正确等待通常写成while检查条件，而不是if后面只await一次。

先发一条没有改变状态的通知

下面程序故意先signal而不设置ready。等待者醒来后检查仍为false，重新等待；第二次设置ready并signal时，主线程暂时保持锁，确认Future还没有完成，随后释放锁，等待者才返回。

代码还给出容量为一的Channel，使用notEmpty与notFull分开等待。Java 17保存为ConditionChannel.java后运行java ConditionChannel.java：

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.*;
public class ConditionChannel {
    static final class Channel {
        private final ReentrantLock lock=new ReentrantLock();
        private final Condition notEmpty=lock.newCondition(),notFull=lock.newCondition();
        private final ArrayDeque<String> values=new ArrayDeque<>();
        private boolean closed;
        void put(String value)throws InterruptedException {
            Objects.requireNonNull(value);
            lock.lockInterruptibly();
            try {
                while(values.size()==1&&!closed)notFull.await();
                if(closed)throw new IllegalStateException("channel closed");
                values.addLast(value);notEmpty.signal();
            }finally{lock.unlock();}
        }
        Optional<String> take()throws InterruptedException {
            lock.lockInterruptibly();
            try {
                while(values.isEmpty()&&!closed)notEmpty.await();
                if(values.isEmpty())return Optional.empty();
                String result=values.removeFirst();notFull.signal();
                return Optional.of(result);
            }finally{lock.unlock();}
        }
        void close() {
            lock.lock();
            try{closed=true;notEmpty.signalAll();notFull.signalAll();}
            finally{lock.unlock();}
        }
    }
    static void await(CountDownLatch latch)throws InterruptedException {
        if(!latch.await(3,TimeUnit.SECONDS))throw new IllegalStateException("coordination timeout");
    }
    public static void main(String[] args)throws Exception {
        ExecutorService worker=Executors.newSingleThreadExecutor();
        ReentrantLock lock=new ReentrantLock();Condition changed=lock.newCondition();
        boolean[] ready={false};AtomicInteger waits=new AtomicInteger();
        CountDownLatch started=new CountDownLatch(1),rewaited=new CountDownLatch(1);
        try {
            Future<String> waiting=worker.submit(()->{
                lock.lockInterruptibly();
                try {
                    while(!ready[0]) {
                        if(waits.incrementAndGet()==1)started.countDown();else rewaited.countDown();
                        changed.await();
                    }
                    return "ready";
                }finally{lock.unlock();}
            });
            await(started);
            lock.lock();try{changed.signal();}finally{lock.unlock();}
            await(rewaited);
            lock.lock();
            try{ready[0]=true;changed.signal();System.out.println("signalled while holding lock, done="+waiting.isDone());}
            finally{lock.unlock();}
            System.out.println("waiter="+waiting.get(3,TimeUnit.SECONDS)+", waited again="+(waits.get()>=2));
            Channel channel=new Channel();channel.put("A");
            CountDownLatch putting=new CountDownLatch(1);
            Future<?> second=worker.submit(()->{putting.countDown();channel.put("B");return null;});
            await(putting);
            try{second.get(40,TimeUnit.MILLISECONDS);throw new AssertionError("full channel should wait");}
            catch(TimeoutException expected){System.out.println("full channel=waiting");}
            System.out.println("take first="+channel.take().orElseThrow());
            second.get(3,TimeUnit.SECONDS);channel.close();
            System.out.println("drain after close="+channel.take().orElseThrow());
            System.out.println("closed and empty="+channel.take().isEmpty());
            try{channel.put("C");throw new AssertionError("closed put");}
            catch(IllegalStateException expected){System.out.println(expected.getMessage());}
            Channel empty=new Channel();
            Future<Optional<String>> stop=worker.submit(empty::take);
            empty.close();
            System.out.println("empty close released consumer="+stop.get(3,TimeUnit.SECONDS).isEmpty());
        }finally{
            worker.shutdownNow();
            if(!worker.awaitTermination(3,TimeUnit.SECONDS))throw new IllegalStateException("worker active");
        }
    }
}

signalled while holding lock, done=false
waiter=ready, waited again=true
full channel=waiting
take first=A
drain after close=B
closed and empty=true
channel closed
empty close released consumer=true

第一项done=false由主线程仍然持有锁来保证，不依赖输出机器足够慢。waited again说明不满足条件的通知没有让等待者误闯后续逻辑。这个实验没有制造或声称测出了虚假唤醒，而是主动发送一条状态未改变的通知，验证同一套循环检查同样必要。

Condition允许虚假唤醒，其他等待者也可能在当前线程重新得到锁前先消费掉资源。即使业务代码保证“只有满足条件才signal”，也不能由这句话删掉while。通知不是一份资源预订，醒来的线程必须根据自己实际观察到的状态重新决定。

await 暂时交回锁，返回之前再取回来

若等待者拿着锁原地睡眠，生产者也就没法取得锁修改条件。await将等待与释放关联锁协调起来，结束等待时重新竞争锁，返回或抛相应中断异常之前遵守接口规定的锁重获过程。调用方仍要在finally释放自己持有的锁。

不能在没有持有关联锁时随意调用ReentrantLock产生的Condition进行await或signal。共享状态检查、修改和通知都放在同一把锁下，才不会形成“检查之后、真正开始等待之前”漏掉状态转换的窗口。单独把ready改成volatile，也不能替代这组等待协议。

程序中的计数器和门闩只是让测试看到等待经过了哪里，不承担Channel的业务同步。Channel只依赖它自己的锁和两个条件；如果把演示钩子、调试计数或sleep变成正确性所必需的步骤，就说明真实合同还没有建立好。

有界通道怎样表达满、空和结束

put在容量已满时等待notFull，take在没有元素时等待notEmpty。取走元素后通知生产者，放入元素后通知消费者。容量固定为一使两种路径都容易观察，实际通道可扩展成经过校验的容量参数，但仍需保持条件与状态对应。

关闭采用“停止新写，允许排空”的政策。close以后put拒绝，已有B仍可被取走；关闭且为空才返回Optional.empty。它没有用null充当正常数据，因此消费者可以明确区分拿到内容与流已结束。若产品需要立即丢弃剩余数据，那是另一项关闭政策，不能在这个实现上悄悄改掉。

close对两个条件都signalAll，因为等待中的生产者需要知道无法再写，消费者也需要知道不会再来新元素。只叫醒一个角色或只叫醒一条线程，其他等待者可能再也没有后续通知。普通一次放入或取走只改变一个槽位，使用signal与这份状态变化相匹配。

示例验证正常背压、关闭后排空、关闭后拒绝写入和空通道关闭返回。最后一个消费者可能在close前或后进入take，两条次序都应得到结束；它不是在声称测试固定证明了消费者已经阻塞。要专门断言“已阻塞者被唤醒”，还需像前半段一样设置受控等待观察点。

中断与超时必须和状态检查放在一起设计

put与take允许等待被中断，调用者可以放弃本次操作。中断并不会自动关闭整个Channel，其他生产者和消费者仍按原状态运行。如果一个工作者决定退出，应由拥有通道生命周期的一方判断是否需要关闭，不能让任意一次取消都摧毁共享资源。

带超时的版本应循环计算剩余预算，例如围绕awaitNanos返回值继续检查条件，而不是每次唤醒又重新等待完整超时。等待预算还不能覆盖用户任意处理函数；拿到数据以后做长时间业务，是下一阶段的责任。

这一实现适合解释条件协议，普通生产者消费者代码可以优先采用现成有界队列。若使用自定义通道，关闭、排空、取消和确认都应成为可测试合同。signal负责让等待者有机会重查，业务状态负责决定能不能继续，两者缺一项，循环就可能在偶然调度下看似正常。