在Java中如何实现线程安全的生产者消费者模型_生产者消费者实践经验

P粉602998670

发布时间：2025-10-30 08:55:02

931人浏览过

来源于php中文网

原创

使用BlockingQueue可简化线程安全的生产者消费者模型，其内置阻塞机制能自动处理队列满或空的情况，避免手动加锁；结合synchronized与wait/notify或ReentrantLock与Condition可实现更细粒度控制，适用于需自定义同步逻辑的场景。

在java中如何实现线程安全的生产者消费者模型_生产者消费者实践经验

在Java中实现线程安全的生产者消费者模型，核心在于协调多个线程对共享资源的访问，避免数据竞争和状态不一致。通常使用阻塞队列、锁机制或信号量来控制生产者和消费者的协作。下面从实践角度出发，介绍几种常见且可靠的实现方式。

使用BlockingQueue实现线程安全

Java提供了java.util.concurrent.BlockingQueue接口及其实现类（如ArrayBlockingQueue、LinkedBlockingQueue），天然支持线程安全的生产者消费者模式。

BlockingQueue的关键特性是：当队列满时，生产者线程自动阻塞；当队列空时，消费者线程自动阻塞，直到有新元素可用。

示例代码：

立即学习“Java免费学习笔记（深入）”；

import java.util.concurrent.*;

public class ProducerConsumerWithQueue {
    private BlockingQueue queue = new ArrayBlockingQueue<>(5);

    class Producer implements Runnable {
        public void run() {
            try {
                for (int i = 1; i <= 10; i++) {
                    String item = "Item-" + i;
                    queue.put(item); // 自动阻塞
                    System.out.println("生产者生产: " + item);
                    Thread.sleep(500);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    class Consumer implements Runnable {
        public void run() {
            try {
                while (true) {
                    String item = queue.take(); // 自动阻塞
                    System.out.println("消费者消费: " + item);
                    Thread.sleep(800);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public static void main(String[] args) {
        ProducerConsumerWithQueue pc = new ProducerConsumerWithQueue();
        new Thread(pc.new Producer()).start();
        new Thread(pc.new Consumer()).start();
    }
}

这种实现无需手动加锁，代码简洁，推荐在大多数场景下优先使用。

使用synchronized与wait/notify机制

在没有阻塞队列的情况下，可以基于synchronized关键字配合wait()和notifyAll()方法手动实现线程通信。

需要一个共享的缓冲区，并确保所有操作都在同步块中进行。

关键点：

使用while循环检查条件，防止虚假唤醒
调用wait()会释放锁，notifyAll()唤醒所有等待线程
生产者和消费者都需获取同一对象的锁

示例片段：

Stable Diffusion 2.1 Demo

最新体验版 Stable Diffusion 2.1

下载

private final List buffer = new ArrayList<>();
private final int MAX_SIZE = 5;

public void produce(String item) throws InterruptedException {
    synchronized (buffer) {
        while (buffer.size() == MAX_SIZE) {
            buffer.wait(); // 等待空间
        }
        buffer.add(item);
        System.out.println("生产: " + item);
        buffer.notifyAll(); // 唤醒消费者
    }
}

public String consume() throws InterruptedException {
    synchronized (buffer) {
        while (buffer.isEmpty()) {
            buffer.wait(); // 等待数据
        }
        String item = buffer.remove(0);
        System.out.println("消费: " + item);
        buffer.notifyAll(); // 唤醒生产者
        return item;
    }
}

这种方式更底层，适合理解原理，但容易出错，需谨慎处理异常和中断。

使用ReentrantLock与Condition

相比synchronized，ReentrantLock提供更灵活的控制能力，结合Condition可分别定义“非满”和“非空”两个等待条件。

优势：

可以创建多个Condition实例，实现精准唤醒
支持尝试获取锁、超时等高级操作
性能在高并发下通常优于synchronized

示例：

private final Lock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();

private final List buffer = new ArrayList<>();
private final int MAX_SIZE = 5;

public void produce(String item) throws InterruptedException {
    lock.lock();
    try {
        while (buffer.size() == MAX_SIZE) {
            notFull.await(); // 等待不满
        }
        buffer.add(item);
        System.out.println("生产: " + item);
        notEmpty.signal(); // 唤醒消费者
    } finally {
        lock.unlock();
    }
}

public String consume() throws InterruptedException {
    lock.lock();
    try {
        while (buffer.isEmpty()) {
            notEmpty.await(); // 等待不空
        }
        String item = buffer.remove(0);
        System.out.println("消费: " + item);
        notFull.signal(); // 唤醒生产者
        return item;
    } finally {
        lock.unlock();
    }
}

这种写法比wait/notify更清晰，尤其适合复杂同步逻辑。