在Java中如何使用BlockingQueue实现生产者消费者模式-java教程-PHP中文网

在Java中如何使用BlockingQueue实现生产者消费者模式

P粉602998670

发布： 2025-10-10 09:42:01

原创

502人浏览过

BlockingQueue是实现生产者消费者模式的线程安全方式，其内部处理同步问题，生产者调用put阻塞入队，消费者调用take阻塞出队，队列满时生产者等待，队列空时消费者等待。

在java中如何使用blockingqueue实现生产者消费者模式

在Java中，BlockingQueue 是实现生产者消费者模式最简单且线程安全的方式。它内部已经处理了线程同步问题，生产者线程向队列添加元素，消费者线程从队列取出元素，当队列满时生产者自动阻塞，队列空时消费者自动阻塞。

1. BlockingQueue 的基本特性

BlockingQueue 是 java.util.concurrent 包中的接口，常用实现类有：

ArrayBlockingQueue：有界阻塞队列，基于数组实现
LinkedBlockingQueue：可选有界队列，基于链表实现
PriorityBlockingQueue：支持优先级的无界阻塞队列
SynchronousQueue：不存储元素的阻塞队列，每个插入必须等待对应移除

关键方法：

put(E e)：将元素插入队列，如果队列满则阻塞
take()：从队列获取并移除元素，如果队列空则阻塞
offer(E e, long timeout, TimeUnit unit)：带超时的插入
poll(long timeout, TimeUnit unit)：带超时的取出

2. 实现生产者消费者模型

下面是一个使用 ArrayBlockingQueue 的简单示例：

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

歌者PPT

歌者PPT，AI 写 PPT 永久免费

197

查看详情

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

// 共享数据
class Data {
    private final int value;

    public Data(int value) {
        this.value = value;
    }

    @Override
    public String toString() {
        return "Data{" + "value=" + value + '}';
    }
}

// 生产者
class Producer implements Runnable {
    private final BlockingQueue<Data> queue;

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

    @Override
    public void run() {
        try {
            for (int i = 1; i <= 5; i++) {
                Data data = new Data(i);
                queue.put(data); // 阻塞插入
                System.out.println("生产者生产: " + data);
                Thread.sleep(500); // 模拟耗时
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// 消费者
class Consumer implements Runnable {
    private final BlockingQueue<Data> queue;

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

    @Override
    public void run() {
        try {
            while (true) {
                Data data = queue.take(); // 阻塞取出
                System.out.println("消费者消费: " + data);
                Thread.sleep(1000); // 消费较慢
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// 主程序
public class ProducerConsumerExample {
    public static void main(String[] args) {
        // 创建容量为3的阻塞队列
        BlockingQueue<Data> queue = new ArrayBlockingQueue<>(3);

        // 启动生产者和消费者线程
        Thread producerThread = new Thread(new Producer(queue));
        Thread consumerThread = new Thread(new Consumer(queue));

        producerThread.start();
        consumerThread.start();

        // 可以设置运行时间后中断
        try {
            Thread.sleep(6000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        producerThread.interrupt();
        consumerThread.interrupt();
    }
}

登录后复制