Java多线程（四）：等待唤醒机制与阻塞队列

前言

锁解决了“多个线程不能同时修改共享数据”的问题，但真实业务中，线程之间往往还需要配合。例如生产者准备好数据以后，消费者才能处理；队列已经满时，生产者应该暂停；队列为空时，消费者也不应该不断空转。

如果只使用死循环检查条件，程序会不断占用 CPU。Java 的等待唤醒机制允许线程在条件不满足时释放锁并进入等待，待其他线程改变条件后再被唤醒。本篇先使用 wait()、notify() 和 notifyAll() 实现生产者—消费者模型，再使用 BlockingQueue 简化代码。

一、生产者—消费者模型

生产者—消费者是一类经典的线程协作模型：

生产者负责产生数据。
消费者负责处理数据。
两者通过一个共享缓冲区传递数据。

假设桌子上最多只能放一碗面：

桌上没有面时，消费者等待，生产者制作。
桌上已经有面时，生产者等待，消费者取走。
任意时刻只能有一个线程操作桌子上的共享状态。

这里既有互斥要求，也有条件协作要求。只加 synchronized 能防止同时修改，却不能自动表达“没面就等、有面就叫醒消费者”。

二、wait、notify 与 notifyAll

这三个方法定义在 Object 类中，因为等待和唤醒是围绕某个对象的监视器锁进行的。

1. wait()

当前线程调用锁对象的 wait() 后会：

释放当前持有的该对象监视器锁。
进入这个对象对应的等待集合。
暂停执行，直到被通知、中断或超时。
被唤醒后重新竞争锁，获得锁后才从 wait() 返回。

“被唤醒”不等于“立刻继续运行”。它只是从等待集合回到竞争锁的队伍。

2. notify()

notify() 从当前对象的等待集合中唤醒一个线程，但不能精确指定是哪一个。存在多类等待者时，随机唤醒可能导致不合适的线程被选中。

3. notifyAll()

notifyAll() 唤醒在当前对象上等待的所有线程。它们会重新检查条件，再竞争锁。虽然可能产生更多竞争，但代码通常更稳妥，不容易因为唤醒了“错误类型”的线程而无法继续。

4. 必须先持有同一把锁

wait()、notify() 和 notifyAll() 必须在当前线程持有对应对象监视器锁时调用，否则会抛出 IllegalMonitorStateException。

synchronized (lock) {
    lock.wait();
    lock.notifyAll();
}

调用哪个对象的 wait()，就必须持有哪个对象的锁。锁对象与调用对象不一致同样是错误的。

三、为什么判断条件要使用 while

下面的写法看起来合理，却存在风险：

if (!condition) {
    lock.wait();
}

线程从 wait() 返回时，不能假设条件一定仍然成立。可能有多个消费者同时被唤醒，第一个消费者已经把数据取走，第二个消费者获得锁时缓冲区又空了；线程也可能出现虚假唤醒。

正确模式是使用 while 反复检查保护条件：

synchronized (lock) {
    while (!condition) {
        lock.wait();
    }
    // 条件此刻满足，执行操作
}

这类结构也叫保护性暂停：条件不满足就等待，被唤醒后重新判断，只有条件真正满足才继续。

四、手写单槽生产者—消费者

先定义共享缓冲区：

public class FoodBox {
    private String food;
    private boolean available;

    public synchronized void put(String newFood)
            throws InterruptedException {
        while (available) {
            wait();
        }

        food = newFood;
        available = true;
        System.out.println("生产：" + newFood);

        notifyAll();
    }

    public synchronized String take()
            throws InterruptedException {
        while (!available) {
            wait();
        }

        String result = food;
        food = null;
        available = false;
        System.out.println("消费：" + result);

        notifyAll();
        return result;
    }
}

put() 与 take() 都是同步方法，因此锁对象都是当前 FoodBox 实例。生产者发现已有食物就等待；消费者发现没有食物也等待。每次状态发生变化后调用 notifyAll()，通知另一方重新检查条件。

生产者任务：

public class Producer implements Runnable {
    private final FoodBox box;

    public Producer(FoodBox box) {
        this.box = box;
    }

    @Override
    public void run() {
        for (int i = 1; i <= 5; i++) {
            try {
                box.put("第 " + i + " 碗面");
                Thread.sleep(300);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
        }
    }
}

消费者任务：

public class Consumer implements Runnable {
    private final FoodBox box;

    public Consumer(FoodBox box) {
        this.box = box;
    }

    @Override
    public void run() {
        for (int i = 1; i <= 5; i++) {
            try {
                String food = box.take();
                System.out.println(Thread.currentThread().getName()
                        + " 吃掉了 " + food);
                Thread.sleep(500);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
        }
    }
}

测试类：

public class WaitNotifyDemo {
    public static void main(String[] args) {
        FoodBox box = new FoodBox();

        new Thread(new Producer(box), "厨师").start();
        new Thread(new Consumer(box), "顾客").start();
    }
}

注意，生产者和消费者必须拿到同一个 FoodBox 对象。如果各自创建一个缓冲区，它们操作的就不是同一份数据，也无法相互通知。

五、wait 与 sleep 的区别

两者都会让线程暂时停止执行，却属于完全不同的机制。

对比项	`wait()`	`sleep()`
所属类	`Object`	`Thread`
是否必须在同步区域调用	是	否
是否释放监视器锁	是	否
恢复方式	通知、中断或超时	时间到达或中断
典型用途	等待共享条件变化	延迟、节奏控制、模拟耗时

生产者发现缓冲区已满时，必须释放锁，消费者才能进入并取走数据，所以此处应该使用 wait()。如果换成 sleep()，生产者会抱着锁睡觉，消费者无法取得锁，程序反而可能一直停住。

六、传统等待唤醒代码的难点

手写 wait/notifyAll 有助于理解底层协作，但也容易犯错：

锁对象与调用 wait() 的对象不一致。
使用 if 而不是 while 检查条件。
修改状态后忘记通知其他线程。
在持锁期间执行过长的任务。
多生产者、多消费者时终止条件设计不完整。
捕获中断后继续循环，导致线程无法正常取消。

因此，实际开发中应优先考虑 JUC 提供的高层并发工具。对于生产者—消费者场景，BlockingQueue 就是非常合适的抽象。

七、BlockingQueue 阻塞队列

BlockingQueue 是一个支持阻塞操作的线程安全队列：

队列满时，put() 可以等待空间出现。
队列空时，take() 可以等待元素出现。
线程安全与等待通知逻辑由队列内部实现。

常见实现包括：

ArrayBlockingQueue：基于数组的有界阻塞队列，容量创建时确定。
LinkedBlockingQueue：基于链表，可指定容量；不指定时容量非常大。

阻塞队列常用操作分为四组：

操作目标	失败时抛异常	失败时返回特殊值	一直阻塞	限时阻塞
插入	`add(e)`	`offer(e)`	`put(e)`	`offer(e, time, unit)`
移除	`remove()`	`poll()`	`take()`	`poll(time, unit)`
查看队首	`element()`	`peek()`	—	—

不能把这些方法随意混用。例如需要“队列满了就等待”时应使用 put()，使用 add() 只会在队列满时抛出异常。

八、使用阻塞队列重写案例

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueDemo {
    public static void main(String[] args) {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);

        Thread producer = new Thread(() -> {
            for (int i = 1; i <= 5; i++) {
                try {
                    String food = "第 " + i + " 碗面";
                    queue.put(food);
                    System.out.println("生产：" + food);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        }, "厨师");

        Thread consumer = new Thread(() -> {
            for (int i = 1; i <= 5; i++) {
                try {
                    String food = queue.take();
                    System.out.println("消费：" + food);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        }, "顾客");

        producer.start();
        consumer.start();
    }
}

与手写版本相比，共享变量、条件判断、wait() 和 notifyAll() 都消失了。我们只需要说明队列容量，并根据需要调用 put() 和 take()。这不是省略了线程协作，而是把正确的协作规则封装到了队列内部。

九、如何让生产者和消费者正确结束

演示程序知道一共生产 5 个元素，所以消费者循环 5 次即可。真实业务中，消费者往往不知道生产者最终会产生多少数据，常见做法是放入一个特殊结束标记，也称“毒丸”。

private static final String END = "__END__";

生产完成后把 END 放入队列，消费者取到它就退出。但多个消费者需要几个结束标记、结束标记会不会与正常数据冲突，都要在协议中明确。另一种方式是结合关闭标记、超时 poll()、任务取消或线程池生命周期管理。

这说明并发程序不仅要设计“如何运行”，还要设计“如何停止”。忽略终止协议，程序可能在任务已经完成后仍有线程永久阻塞。

十、本篇小结

等待唤醒机制解决的是线程之间的条件协作。wait() 会释放当前对象的监视器锁并等待，notify() 唤醒一个等待线程，notifyAll() 唤醒所有等待线程。调用它们时必须持有同一个对象的锁，保护条件必须放在 while 中反复检查。

手写等待唤醒能够帮助理解原理，但生产者—消费者场景更推荐使用 BlockingQueue。put() 与 take() 已经封装了队列满、队列空时的阻塞逻辑，代码更短，也更不容易出错。

下一篇将把前面的知识放进课程中的综合练习：模拟抢红包、两个抽奖箱争抢奖池、统计各自结果，并通过 Callable 比较不同线程产生的最大奖项。

前言