为什么要自己写一遍
ArrayBlockingQueue 已经很好用了，为什么还要手写一个？
三个理由：
- 理解条件队列的代价。
wait/notify版本和Condition版本、单条件与双条件的差别，只有自己写两遍才会真正感受到。 - 表单式理解 JDK 队列的设计决策。读完这一节再去看
ArrayBlockingQueue的源码，能立刻回答「它为什么用两个Condition」「putIndex/takeIndex为什么分开」。 - 面试与排障的公共语言。生产者消费者的每一个错误写法，在线上都有对应的故障形态。
这一节从最朴素的版本开始，逐步加上关闭语义、双条件、中断处理，最后和 JDK 的实现做一次实测对比。
第一版：单锁 + notifyAll
环形缓冲的核心是三个字段：head（读位置）、tail（写位置）、count（当前元素数）。三个都必须在同一把锁里维护。
public final class BoundedBuffer<T> implements AutoCloseable {
private final Object[] items;
private int head;
private int tail;
private int count;
private boolean closed;
public BoundedBuffer(int capacity) {
if (capacity <= 0) {
throw new IllegalArgumentException("capacity must be positive: " + capacity);
}
this.items = new Object[capacity];
}
public void put(T item) throws InterruptedException {
synchronized (this) {
while (count == items.length && !closed) {
wait();
}
if (closed) {
throw new IllegalStateException("buffer closed");
}
items[tail] = item;
tail = (tail + 1) % items.length;
count++;
notifyAll();
}
}
@SuppressWarnings("unchecked")
public T take() throws InterruptedException {
synchronized (this) {
while (count == 0 && !closed) {
wait();
}
if (count == 0) {
return null; // 已关闭且取空
}
T item = (T) items[head];
items[head] = null; // 断开引用，帮助 GC
head = (head + 1) % items.length;
count--;
notifyAll();
return item;
}
}
@Override
public void close() {
synchronized (this) {
closed = true;
notifyAll(); // 唤醒所有等待者，让它们看到 closed
}
}
}
第一个细节值得单独说：count 不能用 head == tail 代替。环形缓冲里空和满都是 head == tail，必须用额外的计数区分，这是环形队列的经典陷阱。其余几处（清引用、wait 循环带 closed、close 里唤醒）都归入后面的「四个坑」。
关闭语义
close() 的三条约定：
| 操作 | 关闭前 | 关闭后 |
|---|---|---|
put | 正常入队或阻塞 | 抛 IllegalStateException |
take | 正常出队或阻塞 | 返回剩余元素，取完返回 null |
选择「take 返回剩余的、然后 null」而不是「立刻抛异常」，是因为关闭通常意味着「停止接收新数据，把已有的处理完」。如果一关就抛异常，缓冲里剩下的元素就丢了。
这套语义和 BlockingQueue 的习惯不同：JDK 的队列没有 close，通常用「放入哨兵元素」或「中断消费者」来结束。哨兵方案要求每个消费者各放一个哨兵，且哨兵类型要能塞进 BlockingQueue<T>；自定义 close 语义更干净，代价是调用方要理解这套约定。
第二版：两个条件
第一版的 notifyAll 每次唤醒所有等待者。生产者放入元素时，等待中的生产者和消费者都会被唤醒——生产者检查发现还是满的，又睡回去。这就是「惊群」。
用 ReentrantLock 的多个 Condition 可以精确唤醒：
public final class ConditionBuffer<T> implements AutoCloseable {
private final Object[] items;
private final ReentrantLock lock = new ReentrantLock();
private final Condition notEmpty = lock.newCondition();
private final Condition notFull = lock.newCondition();
private int head;
private int tail;
private int count;
private boolean closed;
public ConditionBuffer(int capacity) {
if (capacity <= 0) {
throw new IllegalArgumentException("capacity must be positive: " + capacity);
}
this.items = new Object[capacity];
}
public void put(T item) throws InterruptedException {
lock.lock();
try {
while (count == items.length && !closed) {
notFull.await(); // 生产者只等「不满」
}
if (closed) {
throw new IllegalStateException("buffer closed");
}
items[tail] = item;
tail = (tail + 1) % items.length;
count++;
notEmpty.signal(); // 只唤醒一个消费者
} finally {
lock.unlock();
}
}
@SuppressWarnings("unchecked")
public T take() throws InterruptedException {
lock.lock();
try {
while (count == 0 && !closed) {
notEmpty.await(); // 消费者只等「不空」
}
if (count == 0) {
return null;
}
T item = (T) items[head];
items[head] = null;
head = (head + 1) % items.length;
count--;
notFull.signal(); // 只唤醒一个生产者
return item;
} finally {
lock.unlock();
}
}
@Override
public void close() {
lock.lock();
try {
closed = true;
notEmpty.signalAll(); // 关闭时要唤醒全部
notFull.signalAll();
} finally {
lock.unlock();
}
}
}
三个关键改动：
put在notFull上等，take在notEmpty上等，两类等待者分开排队。put成功后signal的是notEmpty（消费者），不是notFull（生产者自己）。这是最常写错的一行：唤醒同类等于没唤醒。close用signalAll，因为关闭影响所有等待者，不能只叫醒一个。
为什么必须两个 Condition
如果只创建一个 Condition，put 成功时 signal() 唤醒的是同一队列里最早的等待者，可能是另一个生产者。那个生产者检查 count < capacity 不成立，又睡回去，真正的消费者没被叫醒——这次唤醒就浪费了。
单条件配 signalAll 能正确工作，但退化成第一版的惊群。所以「双条件」不是可选的优化，而是 signal 能正确使用的前提。
为什么 signal 而不是 signalAll
notEmpty.signal() 唤醒一个消费者，队列里有 1 个新元素，需要 1 个消费者。这是严格配对的。
但要注意：如果 take 一次取多个元素（批量接口），signal 的数目要跟着变。比如一次取 N 个，就要 signal N 次，或者干脆用 signalAll。这是自定义阻塞队列最容易埋下的一类 bug：信号数量少于实际可用资源数量，导致消费者饿死。
ArrayBlockingQueue 用的是 signal，但是逐个元素配对，天然安全。
实测：正确性与耗时
三个实现各跑「单生产者 + 单消费者，10 万个整数，容量 64」，验证收到的元素数量与顺序：
static long run(String name, Box<Integer> box, int n) throws Exception {
List<Integer> got = new ArrayList<>(n);
long start = System.nanoTime();
Thread producer = new Thread(() -> {
try { for (int i = 0; i < n; i++) box.put(i); } catch (Exception e) { throw new RuntimeException(e); }
});
Thread consumer = new Thread(() -> {
try {
Integer v;
while ((v = box.take()) != null) got.add(v);
} catch (Exception e) { throw new RuntimeException(e); }
});
producer.start();
consumer.start();
producer.join();
box.close();
consumer.join();
// 校验 got.size() == n 且 got.get(i) == i
}
本机 JDK 25 连续跑四次，这里摘两次：
MonitorBuffer : 收到 100000/100000, 顺序正确=true, 耗时 43ms
ConditionBuffer: 收到 100000/100000, 顺序正确=true, 耗时 28ms
ArrayBlockingQ : 收到 100000/100000, 顺序正确=true, 耗时 25ms
---
MonitorBuffer : 收到 100000/100000, 顺序正确=true, 耗时 39ms
ConditionBuffer: 收到 100000/100000, 顺序正确=true, 耗时 55ms
ArrayBlockingQ : 收到 100000/100000, 顺序正确=true, 耗时 21ms
另外两次的耗时分布：MonitorBuffer 39ms / 47ms，ConditionBuffer 29ms / 23ms，ArrayBlockingQ 21ms / 19ms。
怎么读这组数：
- 正确性全部通过：三个实现的元素数量和顺序都正确，说明关闭语义、环形指针、条件判断都没写错。
- 单锁 +
notifyAll稳定偏慢：四次都在 39–47ms，因为每次唤醒都要处理一批「被误唤醒又睡回去」的线程。 - 双条件与 JDK 队列在 19–55ms 之间波动，这组数据受系统负载影响很大（本机跑实验时
load averages在 50 以上），不能用来比较后两者的优劣；要比较吞吐，得在空闲机器上用 JMH 做多轮 fork 测量。
这组数据能支撑的结论只有一条：单锁 notifyAll 有明显的额外唤醒成本。这个结论在原理上成立，实测也一致。
和 ArrayBlockingQueue 的对比
| 维度 | 本文 ConditionBuffer | ArrayBlockingQueue |
|---|---|---|
| 锁结构 | ReentrantLock + 两个 Condition | 同样是 ReentrantLock + 两个 Condition |
| 公平性 | 非公平（可加参数支持） | 构造参数 fair 可选公平 |
| 中断 | await 抛 InterruptedException | put/take 同样响应中断 |
| 超时 | 未提供 | offer(e, timeout) / poll(timeout) |
| 批量操作 | 未提供 | drainTo、addAll、containsAll |
| 迭代器 | 未提供 | 弱一致快照迭代器 |
| 关闭 | close() + take 返回 null | 无，用哨兵或中断 |
| 容量 | 固定 | 固定（构造时指定） |
| 元素空位 | 简单环形 | putIndex/takeIndex/count 三项 |
ArrayBlockingQueue 的 putIndex/takeIndex 与本文的 tail/head 是同一个东西，只是命名不同。它的 count 同样是独立字段，不是靠指针判断空满。
它在细节上更完整：offer 提供不阻塞的快速失败路径，drainTo 支持批量迁移，迭代器是弱一致快照（不抛 ConcurrentModificationException），take 被中断时也不会留下悬空引用。这些都不是「更优雅」，而是把边界情况一个个补完了——自己写的版本在正常路径上没问题，异常路径和并发边界上通常会有遗漏。
所以实践结论是：能用 ArrayBlockingQueue 就用它；需要自定义语义（比如关闭、优先级、批量聚合）时再考虑手写，并且参考 JDK 的实现补齐边界。
四个常见的坑
| 坑 | 后果 | 修法 |
|---|---|---|
| 容量为 0 或负数 | new Object[0] 合法，但 put 永远阻塞（count == items.length 恒真） | 构造器校验 |
| 关闭时只置标志不唤醒 | 已在 wait 中的线程不会自己醒 | close 里必须 notifyAll/signalAll |
take 取走后不清引用 | 长生命周期缓冲区保留所有消费过的对象 | items[head] = null |
| 批量接口信号数量不匹配 | 一次 put 多个却只 signal 一次，消费者饿死 | 信号数 ≥ 可用资源数 |
在 libraryd-conc 里落地
借阅服务里有一个适合用有界缓冲的地方：批量写借阅流水。reserve 成功后把审计事件放进缓冲区，后台线程批量落库。
public final class AuditBuffer implements AutoCloseable {
private final BoundedBuffer<Loan> buffer;
private final Thread flusher;
private final LoanRepository loans;
public AuditBuffer(int capacity, LoanRepository loans) {
this.buffer = new BoundedBuffer<>(capacity);
this.loans = loans;
this.flusher = Thread.ofPlatform()
.name("audit-flusher")
.daemon(true)
.start(this::flushLoop);
}
public void record(Reservation r) {
try {
buffer.put(new Loan(r.loanId(), r.isbn(), r.memberId(), r.dueAt()));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new LoanException(ErrorCode.INTERRUPTED, "audit record interrupted");
} catch (IllegalStateException e) {
// 已关闭：丢弃，不阻断主流程
}
}
private void flushLoop() {
List<Loan> batch = new ArrayList<>(64);
try {
while (true) {
Loan first = buffer.take();
if (first == null) break; // 关闭且取空
batch.add(first);
for (int i = 0; i < 63; i++) { // 攒一批
Loan next = buffer.poll();
if (next == null) break;
batch.add(next);
}
batch.forEach(loans::save);
batch.clear();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
@Override
public void close() throws InterruptedException {
buffer.close();
flusher.join(Duration.ofSeconds(5));
}
}
这里有三个设计选择：
buffer.put满了会阻塞，形成天然的背压——审计写不过来时，主流程被拖慢，而不是把内存撑爆。close只关缓冲区，不打断 flush：flush 线程把剩余元素排空后，take返回null才退出，join兜住 5 秒上限。record捕获IllegalStateException后丢弃，因为关闭期间的审计丢失不应该让借阅失败。
poll() 需要给 BoundedBuffer 补一个非阻塞版本：
@SuppressWarnings("unchecked")
public T poll() {
synchronized (this) {
if (count == 0) {
return null;
}
T item = (T) items[head];
items[head] = null;
head = (head + 1) % items.length;
count--;
notifyAll();
return item;
}
}
这个 notifyAll 是必要的：取走元素意味着「有空位」，等待中的生产者需要被唤醒。
批量落库把 N 次单独写入变成一次批量写入，这是审计、日志、指标上报里最常见的吞吐优化。攒批的代价是延迟：最坏情况下一个事件要等到缓冲区攒够或超时才落库。所以实际实现通常还会加一个「超时就 flush」的分支，用 poll(timeout) 实现。
小结
- 环形缓冲用
head/tail/count三个字段，空满判断必须靠count，不能靠指针相等；关闭语义要三件事：置标志、唤醒所有等待者、take返回剩余元素后返回null。 ReentrantLock两个Condition的价值在于精确唤醒：put唤醒消费者，take唤醒生产者；只用一个Condition时signal会唤醒同类，等于没唤醒。- 实测三个实现各 10 万元素全部有序正确；单锁
notifyAll四次都在 39ms 以上，明显慢于双条件版，但双条件版与ArrayBlockingQueue的差异在本次负载下不可信。 ArrayBlockingQueue在公平性、超时、批量、迭代器、中断清理上都更完整；自定义缓冲只在需要特殊语义时才值得写。
第二部分讲线程的生命周期与协作式取消。第 6 章的第一节从 interrupt 的语义开始：它到底打断了什么，又不打断什么。
阅读导航：上一节 5.2 虚假唤醒与条件谓词循环 ｜ 下一节 6.1 interrupt 的真实语义
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。