深入浅出Java生产者消费者模式：多线程协作的核心套路

在Java多线程编程中，生产者消费者模式是最经典的设计模式之一。它不仅能够解耦生产者和消费者的逻辑，还能有效平衡两者之间的处理速度差异。本文将带你全面掌握Java生产者消费者模式的精髓。

一、生产者消费者模式基础概念

生产者消费者模式（Producer-Consumer Pattern）是指多个线程共享一个固定大小的缓冲区，其中生产者线程负责生成数据并放入缓冲区，而消费者线程负责从缓冲区取出数据进行消费。这种模式的核心价值在于：

1. 解耦生产与消费过程
2. 平衡处理速度差异
3. 提高系统整体吞吐量

二、基础实现方式

1. 使用wait()和notify()实现

这是最经典的实现方式，通过Object类的wait()和notify()方法实现线程间通信：

public class ClassicProducerConsumer {
    private Queue<Integer> queue = new LinkedList<>();
    private final int LIMIT = 10;
    private Object lock = new Object();

    public void produce() throws InterruptedException {
        int value = 0;
        while (true) {
            synchronized (lock) {
                while (queue.size() == LIMIT) {
                    lock.wait();
                }
                queue.offer(value++);
                lock.notify();
            }
        }
    }

    public void consume() throws InterruptedException {
        while (true) {
            synchronized (lock) {
                while (queue.isEmpty()) {
                    lock.wait();
                }
                int value = queue.poll();
                System.out.println("Consumed: " + value);
                lock.notify();
            }
            Thread.sleep(1000);
        }
    }
}

2. 使用BlockingQueue实现

Java并发包中的BlockingQueue大大简化了实现：

public class BlockingQueuePC {
    private BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);

    public void produce() throws InterruptedException {
        int value = 0;
        while (true) {
            queue.put(value++);
            Thread.sleep(100);
        }
    }

    public void consume() throws InterruptedException {
        while (true) {
            int value = queue.take();
            System.out.println("Consumed: " + value);
            Thread.sleep(1000);
        }
    }
}

三、高级实现与优化

1. 使用Lock和Condition

相比synchronized，Lock提供了更灵活的锁机制：

public class LockConditionPC {
    private Queue<Integer> queue = new LinkedList<>();
    private final int LIMIT = 10;
    private Lock lock = new ReentrantLock();
    private Condition notFull = lock.newCondition();
    private Condition notEmpty = lock.newCondition();

    public void produce() throws InterruptedException {
        int value = 0;
        while (true) {
            lock.lock();
            try {
                while (queue.size() == LIMIT) {
                    notFull.await();
                }
                queue.offer(value++);
                notEmpty.signal();
            } finally {
                lock.unlock();
            }
        }
    }

    public void consume() throws InterruptedException {
        while (true) {
            lock.lock();
            try {
                while (queue.isEmpty()) {
                    notEmpty.await();
                }
                int value = queue.poll();
                System.out.println("Consumed: " + value);
                notFull.signal();
            } finally {
                lock.unlock();
            }
            Thread.sleep(1000);
        }
    }
}

2. 高性能无锁实现

对于极高并发场景，可以考虑无锁实现：

public class DisruptorPC {
    class Event {
        private int value;
        // getters & setters
    }

    public static void main(String[] args) {
        Disruptor<Event> disruptor = new Disruptor<>(
            Event::new, 1024, DaemonThreadFactory.INSTANCE);

        disruptor.handleEventsWith((event, sequence, endOfBatch) -> 
            System.out.println("Consumed: " + event.getValue()));

        disruptor.start();

        RingBuffer<Event> ringBuffer = disruptor.getRingBuffer();
        for (int i = 0; true; i++) {
            long sequence = ringBuffer.next();
            try {
                Event event = ringBuffer.get(sequence);
                event.setValue(i);
            } finally {
                ringBuffer.publish(sequence);
            }
            Thread.sleep(100);
        }
    }
}

四、性能对比与选型建议

我们对几种实现方式进行了基准测试（JMH），结果如下：

选型建议：
1. 对于简单场景，BlockingQueue是最佳选择
2. 需要精细控制时使用Lock/Condition
3. 极端性能要求考虑Disruptor
4. 传统wait/notify已不推荐在新项目中使用

五、常见问题与解决方案

1. 死锁问题

症状：生产者和消费者都在等待对方唤醒
解决方案：
- 确保每次wait()都有对应的notify()
- 使用超时版本的wait(long timeout)

2. 活锁问题

症状：线程不断被唤醒但无法取得进展
解决方案：
- 增加随机退避时间
- 检查系统负载是否过高

3. 内存溢出

症状：生产者速度远大于消费者
解决方案：
- 设置合理的队列上限
- 实现背压机制

六、实际应用案例

1. 日志处理系统

生产者：应用线程生成日志
消费者：日志收集线程批量写入文件/数据库

2. 订单处理系统

生产者：用户下单
消费者：订单处理服务

3. 消息队列

Kafka、RabbitMQ等消息中间件的核心模式

七、总结

Java生产者消费者模式是多线程编程的基石，掌握不同实现方式及其适用场景对构建高性能并发系统至关重要。从基础的wait/notify到高性能的Disruptor，开发者应根据实际需求选择最合适的实现方案。记住，没有最好的方案，只有最适合的方案。

希望本文能帮助你全面理解Java生产者消费者模式，在实际开发中游刃有余地处理各种并发场景。