Java EE 多线程核心：阻塞队列与生产者消费者模型

mikasa667

372人浏览 · 2026-03-26 21:38:55

mikasa667 · 2026-03-26 21:38:55 发布

一、阻塞队列（BlockingQueue）

1. 定义

阻塞队列是一个支持阻塞式插入与阻塞式获取的线程安全队列。

当队列已满时，执行入队操作的线程会被阻塞，直到队列有空闲位置。
当队列为空时，执行出队操作的线程会被阻塞，直到队列中有可用元素。阻塞队列内部通过锁与等待 / 唤醒机制实现线程间协同，是实现高效并发协作的基础组件。

2. 打个比方哈

可以把阻塞队列比作一个有容量限制的自助取餐台：

取餐台满了，厨师（生产者）就不能继续放菜，必须停下来等。
取餐台空了，顾客（消费者）就拿不到菜，必须等待新菜做好。
一旦有位置 / 有菜，等待的人会被自动唤醒继续工作，不用反复查看。

3. 核心特点

线程安全，内部自带同步控制
阻塞等待，避免无效循环（自旋）
典型实现：ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue

二、生产者消费者模型

1.定义

生产者消费者模型是一种经典的线程间协作并发设计模式。它将 ** 生产任务的线程（生产者）与消费任务的线程（消费者）** 分离，通过一个共享缓冲区（队列）解耦。生产者只负责向缓冲区提交数据，消费者只负责从缓冲区取出数据处理，彼此不直接通信，从而平衡生产速度与消费速度不匹配的问题。

2.再打个比方哈

好比工厂流水线 + 仓库：

工人（生产者）只管生产产品，放进仓库。
销售员（消费者）只管从仓库取货卖出。
仓库满了工人就暂停，仓库空了销售员就等待。工人和销售员互不打扰、互不等待对方，只跟仓库打交道，效率更高更稳定。

三、如何用阻塞队列构造生产者消费者模型

实现思路

创建一个有界阻塞队列作为共享缓冲区。
生产者线程：不断生产数据，调用 put() 放入队列（满则阻塞）。
消费者线程：不断消费数据，调用 take() 从队列取出（空则阻塞）。
JDK 的 BlockingQueue 内部已处理同步、阻塞、唤醒，无需手动加锁。

代码示例

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

public class ProducerConsumerDemo {
    public static void main(String[] args) {
        // 阻塞队列作为缓冲区，容量为5
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);

        // 生产者线程
        new Thread(() -> {
            for (int i = 1; i <= 10; i++) {
                try {
                    queue.put(i);
                    System.out.println("生产者生产：" + i);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "生产者").start();

        // 消费者线程
        new Thread(() -> {
            while (true) {
                try {
                    Integer val = queue.take();
                    System.out.println("消费者消费：" + val);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "消费者").start();
    }
}

四、生产者消费者模型的优缺点

优点

解耦：生产者与消费者不直接依赖，只依赖队列，代码更易维护扩展。
削峰填谷：平衡生产与消费速度差异，避免一方过快压垮系统。
异步处理：生产与消费可并发执行，提高系统吞吐量。
线程安全：阻塞队列内置同步，避免手动编写复杂锁逻辑。

依旧打个比方

工人和销售互不干扰，各司其职。
忙时仓库囤货，闲时消化库存，不会混乱。
不用互相喊 “做好没”“取完没”，自动等待唤醒。

缺点）

队列容量设置不合理会导致频繁阻塞或内存占用过高。
异常场景下（如消费者卡死）会导致队列堆积，引发内存溢出。
调试复杂度略高，问题定位需要关注队列状态。

再再打个比方

仓库太小总停工，太大会占地方。
销售一直不来，仓库会堆满爆。
出问题时不好一眼看出是生产慢、消费慢还是堵死了。

五、使用 wait () 和 notify () 模拟实现阻塞队列

1. 思路

通过自定义数组队列 + Object 锁 + wait() / notifyAll() 实现：

队列满时，生产者调用 wait() 阻塞。
队列空时，消费者调用 wait() 阻塞。
每次入队 / 出队后，使用 notifyAll() 唤醒等待线程。

2. 思路2

自己做一个简易 “仓库”：

满了就喊 “别放了”，生产者睡觉等通知。
空了就喊 “没货了”，消费者睡觉等通知。
每放一个 / 拿一个，就喊一声 “可以动了”，唤醒睡觉的人。

3. 代码实现


class MyBlockingQueue {
    private final Object[] items;
    private int putIndex;   // 入队下标
    private int takeIndex;  // 出队下标
    private int count;      // 元素数量

    public MyBlockingQueue(int capacity) {
        items = new Object[capacity];
    }

    // 入队（生产者调用）
    public synchronized void put(Object val) throws InterruptedException {
        // 队列满，阻塞
        while (count == items.length) {
            wait();
        }
        items[putIndex] = val;
        putIndex = (putIndex + 1) % items.length;
        count++;
        notifyAll(); // 唤醒消费者
    }

    // 出队（消费者调用）
    public synchronized Object take() throws InterruptedException {
        // 队列空，阻塞
        while (count == 0) {
            wait();
        }
        Object val = items[takeIndex];
        items[takeIndex] = null;
        takeIndex = (takeIndex + 1) % items.length;
        count--;
        notifyAll(); // 唤醒生产者
        return val;
    }
}

4. 测试自定义阻塞队列


public class MyQueueTest {
    public static void main(String[] args) {
        MyBlockingQueue queue = new MyBlockingQueue(3);

        // 生产者
        new Thread(() -> {
            for (int i = 1; i <= 5; i++) {
                try {
                    queue.put(i);
                    System.out.println("生产：" + i);
                } catch (InterruptedException e) {
                }
            }
        }).start();

        // 消费者
        new Thread(() -> {
            while (true) {
                try {
                    Object val = queue.take();
                    System.out.println("消费：" + val);
                } catch (InterruptedException e) {
                }
            }
        }).start();
    }
}