synchronized 与监视器：互斥、等待和唤醒要用同一套条件

import java.util.*;
import java.util.concurrent.*;

public class Main {
    static final class Buffer {
        final int[] values;
        int head, tail, size;
        boolean closed;
        Buffer(int capacity) {
            if (capacity<=0) throw new IllegalArgumentException();
            values=new int[capacity];
        }
        synchronized void put(int value) throws InterruptedException {
            while (size==values.length && !closed) wait(); // 唤醒后重新检查满与关闭条件。
            if (closed) throw new IllegalStateException("closed");
            values[tail]=value; tail=(tail+1)%values.length; size++;
            notifyAll(); // 数据增加，消费者可能可以继续。
        }
        synchronized OptionalInt take() throws InterruptedException {
            while (size==0 && !closed) wait();
            if (size==0) return OptionalInt.empty(); // 关闭且排空，返回明确结束标记。
            int result=values[head]; head=(head+1)%values.length; size--;
            notifyAll(); // 空间增加，生产者可能可以继续。
            return OptionalInt.of(result);
        }
        synchronized void close() { closed=true; notifyAll(); }
    }
    static void check(boolean ok) { if (!ok) throw new AssertionError(); }
    public static void main(String[] args) throws Exception {
        Buffer b=new Buffer(2); b.put(1); b.put(2); b.close();
        check(b.take().orElseThrow()==1 && b.take().orElseThrow()==2 && b.take().isEmpty());
        try { b.put(3); throw new AssertionError(); }
        catch (IllegalStateException expected) { }
        Buffer empty=new Buffer(1); empty.close(); empty.close(); check(empty.take().isEmpty());
        Buffer round=new Buffer(1); round.put(9); check(round.take().orElseThrow()==9);
        round.put(8); check(round.take().orElseThrow()==8); round.close();
        Buffer concurrent=new Buffer(3);
        ExecutorService executor=Executors.newFixedThreadPool(2);
        try {
            Future<?> producer=executor.submit(() -> {
                try { for(int i=0;i<100;i++) concurrent.put(i); }
                catch(InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(e); }
                finally { concurrent.close(); }
            });
            Future<Integer> consumer=executor.submit(() -> {
                int sum=0; OptionalInt item;
                while ((item=concurrent.take()).isPresent()) sum+=item.getAsInt();
                return sum;
            });
            producer.get(5,TimeUnit.SECONDS); check(consumer.get(5,TimeUnit.SECONDS)==4950);
        } finally { concurrent.close(); executor.shutdownNow(); }
        Buffer interrupted=new Buffer(1);
        Thread.currentThread().interrupt();
        try { interrupted.take(); throw new AssertionError(); }
        catch (InterruptedException expected) { check(!Thread.currentThread().isInterrupted()); }
        finally { Thread.interrupted(); interrupted.close(); }
        System.out.println("PASS 6");
    }
}