彻底搞懂Java阻塞队列:从源码解读到性能优化

在Java并发编程中，阻塞队列（BlockingQueue）是一个既能保证线程安全，又能优雅解决生产者-消费者模式的核心工具。

本文将深入剖析阻塞队列的实现原理、6大实现类的特点与应用场景，并带你阅读核心源码。

什么是阻塞队列

阻塞队列（BlockingQueue）是 java.util.concurrent 包下的接口，继承自 Queue，它最大的特点是：

支持阻塞：当队列为空时，take() 操作会阻塞消费者线程；当队列满时，put() 操作会阻塞生产者线程。

线程安全：内部使用 ReentrantLock 和 Condition 实现，无需开发者手动加锁

这与普通队列（如LinkedList）完全不同：普通队列在空时返回null，在满时抛出异常；而阻塞队列提供了线程安全的阻塞等待机制。

三大核心方法

方法类型	抛出异常	特殊值	阻塞	超时
插入	add(e)	offer(e)	put(e)	offer(e, time, unit)
移除	remove()	poll()	take()	poll(time, unit)
检查	element()	peek()	-	-

推荐使用put() / take()（阻塞式）或带超时的 offer / poll。

阻塞队列的6大实现类

实现类	底层结构	容量	排序特性	适用场景	性能特点
ArrayBlockingQueue	数组	有界	FIFO	固定大小、吞吐量要求高	性能最高（数组连续内存）
LinkedBlockingQueue	链表	可无界	FIFO	生产者消费者经典模型	吞吐量高，内存占用灵活
PriorityBlockingQueue	堆（二叉堆）	无界	按优先级排序	需要优先级处理的任务	插入 O(log n)
DelayQueue	优先级队列	无界	按延迟时间排序	定时任务、缓存过期	延迟执行
SynchronousQueue	无存储空间	0	直接传递	线程间直接交接（无缓冲）	极致性能（无队列开销）
LinkedTransferQueue	链表	无界	FIFO + 转移	高性能场景（支持 transfer）	最强大，JDK7+ 推荐

重点剖析

1. ArrayBlockingQueue vs LinkedBlockingQueue

对比项	ArrayBlockingQueue	LinkedBlockingQueue
底层结构	数组	链表（Node节点）
边界	有界（创建时必须指定容量）	可选有界（默认Integer.MAX_VALUE）
锁机制	一把锁（生产消费共用）	两把锁（生产锁+消费锁）
吞吐量	较低	较高
内存占用	较低	较高（每个节点有额外开销）

源码分析：LinkedBlockingQueue的两把锁设计

// LinkedBlockingQueue 核心字段privatefinal ReentrantLock takeLock = new ReentrantLock(); // 消费锁privatefinal Condition notEmpty = takeLock.newCondition();privatefinal ReentrantLock putLock = new ReentrantLock(); // 生产锁privatefinal Condition notFull = putLock.newCondition();

生产者（put）只竞争putLock，消费者（take）只竞争takeLock，生产和消费可以真正并行，吞吐量高。

而ArrayBlockingQueue只有一把锁：

// ArrayBlockingQueue 核心字段privatefinal ReentrantLock lock = new ReentrantLock();privatefinal Condition notEmpty = lock.newCondition();privatefinal Condition notFull = lock.newCondition();

生产者和消费者竞争同一把锁，效率相对较低。

2. SynchronousQueue

它不存储元素，容量为0。每个put()操作必须等待一个take()操作，否则阻塞。

SynchronousQueue<String> queue = new SynchronousQueue<>();// 下面的put会阻塞，直到有线程调用takenew Thread(() -> {try {queue.put("data");    } catch (InterruptedException e) {}}).start();// 主线程取走数据，上面阻塞解除String data = queue.take(); // 拿到 "data"

它也是Executors.newCachedThreadPool()的默认工作队列。

3. DelayQueue

元素必须实现Delayed接口，只有延迟时间已到的元素才能被take()取出。常用于：

定时任务调度

缓存过期清理

订单超时关闭

classTaskimplementsDelayed{private String name;privatelong startTime; // 执行时间戳@OverridepubliclonggetDelay(TimeUnit unit){return unit.convert(startTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);    }@OverridepublicintcompareTo(Delayed o){return Long.compare(this.getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));    }}

源码解析

1. ArrayBlockingQueue的put()实现

publicvoidput(E e) throws InterruptedException {    checkNotNull(e);    final ReentrantLock lock = this.lock;lock.lockInterruptibly(); // 可中断加锁try {while (count == items.length) // 队列满了            notFull.await(); // 阻塞等待        enqueue(e); // 入队    } finally {lock.unlock();    }}privatevoidenqueue(E x) {    final Object[] items = this.items;    items[putIndex] = x;if (++putIndex == items.length) putIndex = 0; // 环形数组    count++;    notEmpty.signal(); // 唤醒等待的消费者}

使用while循环检查而非if，防止虚假唤醒

入队后调用notEmpty.signal()唤醒消费者

2. LinkedBlockingQueue的take()实现

public E take()throws InterruptedException {    E x;int c = -1;final AtomicInteger count = this.count;final ReentrantLock takeLock = this.takeLock;    takeLock.lockInterruptibly();try {while (count.get() == 0) // 队列为空            notEmpty.await(); // 阻塞等待        x = dequeue(); // 出队        c = count.getAndDecrement();if (c > 1) // 出队后队列仍非空            notEmpty.signal(); // 唤醒其他消费者    } finally {        takeLock.unlock();    }if (c == capacity) // 之前队列满了        signalNotFull(); // 唤醒生产者return x;}

消费完成后，如果队列仍非空，唤醒其他消费者

如果消费前队列是满的，消费后唤醒生产者（通过signalNotFull()）

经典应用场景与代码示例

场景1：生产者-消费者模型

publicclassProducerConsumerDemo{privatestaticfinal BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);staticclassProducerimplementsRunnable{@Overridepublicvoidrun(){int i = 0;while (true) {try {                    queue.put(i++); // 满了就阻塞                    System.out.println("生产: " + i);                    Thread.sleep(500);                } catch (InterruptedException e) {                    Thread.currentThread().interrupt();                }            }        }    }staticclassConsumerimplementsRunnable{@Overridepublicvoidrun(){while (true) {try {                    Integer val = queue.take(); // 空了就阻塞                    System.out.println("消费: " + val);                    Thread.sleep(1000);                } catch (InterruptedException e) {                    Thread.currentThread().interrupt();                }            }        }    }publicstaticvoidmain(String[] args){new Thread(new Producer()).start();new Thread(new Consumer()).start();    }}

场景2：线程池中的阻塞队列

// 固定线程池采用 LinkedBlockingQueue（无界）ExecutorService pool = Executors.newFixedThreadPool(5);// 内部核心：return new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,// new LinkedBlockingQueue<Runnable>());// 缓存线程池采用 SynchronousQueue（直接传递）ExecutorService cachedPool = Executors.newCachedThreadPool();// 内部核心：return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,// new SynchronousQueue<Runnable>());

场景3：延迟订单关闭

publicclassOrderimplementsDelayed{private String orderId;privatelong expireTime;publicOrder(String orderId, long delayMillis){this.orderId = orderId;this.expireTime = System.currentTimeMillis() + delayMillis;    }@OverridepubliclonggetDelay(TimeUnit unit){long diff = expireTime - System.currentTimeMillis();return unit.convert(diff, TimeUnit.MILLISECONDS);    }@OverridepublicintcompareTo(Delayed o){return Long.compare(this.expireTime, ((Order) o).expireTime);    }}// 使用DelayQueue<Order> queue = new DelayQueue<>();queue.put(new Order("001", 30 * 60 * 1000)); // 30分钟后关闭Order order = queue.take(); // 阻塞直到30分钟后closeOrder(order.getOrderId());