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

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

P粉602998670

发布： 2025-10-04 14:05:02

原创

210人浏览过

BlockingQueue是实现生产者消费者模型的线程安全方式，其put()和take()方法自动处理阻塞。1. 特性：BlockingQueue为接口，常用实现有ArrayBlockingQueue（有界数组队列）、LinkedBlockingQueue（可选有界链表队列）和SynchronousQueue（不存储元素的同步队列）。2. 示例：通过SharedData封装BlockingQueue，生产者调用put()添加数据，消费者调用take()获取数据，队列满时生产者阻塞，队列空时消费者阻塞，实现高效线程协作。

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

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

1. BlockingQueue 的基本特性

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

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

关键方法说明：

put(e)：将元素放入队列，如果队列满则阻塞
take()：从队列取出一个元素，如果队列空则阻塞
offer(e, time, unit) / poll(time, unit)：带超时的插入和取出

2. 实现生产者消费者示例

以下是一个使用 ArrayBlockingQueue 的完整示例：

歌者PPT

歌者PPT，AI 写 PPT 永久免费

197

查看详情

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

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

// 共享数据
class SharedData {
    private final BlockingQueue<Integer> queue;
    private final int capacity;

    public SharedData(int capacity) {
        this.queue = new ArrayBlockingQueue<>(capacity);
        this.capacity = capacity;
    }

    public void produce(int value) throws InterruptedException {
        queue.put(value);
        System.out.println("生产：" + value + "，队列大小：" + queue.size());
    }

    public int consume() throws InterruptedException {
        int value = queue.take();
        System.out.println("消费：" + value + "，队列大小：" + queue.size());
        return value;
    }
}

// 生产者线程
class Producer implements Runnable {
    private final SharedData data;
    private final int maxItems;

    public Producer(SharedData data, int maxItems) {
        this.data = data;
        this.maxItems = maxItems;
    }

    @Override
    public void run() {
        for (int i = 1; i <= maxItems; i++) {
            try {
                data.produce(i);
                Thread.sleep(500); // 模拟生产耗时
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
}

// 消费者线程
class Consumer implements Runnable {
    private final SharedData data;
    private final int maxItems;

    public Consumer(SharedData data, int maxItems) {
        this.data = data;
        this.maxItems = maxItems;
    }

    @Override
    public void run() {
        for (int i = 0; i < maxItems; i++) {
            try {
                data.consume();
                Thread.sleep(800); // 消费比生产慢
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
}

// 主程序
public class ProducerConsumerDemo {
    public static void main(String[] args) {
        SharedData sharedData = new SharedData(5); // 队列容量为5

        Thread producer = new Thread(new Producer(sharedData, 10));
        Thread consumer = new Thread(new Consumer(sharedData, 10));

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

        try {
            producer.join();
            consumer.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        System.out.println("执行完成");
    }
}

登录后复制