Java 实例 – 生产者/消费者问题（长文解析）

Java 实例 – 生产者/消费者问题

在多线程编程的世界里，有一个经典问题被反复提及，它不仅是面试常客，更是理解线程同步机制的绝佳入口——生产者/消费者问题。这个问题模拟的是一个典型的协作场景：一个或多个生产者线程不断生成数据，而一个或多个消费者线程则从共享缓冲区中取出数据进行处理。如何保证数据不丢失、不重复、不越界？这正是 Java 实例 – 生产者/消费者问题的核心所在。

想象一下，你开了一家快递驿站。快递员（生产者）把包裹源源不断地送进来，而客户（消费者）则排队取走包裹。如果驿站没有管理好，就会出现两种尴尬：一是包裹堆满，驿站爆仓；二是客户来取时却发现没货，白白跑一趟。这种“生产过快”和“消费过慢”之间的协调问题，正是我们要解决的。

Java 提供了 synchronized 关键字和 wait() / notify() / notifyAll() 方法来实现线程间的协作。接下来，我们通过一个完整的 Java 实例，一步步实现这个经典模型。

共享缓冲区的设计与实现

共享缓冲区是生产者与消费者之间的“中转站”。它必须满足两个核心要求：一是线程安全，二是具备阻塞与唤醒的能力。我们使用一个固定大小的数组作为缓冲区，配合一个计数器来跟踪当前已存放的数据量。

public class BoundedBuffer {
    private final int[] buffer;
    private int count; // 当前缓冲区中的元素数量
    private int putIndex; // 下一个写入位置
    private int takeIndex; // 下一个读取位置

    // 构造函数：初始化缓冲区大小
    public BoundedBuffer(int size) {
        buffer = new int[size];
        count = 0;
        putIndex = 0;
        takeIndex = 0;
    }

    // 生产者调用的方法：放入一个数据
    public synchronized void put(int value) throws InterruptedException {
        // 如果缓冲区已满，生产者线程进入等待状态
        while (count == buffer.length) {
            wait(); // 释放锁并进入等待队列
        }

        // 将数据放入缓冲区
        buffer[putIndex] = value;
        putIndex = (putIndex + 1) % buffer.length; // 循环写入
        count++;

        // 通知所有等待的消费者线程，有新数据可取
        notifyAll();
    }

    // 消费者调用的方法：取出一个数据
    public synchronized int take() throws InterruptedException {
        // 如果缓冲区为空，消费者线程进入等待状态
        while (count == 0) {
            wait(); // 释放锁并进入等待队列
        }

        // 从缓冲区取出数据
        int value = buffer[takeIndex];
        takeIndex = (takeIndex + 1) % buffer.length; // 循环读取
        count--;

        // 通知所有等待的生产者线程，有空位可放
        notifyAll();

        return value;
    }
}

关键点解析：

• synchronized 确保同一时刻只有一个线程能访问共享资源。
• wait() 会释放当前对象的锁，并将线程加入等待队列，直到被 notify() 或 notifyAll() 唤醒。
• notifyAll() 比 notify() 更安全，因为它能唤醒所有等待的线程，避免“唤醒错误线程”的问题。
• 使用 while 而非 if 判断条件，是因为线程被唤醒后可能因其他线程抢占而再次不满足条件（虚假唤醒）。

生产者线程的实现

生产者线程的任务是不断生成数据，并尝试将它们放入共享缓冲区。当缓冲区满时，它必须等待，直到有空间可用。

public class Producer implements Runnable {
    private final BoundedBuffer buffer;
    private final int max;

    public Producer(BoundedBuffer buffer, int max) {
        this.buffer = buffer;
        this.max = max;
    }

    @Override
    public void run() {
        for (int i = 1; i <= max; i++) {
            try {
                // 模拟生产过程：生成一个数字
                System.out.println("生产者：生产 " + i);
                buffer.put(i); // 尝试放入缓冲区
                Thread.sleep(100); // 模拟生产耗时
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("生产者被中断");
                break;
            }
        }
        System.out.println("生产者完成生产");
    }
}

说明：

• Thread.sleep(100) 模拟生产任务的耗时，让程序更接近真实场景。
• 使用 try-catch 捕获 InterruptedException，确保线程中断时能优雅退出。

消费者线程的实现

消费者线程负责从缓冲区中取出数据并处理。当缓冲区为空时，它会等待，直到有数据可用。

public class Consumer implements Runnable {
    private final BoundedBuffer buffer;

    public Consumer(BoundedBuffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        while (true) {
            try {
                int value = buffer.take(); // 从缓冲区取出数据
                System.out.println("消费者：消费 " + value);
                Thread.sleep(200); // 模拟处理耗时

                // 当生产者不再生产时，消费者可以退出
                if (value == 10) {
                    System.out.println("消费者：已消费完所有数据，停止工作");
                    break;
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("消费者被中断");
                break;
            }
        }
    }
}

说明：

• 消费者使用无限循环 while(true)，直到检测到最后一个数据（这里设为 10）才退出。
• Thread.sleep(200) 模拟消费处理过程，比生产慢，体现“消费滞后”的现实场景。

主程序与线程启动

现在我们编写主类，创建生产者和消费者线程，并启动它们。

public class ProducerConsumerDemo {
    public static void main(String[] args) {
        BoundedBuffer buffer = new BoundedBuffer(5); // 缓冲区大小为 5

        // 创建生产者和消费者线程
        Thread producer = new Thread(new Producer(buffer, 10));
        Thread consumer = new Thread(new Consumer(buffer));

        // 启动线程
        producer.start();
        consumer.start();

        // 等待线程执行完毕
        try {
            producer.join();
            consumer.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println("主线程被中断");
        }

        System.out.println("Java 实例 – 生产者/消费者问题：执行完成");
    }
}

运行说明：

• join() 方法让主线程等待子线程结束，确保程序完整执行。
• 缓冲区大小为 5，意味着最多能存放 5 个数据，生产者在满时会等待，消费者在空时会等待。

线程协作机制详解

为什么使用 wait() 和 notifyAll()？这背后是 Java 的监视器模型（Monitor Model）。每个对象都有一个内置锁（monitor），synchronized 块/方法会获取该锁。

• 当线程调用 wait()，它会释放锁并进入等待队列。
• notifyAll() 会唤醒所有在该对象上等待的线程，它们重新竞争锁。
• 一旦某个线程获得锁，它就会从 wait() 返回，继续执行。

这个机制确保了线程间不会“争抢”资源，而是有条不紊地协作。就像一个工厂的调度员：当流水线满时，生产员暂停；当有空位时，调度员通知他们继续。

运行结果示例

以下是程序可能的输出：

生产者：生产 1
消费者：消费 1
生产者：生产 2
消费者：消费 2
生产者：生产 3
消费者：消费 3
生产者：生产 4
消费者：消费 4
生产者：生产 5
消费者：消费 5
生产者：生产 6
消费者：消费 6
生产者：生产 7
消费者：消费 7
生产者：生产 8
消费者：消费 8
生产者：生产 9
消费者：消费 9
生产者：生产 10
消费者：消费 10
消费者：已消费完所有数据，停止工作
生产者完成生产
Java 实例 – 生产者/消费者问题：执行完成

观察要点：

• 生产和消费交替进行，没有冲突。
• 缓冲区大小为 5，当达到上限时，生产者会自动暂停，直到消费者取走数据。
• 所有数据都被正确消费，无丢失或重复。

常见问题与优化建议

进阶建议：在真实项目中，建议使用 java.util.concurrent.BlockingQueue 接口，如 ArrayBlockingQueue，它已经内置了阻塞和同步机制，代码更简洁、更安全。

总结

Java 实例 – 生产者/消费者问题 是多线程编程中的基石。通过本例，我们不仅掌握了 synchronized、wait()、notifyAll() 的使用方法，还深入理解了线程协作的底层逻辑。从一个简单的缓冲区设计，到生产者与消费者的协调运作，每一个细节都体现了并发编程的严谨与智慧。

对于初学者来说，这个例子是理解“共享资源”与“线程安全”的绝佳起点；对于中级开发者，它提醒我们：在并发场景下，逻辑正确性与执行效率同样重要。

掌握这个模式，你不仅能应对面试，更能在实际项目中写出健壮、可靠的多线程代码。