Java中如何实现生产者消费者模型

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

// 生产者
class Producer implements Runnable {
    private final BlockingQueue<Integer> queue;
    private final AtomicInteger producedCount;
    private volatile boolean running = true;

    public Producer(BlockingQueue<Integer> queue, AtomicInteger producedCount) {
        this.queue = queue;
        this.producedCount = producedCount;
    }

    @Override
    public void run() {
        try {
            while (running) {
                int data = producedCount.incrementAndGet();
                System.out.println(Thread.currentThread().getName() + " 生产: " + data);
                queue.put(data); // 队列满时阻塞
                TimeUnit.MILLISECONDS.sleep(500); // 模拟生产耗时
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println(Thread.currentThread().getName() + " 生产者被中断。");
        } finally {
            System.out.println(Thread.currentThread().getName() + " 生产者停止。");
        }
    }

    public void stop() {
        running = false;
    }
}

// 消费者
class Consumer implements Runnable {
    private final BlockingQueue<Integer> queue;
    private final AtomicInteger consumedCount;
    private volatile boolean running = true;

    public Consumer(BlockingQueue<Integer> queue, AtomicInteger consumedCount) {
        this.queue = queue;
        this.consumedCount = consumedCount;
    }

    @Override
    public void run() {
        try {
            while (running || !queue.isEmpty()) { // 即使停止，也要清空队列
                Integer data = queue.poll(100, TimeUnit.MILLISECONDS); // 队列空时阻塞，带超时
                if (data != null) {
                    consumedCount.incrementAndGet();
                    System.out.println(Thread.currentThread().getName() + " 消费: " + data + ", 队列剩余: " + queue.size());
                } else if (!running) { // 如果已经停止并且队列为空，则退出
                    break;
                }
                TimeUnit.MILLISECONDS.sleep(800); // 模拟消费耗时
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println(Thread.currentThread().getName() + " 消费者被中断。");
        } finally {
            System.out.println(Thread.currentThread().getName() + " 消费者停止。");
        }
    }

    public void stop() {
        running = false;
    }
}

public class ProducerConsumerDemo {
    public static void main(String[] args) throws InterruptedException {
        // 创建一个容量为5的阻塞队列
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);
        AtomicInteger producedCount = new AtomicInteger(0);
        AtomicInteger consumedCount = new AtomicInteger(0);

        // 使用线程池管理生产者和消费者
        ExecutorService producerPool = Executors.newFixedThreadPool(2);
        ExecutorService consumerPool = Executors.newFixedThreadPool(3);

        Producer p1 = new Producer(queue, producedCount);
        Producer p2 = new Producer(queue, producedCount);
        Consumer c1 = new Consumer(queue, consumedCount);
        Consumer c2 = new Consumer(queue, consumedCount);
        Consumer c3 = new Consumer(queue, consumedCount);

        producerPool.execute(p1);
        producerPool.execute(p2);
        consumerPool.execute(c1);
        consumerPool.execute(c2);
        consumerPool.execute(c3);

        // 运行一段时间后停止
        TimeUnit.SECONDS.sleep(10);
        System.out.println("\n--- 停止生产者和消费者 ---");

        p1.stop();
        p2.stop();
        producerPool.shutdown();
        producerPool.awaitTermination(2, TimeUnit.SECONDS); // 等待生产者停止

        c1.stop();
        c2.stop();
        c3.stop();
        consumerPool.shutdown();
        consumerPool.awaitTermination(5, TimeUnit.SECONDS); // 等待消费者停止并清空队列

        System.out.println("\n总生产数量: " + producedCount.get());
        System.out.println("总消费数量: " + consumedCount.get());
        System.out.println("队列最终剩余: " + queue.size());
    }
}