📚 课程目录第4章生产者-消费者模型

4. 生产者-消费者模型

任务说明

本案例实现一个基于阻塞队列的多生产者-多消费者模型。多个生产者线程向有界阻塞队列中写入任务数据，多个消费者线程从队列中取出并处理数据。核心要求是：队列满时生产者阻塞等待，队列空时消费者阻塞等待，整个过程线程安全且不丢失任务。

核心实现


import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ProducerConsumerDemo {
    // 有界阻塞队列：容量为5
    private static final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);
    private static final AtomicInteger taskId = new AtomicInteger(0);
    private static final int TOTAL_TASKS = 20;   // 总任务数
    private static final int PRODUCER_COUNT = 3; // 生产者线程数
    private static final int CONSUMER_COUNT = 2; // 消费者线程数

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch done = new CountDownLatch(TOTAL_TASKS);
        ExecutorService pool = Executors.newFixedThreadPool(PRODUCER_COUNT + CONSUMER_COUNT);

        // 启动消费者：每个消费者持续取任务，直到收到-1哨兵值
        for (int i = 0; i < CONSUMER_COUNT; i++) {
            final String name = "C" + i;
            pool.submit(() -> {
                try {
                    while (true) {
                        Integer task = queue.take(); // 队列空时阻塞
                        if (task == -1) {           // 哨兵：结束
                            break;
                        }
                        // 模拟处理耗时
                        Thread.sleep(20);
                        System.out.println(name + " 处理任务 #" + task);
                        done.countDown();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 启动生产者：每个生产者生成若干任务
        for (int i = 0; i < PRODUCER_COUNT; i++) {
            final String name = "P" + i;
            pool.submit(() -> {
                try {
                    while (true) {
                        int id = taskId.getAndIncrement();
                        if (id >= TOTAL_TASKS) {
                            break; // 任务已分配完
                        }
                        queue.put(id); // 队列满时阻塞
                        System.out.println(name + " 生产任务 #" + id);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 等待所有任务处理完成
        done.await();
        // 向队列放入哨兵，通知消费者退出
        for (int i = 0; i < CONSUMER_COUNT; i++) {
            queue.put(-1);
        }
        pool.shutdown();
        System.out.println("全部 " + TOTAL_TASKS + " 个任务处理完毕");
    }
}

关键设计说明

线程安全队列：使用 ArrayBlockingQueue，其内部通过锁和条件变量实现线程安全，put() 在队列满时自动阻塞，take() 在队列空时自动阻塞，无需额外同步。
任务分配：多个生产者通过 AtomicInteger 原子递增获取任务编号，确保每个任务只被一个生产者生产，不重复不遗漏。
消费者退出机制：消费者无法预知任务何时结束，因此主线程在所有任务处理完成后向队列放入 -1 哨兵值，消费者收到哨兵即退出循环。
完成计数：使用 CountDownLatch 统计已处理任务数，主线程等待其归零后再发送哨兵，避免消费者提前退出导致任务丢失。

运行与验证

编译运行：


javac ProducerConsumerDemo.java
java ProducerConsumerDemo

验证方法：观察控制台输出，确认以下条件全部满足：

每个任务编号 #0 至 #19 恰好被某个消费者输出一次，无重复、无遗漏；
生产者输出“生产任务”与消费者输出“处理任务”交替出现，且任意时刻队列中未处理任务数不超过5（可通过输出节奏间接观察）；
程序最终打印 全部 20 个任务处理完毕 并正常退出，无死锁或异常。

判定标准：若输出中20个任务编号各出现一次且程序正常终止，则说明阻塞队列正确实现了生产者-消费者的线程安全协作；若出现重复处理、遗漏任务或程序卡死，则实现有误。