1. 引言为什么需要 ArrayBlockingQueue在Java并发编程中生产者-消费者模式是一种非常经典的解耦设计。而阻塞队列正是这一模式的核心组件。ArrayBlockingQueue作为JUC包中一个重要的有界阻塞队列实现它通过数组存储数据并利用ReentrantLock和Condition实现了线程安全的阻塞存取操作。本文将带你从类图、核心属性、构造器到put/take及offer/poll的核心源码逐行分析其背后原理并解答一个关键问题为什么使用 while 而不是 if 进行条件判断2. 核心结构与类图解析我们先通过一个简化的类图一览ArrayBlockingQueue的内部骨架关键点说明items、takeIndex、putIndex、count等变量没有使用 volatile因为所有读写都在锁保护范围内锁已经保证了内存可见性。lock是全局独占锁同一时刻只允许一个线程执行入队或出队操作读写互斥。notEmpty和notFull是两个条件变量用于线程间的等待与唤醒。3. 构造器初始化队列与锁public ArrayBlockingQueue(int capacity) { this(capacity, false); } public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity 0) throw new IllegalArgumentException(); this.items new Object[capacity]; lock new ReentrantLock(fair); notEmpty lock.newCondition(); notFull lock.newCondition(); }capacity必须 0数组大小一旦确定就不能扩容有界。fair参数决定锁是否为公平锁。公平锁能避免线程饥饿但会降低吞吐量。4. 入队操作源码分析4.1 offer(e) —— 非阻塞入队public boolean offer(E e) { checkNotNull(e); final ReentrantLock lock this.lock; lock.lock(); try { if (count items.length) return false; else { enqueue(e); return true; } } finally { lock.unlock(); } }流程解析加锁保证线程安全。队列满直接返回false不阻塞。队列未满调用enqueue真正入队。解锁释放锁修改后的变量如count会立即刷回主内存。优点快速失败适合高并发下不希望等待的场景。4.2 enqueue —— 真正的入队逻辑private void enqueue(E x) { final Object[] items this.items; items[putIndex] x; // 放置元素 if (putIndex items.length) // 环形数组 putIndex 0; count; // 元素个数1 notEmpty.signal(); // 唤醒一个等待取元素的线程 }环形数组通过putIndex自增并取模 length时重置为0实现数组的循环利用。当putIndex到达数组末尾时下次插入会从0开始。信号唤醒每次成功入队后都会调用notEmpty.signal()通知可能正在等待“队列非空”的消费者线程。4.3 put(e) —— 可阻塞入队public void put(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(); } }为什么用lockInterruptibly()允许线程在等待锁的过程中响应中断抛出InterruptedException这是一种优雅的退出机制。为什么是while而不是if防止虚假唤醒。线程可能在未真正被 signal 的情况下醒来spurious wakeup用while重新检查条件确保队列真的未满才继续执行。5. 出队操作源码分析5.1 poll() —— 非阻塞出队public E poll() { final ReentrantLock lock this.lock; lock.lock(); try { return (count 0) ? null : dequeue(); } finally { lock.unlock(); } }队列为空直接返回null不阻塞。否则调用dequeue取出头元素。5.2 dequeue —— 真正的出队逻辑private E dequeue() { final Object[] items this.items; SuppressWarnings(unchecked) E x (E) items[takeIndex]; items[takeIndex] null; // 帮助GC if (takeIndex items.length) takeIndex 0; count--; notFull.signal(); // 唤醒一个等待放入的线程 return x; }细节分析取出takeIndex位置的元素并置为null让GC及时回收。更新takeIndex和count。notFull.signal()队列空出一个位置通知可能正在等待“队列未满”的生产者。5.3 take() —— 可阻塞出队public E take() throws InterruptedException { final ReentrantLock lock this.lock; lock.lockInterruptibly(); try { while (count 0) notEmpty.await(); return dequeue(); } finally { lock.unlock(); } }while(count 0)如果队列为空当前线程进入notEmpty条件队列等待。与poll的唯一区别take会在队列空时阻塞而不是立即返回null。6. 条件变量与等待通知机制图解生产者线程 消费者线程 | | put() 队列满 take() 队列空 | | notFull.await() notEmpty.await() | | 等待 等待 | | dequeue() 出队 enqueue() 入队 | | notFull.signal() notEmpty.signal()notFull队列满时生产者等待。notEmpty队列空时消费者等待。每次成功的enqueue会唤醒一个消费者每次成功的dequeue会唤醒一个生产者。7. 与 LinkedBlockingQueue 的对比特性ArrayBlockingQueueLinkedBlockingQueue数据结构数组链表是否无界有界必须指定容量可选有界/无界锁粒度一个锁读写互斥两个锁读写分离内存效率更好连续内存一般吞吐量较低锁竞争大较高size() 精确度精确锁内计算精确但基于原子变量8. 最佳实践与注意事项选择合适的容量ArrayBlockingQueue是有界的容量过小会导致频繁阻塞过大则浪费内存。公平锁 vs 非公平锁默认非公平锁性能更高但公平锁可避免线程饥饿。使用offer或poll代替put/take在非阻塞或超时场景下避免线程长时间挂起。正确处理中断使用lockInterruptibly()并捕获InterruptedException在取消任务时及时释放资源。9. 总结ArrayBlockingQueue虽然名字简单但背后蕴含了并发编程中的核心思想锁保证原子性与可见性。条件变量实现高效的等待/通知机制。环形数组实现内存复用。while 循环 await防止虚假唤醒。
Java并发编程：深入剖析 ArrayBlockingQueue
1. 引言为什么需要 ArrayBlockingQueue在Java并发编程中生产者-消费者模式是一种非常经典的解耦设计。而阻塞队列正是这一模式的核心组件。ArrayBlockingQueue作为JUC包中一个重要的有界阻塞队列实现它通过数组存储数据并利用ReentrantLock和Condition实现了线程安全的阻塞存取操作。本文将带你从类图、核心属性、构造器到put/take及offer/poll的核心源码逐行分析其背后原理并解答一个关键问题为什么使用 while 而不是 if 进行条件判断2. 核心结构与类图解析我们先通过一个简化的类图一览ArrayBlockingQueue的内部骨架关键点说明items、takeIndex、putIndex、count等变量没有使用 volatile因为所有读写都在锁保护范围内锁已经保证了内存可见性。lock是全局独占锁同一时刻只允许一个线程执行入队或出队操作读写互斥。notEmpty和notFull是两个条件变量用于线程间的等待与唤醒。3. 构造器初始化队列与锁public ArrayBlockingQueue(int capacity) { this(capacity, false); } public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity 0) throw new IllegalArgumentException(); this.items new Object[capacity]; lock new ReentrantLock(fair); notEmpty lock.newCondition(); notFull lock.newCondition(); }capacity必须 0数组大小一旦确定就不能扩容有界。fair参数决定锁是否为公平锁。公平锁能避免线程饥饿但会降低吞吐量。4. 入队操作源码分析4.1 offer(e) —— 非阻塞入队public boolean offer(E e) { checkNotNull(e); final ReentrantLock lock this.lock; lock.lock(); try { if (count items.length) return false; else { enqueue(e); return true; } } finally { lock.unlock(); } }流程解析加锁保证线程安全。队列满直接返回false不阻塞。队列未满调用enqueue真正入队。解锁释放锁修改后的变量如count会立即刷回主内存。优点快速失败适合高并发下不希望等待的场景。4.2 enqueue —— 真正的入队逻辑private void enqueue(E x) { final Object[] items this.items; items[putIndex] x; // 放置元素 if (putIndex items.length) // 环形数组 putIndex 0; count; // 元素个数1 notEmpty.signal(); // 唤醒一个等待取元素的线程 }环形数组通过putIndex自增并取模 length时重置为0实现数组的循环利用。当putIndex到达数组末尾时下次插入会从0开始。信号唤醒每次成功入队后都会调用notEmpty.signal()通知可能正在等待“队列非空”的消费者线程。4.3 put(e) —— 可阻塞入队public void put(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(); } }为什么用lockInterruptibly()允许线程在等待锁的过程中响应中断抛出InterruptedException这是一种优雅的退出机制。为什么是while而不是if防止虚假唤醒。线程可能在未真正被 signal 的情况下醒来spurious wakeup用while重新检查条件确保队列真的未满才继续执行。5. 出队操作源码分析5.1 poll() —— 非阻塞出队public E poll() { final ReentrantLock lock this.lock; lock.lock(); try { return (count 0) ? null : dequeue(); } finally { lock.unlock(); } }队列为空直接返回null不阻塞。否则调用dequeue取出头元素。5.2 dequeue —— 真正的出队逻辑private E dequeue() { final Object[] items this.items; SuppressWarnings(unchecked) E x (E) items[takeIndex]; items[takeIndex] null; // 帮助GC if (takeIndex items.length) takeIndex 0; count--; notFull.signal(); // 唤醒一个等待放入的线程 return x; }细节分析取出takeIndex位置的元素并置为null让GC及时回收。更新takeIndex和count。notFull.signal()队列空出一个位置通知可能正在等待“队列未满”的生产者。5.3 take() —— 可阻塞出队public E take() throws InterruptedException { final ReentrantLock lock this.lock; lock.lockInterruptibly(); try { while (count 0) notEmpty.await(); return dequeue(); } finally { lock.unlock(); } }while(count 0)如果队列为空当前线程进入notEmpty条件队列等待。与poll的唯一区别take会在队列空时阻塞而不是立即返回null。6. 条件变量与等待通知机制图解生产者线程 消费者线程 | | put() 队列满 take() 队列空 | | notFull.await() notEmpty.await() | | 等待 等待 | | dequeue() 出队 enqueue() 入队 | | notFull.signal() notEmpty.signal()notFull队列满时生产者等待。notEmpty队列空时消费者等待。每次成功的enqueue会唤醒一个消费者每次成功的dequeue会唤醒一个生产者。7. 与 LinkedBlockingQueue 的对比特性ArrayBlockingQueueLinkedBlockingQueue数据结构数组链表是否无界有界必须指定容量可选有界/无界锁粒度一个锁读写互斥两个锁读写分离内存效率更好连续内存一般吞吐量较低锁竞争大较高size() 精确度精确锁内计算精确但基于原子变量8. 最佳实践与注意事项选择合适的容量ArrayBlockingQueue是有界的容量过小会导致频繁阻塞过大则浪费内存。公平锁 vs 非公平锁默认非公平锁性能更高但公平锁可避免线程饥饿。使用offer或poll代替put/take在非阻塞或超时场景下避免线程长时间挂起。正确处理中断使用lockInterruptibly()并捕获InterruptedException在取消任务时及时释放资源。9. 总结ArrayBlockingQueue虽然名字简单但背后蕴含了并发编程中的核心思想锁保证原子性与可见性。条件变量实现高效的等待/通知机制。环形数组实现内存复用。while 循环 await防止虚假唤醒。
