在Java中如何使用BlockingQueue实现阻塞生产者消费者_技术教程

BlockingQueue是实现生产者消费者模型的线程安全方式，1.它通过put/take方法自动阻塞满/空队列，2.使用ArrayBlockingQueue等实现类创建共享队列，3.生产者线程生成任务并放入队列，4.消费者线程从队列取出并处理任务，5.主程序启动线程即可，无需手动同步，代码简洁安全。

在Java中，BlockingQueue 是实现生产者消费者模型最简单且线程安全的方式。它内部已经处理了线程同步问题，当队列满时生产者线程会自动阻塞，当队列空时消费者线程也会自动阻塞，直到有数据可取。

1. 使用 BlockingQueue 的基本原理

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

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

它提供的关键方法：

put(item)：将元素插入队列，如果队列满则阻塞
take()：从队列获取元素并删除，如果队列空则阻塞

2. 定义共享队列和数据模型

先创建一个共享的 BlockingQueue，并定义任务或消息类型：

// 共享缓冲区，最多存放5个任务
BlockingQueue queue = new ArrayBlockingQueue<>(5);

3. 实现生产者线程

生产者不断生成数据并放入队列：

class Producer implements Runnable {
    private final BlockingQueue queue;

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

    public void run() {
        try {
            for (int i = 1; i <= 10; i++) {
                String task = "任务-" + i;
                queue.put(task);
                System.out.println("生产者生产: " + task);
                Thread.sleep(500); // 模拟生产耗时
            }
            // 生产结束后插入结束标志（可选）
            queue.put("END");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

4. 实现消费者线程

消费者从队列中取出数据进行处理：

class Consumer implements Runnable {
    private final BlockingQueue queue;

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

    public void run() {
        try {
            while (true) {
                String task = queue.take();
                if ("END".equals(task)) {
                    System.out.println("消费者收到结束信号，退出");
                    queue.put(task); // 可广播给其他消费者
                    break;
                }
                System.out.println("消费者消费: " + task);
                Thread.sleep(800); // 模拟处理时间
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

5. 启动生产者和消费者线程

在主程序中创建线程并启动：

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

        Thread producerThread = new Thread(new Producer(queue));
        Thread consumerThread = new Thread(new Consumer(queue));

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

        try {
            producerThread.join();
            consumerThread.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

基本上就这些。BlockingQueue 自动处理了锁和等待唤醒逻辑，不需要手动使用 synchronized、wait 或 notify，代码简洁且不易出错。适合大多数并发场景下的生产者消费者问题。