在Java中如何使用BlockingQueue实现生产者消费者模式_BlockingQueue集合实践_技术教程

BlockingQueue是Java中实现生产者消费者模式的线程安全队列，位于java.util.concurrent包下，内部自动处理线程同步。2. 其核心方法put()和take()分别在队列满或空时阻塞生产者或消费者线程，实现自然协调。3. 常用实现类包括有界的ArrayBlockingQueue、可选边界的LinkedBlockingQueue和不存储元素的SynchronousQueue。4. 示例代码使用ArrayBlockingQueue和Food对象演示了多线程环境下生产者向队列添加任务、消费者从队列取出任务的完整流程。

在Java中，BlockingQueue 是实现生产者消费者模式最简单且线程安全的方式之一。它位于 java.util.concurrent 包下，内部已经实现了线程同步机制，能够自动处理生产者和消费者之间的等待与唤醒逻辑，无需手动使用 wait() 和 notify()。

BlockingQueue 的核心特性

BlockingQueue 是一个支持两个关键操作的队列：

put(E e)：当队列满时，生产者线程会阻塞，直到有空间可用。
take()：当队列为空时，消费者线程会阻塞，直到有元素可取。

这些阻塞行为让生产者消费者之间的协调变得非常自然。

常用实现类选择

常见的 BlockingQueue 实现有以下几种，可根据需求选择：

ArrayBlockingQueue：基于数组的有界阻塞队列，需指定容量大小。
LinkedBlockingQueue：基于链表的可选有界队列，默认无界（但建议设上限）。
SynchronousQueue：不存储元素的队列，每个 put 必须等待 take，适合高并发传递场景。

一般情况下，ArrayBlockingQueue 是最常用的实现。

代码示例：生产者消费者模型

下面是一个使用 ArrayBlockingQueue 实现的完整示例：

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;

// 共享数据
class Food {
    int id;
    public Food(int id) {
        this.id = id;
    }
}

// 生产者
class Producer implements Runnable {
    private BlockingQueue queue;

    public Producer(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            for (int i = 1; i <= 10; i++) {
                Food food = new Food(i);
                queue.put(food);
                System.out.println("生产了: " + food.id);
                Thread.sleep(100); // 模拟生产耗时
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// 消费者
class Consumer implements Runnable {
    private BlockingQueue queue;

    public Consumer(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                Food food = queue.take();
                System.out.println("消费了: " + food.id);
                Thread.sleep(200); // 模拟消费耗时
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// 主程序
public class ProducerConsumerDemo {
    public static void main(String[] args) {
        BlockingQueue queue = new ArrayBlockingQueue<>(5);

        Thread producer = new Thread(new Producer(queue));
        Thread consumer = new Thread(new Consumer(queue));

        producer.start();
        consumer.start();

        try {
            producer.join();
            consumer.interrupt(); // 生产结束，中断消费者
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

说明：