【阻塞队列的等待唤醒机制】-平芜编程栈

阻塞队列的等待唤醒机制

Java阻塞队列的等待唤醒机制详解及自定义实现
- 1. 阻塞队列的基本概念
- 2. ArrayBlockingQueue源码剖析（等待唤醒机制）
- - 核心字段
- 3. 等待唤醒机制的两种实现方式
- 4. 自定义一个简单阻塞队列
- 5. 常见面试延伸问题
总结

Java阻塞队列的等待唤醒机制详解及自定义实现

今天来分享一个经典的多线程话题：阻塞队列（BlockingQueue）的等待唤醒机制。

在多线程编程中，生产者-消费者模式是最常见的场景之一。而阻塞队列正是解决这个模式的核心工具。它能在队列满时阻塞生产者、在队列空时阻塞消费者，从而实现线程间安全高效的协作。

本文将从以下几个方面进行讲解：

阻塞队列的基本概念和作用
Java并发包中阻塞队列的实现原理（以ArrayBlockingQueue为例）
等待唤醒机制的核心：wait()、notify() 与 Lock + Condition
手把手自定义一个简单阻塞队列
常见面试题延伸

1. 阻塞队列的基本概念

阻塞队列（BlockingQueue）是java.util.concurrent包下的接口，它的典型实现有：

ArrayBlockingQueue：基于数组的有界阻塞队列
LinkedBlockingQueue：基于链表的有界/无界阻塞队列
PriorityBlockingQueue：支持优先级的无界阻塞队列
SynchronousQueue：不存储元素的阻塞队列（容量为0）

阻塞队列的核心操作：

put(e)：向队列尾部添加元素，如果队列已满，则阻塞当前线程，直到有空间
take()：从队列头部移除元素，如果队列为空，则阻塞当前线程，直到有元素

这种“阻塞”行为，正是通过等待唤醒机制实现的。

2. ArrayBlockingQueue源码剖析（等待唤醒机制）

我们以最常用的ArrayBlockingQueue为例，看看它是如何实现阻塞的。

核心字段

final ReentrantLock lock;// 单一锁控制所有操作privatefinal Condition notEmpty;// 消费者等待条件（队列不空）privatefinal Condition notFull;// 生产者等待条件（队列不满）privatefinal Object[]items;// 底层数组int count;// 当前元素数量put操作（入队） Javapublicvoidput(Ee)throws InterruptedException{Objects.requireNonNull(e);final ReentrantLock lock=this.lock;lock.lockInterruptibly();try{while(count==items.length){// 队列满时notFull.await();// 阻塞生产者，释放锁}enqueue(e);// 实际入队if(++count==items.length)// 如果入队后变满，无需唤醒生产者return;notEmpty.signal();// 唤醒一个等待的消费者}finally{lock.unlock();}}take操作（出队） JavapublicEtake()throws InterruptedException{final ReentrantLock lock=this.lock;lock.lockInterruptibly();try{while(count==0){// 队列空时notEmpty.await();// 阻塞消费者，释放锁}Eitem=dequeue();// 实际出队if(--count==0)// 如果出队后变空，无需唤醒消费者returnitem;notFull.signal();// 唤醒一个等待的生产者}finally{lock.unlock();}}

关键点总结：

使用单一ReentrantLock保证线程安全
通过两个Condition（notEmpty、notFull）精确控制唤醒：
生产者只唤醒消费者（notEmpty.signal()）
消费者只唤醒生产者（notFull.signal()）
await() 会释放锁，让其他线程有机会操作队列
signal() 只唤醒一个线程，避免惊群效应
这比使用synchronized + wait/notify更高效、更灵活（可以有多个Condition）。

3. 等待唤醒机制的两种实现方式

方式一：synchronized + wait/notify（传统方式）
wait()：当前线程释放锁并进入WAITING状态
notify()：随机唤醒一个等待线程
notifyAll()：唤醒所有等待线程（容易造成惊群）
方式二：Lock + Condition（推荐方式）
condition.await()：类似wait()
condition.signal()：类似notify()
condition.signalAll()：类似notifyAll()
优势：一个Lock可以创建多个Condition，实现精确唤醒
阻塞队列普遍采用第二种方式。

4. 自定义一个简单阻塞队列

我们来手写一个基于数组的有界阻塞队列，使用synchronized + wait/notify实现（便于理解）：

Javaimportjava.util.concurrent.atomic.AtomicInteger;publicclassMyBlockingQueue<E>{privatefinal Object[]items;privateint takeIndex;// 消费指针privateint putIndex;// 生产指针privatefinal AtomicInteger count=newAtomicInteger(0);privatefinal int capacity;publicMyBlockingQueue(int capacity){this.capacity=capacity;this.items=newObject[capacity];}publicsynchronizedvoidput(Ee)throws InterruptedException{while(count.get()==capacity){// 队列满wait();// 释放锁，阻塞生产者}items[putIndex]=e;if(++putIndex==capacity){putIndex=0;}count.incrementAndGet();notifyAll();// 唤醒消费者（这里用notifyAll避免遗漏）}publicsynchronizedEtake()throws InterruptedException{while(count.get()==0){// 队列空wait();// 释放锁，阻塞消费者}@SuppressWarnings("unchecked")Eitem=(E)items[takeIndex];items[takeIndex]=null;// help GCif(++takeIndex==capacity){takeIndex=0;}count.decrementAndGet();notifyAll();// 唤醒生产者returnitem;}}

测试代码：

JavapublicclassTest{publicstaticvoidmain(String[]args){MyBlockingQueue<Integer>queue=newMyBlockingQueue<>(5);// 生产者newThread(()->{for(int i=1;i<=10;i++){try{queue.put(i);System.out.println("生产: "+i);}catch(InterruptedException e){Thread.currentThread().interrupt();}}}).start();// 消费者newThread(()->{for(int i=1;i<=10;i++){try{Integer val=queue.take();System.out.println("消费: "+val);}catch(InterruptedException e){Thread.currentThread().interrupt();}}}).start();}}

运行结果会看到生产者和消费者交替进行，当队列满/空时自动阻塞。

注意：实际生产中推荐使用Lock + Condition，因为notifyAll()可能导致所有线程都被唤醒后再竞争锁，性能较低。

5. 常见面试延伸问题

为什么不直接用notify而是用notifyAll？
在环形队列中，生产者和消费者都在同一个锁上等待，使用notify()可能唤醒同类型线程，导致继续等待（死等）。notifyAll()更安全。
ArrayBlockingQueue为什么用单一锁而不是读写分离？
为了实现强一致性（入队立即对出队可见），且数组实现下读写锁分离收益不大。
LinkedBlockingQueue为什么用两把锁？
链表头尾操作相对独立，使用takeLock和putLock分离，提高并发度。