本节目标：把「借阅成功后要做的几件慢事」从请求线程搬走，同时保证过载可见、停机不丢数据。
适用版本：JDK 25（Temurin 25.0.5）；只使用 JDK 标准库。
承接的问题
5.1 的监听器是在请求线程里同步 fire 的。审计监听器要往 loan_audit 表插一条记录，积分监听器要更新会员积分，推送监听器要调外部 HTTP。一次借阅因此变成「一次数据库写 + 一次审计写 + 一次积分写 + 一次外部调用」，响应时间从 20ms 涨到 120ms，而且外部服务抖动会直接体现在借阅接口的 P99 上。
自然的想法是异步：请求线程只负责把「发生了什么」放进队列，后台线程慢慢处理。这引入四个新问题：
- 队列有没有上限？没有上限就是拿内存当缓冲。
- 队列满了以后请求线程怎么办？阻塞、丢弃、还是拒绝？
- 后台消费者挂了怎么办？请求线程会不会永远塞不进去？
- 停机时队列里还有 200 条事件，是丢掉还是等它跑完？
这一节用一条真实形状的流水线回答这四个问题。
队列的四组操作
BlockingQueue 的方法按「队列满/空时怎么办」分成四组，选错一组语义就全变了：
| 操作 | 队列满时 | 队列空时 |
|---|---|---|
add(e) / remove() | 抛 IllegalStateException / NoSuchElementException | 同左 |
offer(e) / poll() | 立即返回 false / null | 同左 |
put(e) / take() | 一直阻塞 | 一直阻塞 |
offer(e, t, u) / poll(t, u) | 等 t 后返回 false | 等 t 后返回 null |
libraryd 的选择是 offer(event, timeout)：请求线程最多等 50ms，等不到就承认过载，返回 503 或降级。不要用无超时的 put：消费者一旦卡住（比如审计库连接池耗尽），所有请求线程会一起挂在这条队列上，HTTP 层的线程池很快被吃光，整个服务不可用。
put 适合「生产者本身就是后台线程、可以无限等待」的场景，比如对账任务往流水线里灌数据。请求线程上永远不要用 put。
ArrayBlockingQueue 与 LinkedBlockingQueue
两者都实现了 BlockingQueue，差别在锁的粒度和内存布局：
ArrayBlockingQueue是一个固定长度的数组，一把锁同时管put和take。LinkedBlockingQueue是链表，两把锁（putLock、takeLock），生产者和消费者可以并行。LinkedBlockingQueue不传容量时容量是Integer.MAX_VALUE，等于无界。
用 1 个生产者、1 个消费者传 20 万个整数，队列容量统一给 64：
javac --release 25 -d /tmp/vol3-b/queues/out /tmp/vol3-b/queues/QueueBench.java
java -cp /tmp/vol3-b/queues/out QueueBench 64
java -cp /tmp/vol3-b/queues/out QueueBench 64
本机两次输出：
ArrayBlockingQueue capacity=64 events=200000 consumed=200000 elapsedMs=52 opsPerMs=3815.1
LinkedBlockingQueue capacity=64 events=200000 consumed=200000 elapsedMs=48 opsPerMs=4111.9
LinkedBlockingQueue(unbounded) capacity=2147483647 events=200000 consumed=200000 elapsedMs=21 opsPerMs=9503.4
ArrayBlockingQueue capacity=64 events=200000 consumed=200000 elapsedMs=71 opsPerMs=2802.0
LinkedBlockingQueue capacity=64 events=200000 consumed=200000 elapsedMs=54 opsPerMs=3673.2
LinkedBlockingQueue(unbounded) capacity=2147483647 events=200000 consumed=200000 elapsedMs=19 opsPerMs=10491.7
三点观察：
- 有界队列下两者接近，
LinkedBlockingQueue略快（双锁让生产和消费重叠），但差距在 20% 以内，不构成选型理由。 - 无界版本快 2–4 倍，因为它几乎不让生产者阻塞。这个「快」是假的：生产者把 20 万个元素全塞进链表就返回了，消费速度根本没跟上。真实系统里，这些元素会变成堆内存和延迟。
ArrayBlockingQueue的优势是内存可预测：数组一次分配，不会因为突发流量产生大量节点对象。GC 敏感的场合值得为此放弃一点吞吐。
libraryd 选 ArrayBlockingQueue，因为容量必须显式给出，这个数字可以写进配置、可以做监控、可以在压测里调。
背压：队列满了会发生什么
做一个饱和实验：容量 32，2 个消费者，每个事件处理 5ms，生产者连续投 300 个事件，每次 offer 的超时设为 0（等价于 tryOffer）：
new LoanEventPipeline(32, 2, Duration.ofMillis(5));
for (int i = 0; i < 300; i++) {
if (pipeline.offer(event(i), Duration.ZERO)) {
accepted++;
} else {
rejected++;
}
}
本机输出（两次运行一致）：
produced=300 accepted=34 rejected=266 elapsedMs=1 maxDepth=32
before close pending=26
close: pendingAtClose=26 drainedDuringClose=28 abandoned=0 elapsedMs=83
- 300 个事件只有 34 个被接受：2 个在消费者手里，32 个在队列里。
- 266 个被拒绝，占 88.7%。这不是失败，是背压生效：系统明确告诉调用方「现在处理不了」。
maxDepth=32说明队列确实被填满过，配置的上限被用到了。- 生产者 1ms 就返回了，没有因为队列满而阻塞 300 次。
如果超时改成 50ms 而不是 0，被拒绝的会少一些，但请求线程会被拖慢。这两个数是一对权衡：
| 超时 | 拒绝率 | 请求延迟 |
|---|---|---|
| 0ms | 高 | 不受影响 |
| 50ms | 低 | 最多 +50ms |
无超时 put | 0 | 可能无限 |
libraryd 对审计事件用 50ms：审计不能丢，但也不能把借阅接口拖垮。对推送事件用 0ms：推送丢一条可以补，接口延迟更重要。同一个流水线，不同事件类型可以有不同的超时，这比全局一个参数更贴近业务。
优雅停止：先排空，再退出
停机时队列里还有事件，直接 interrupt 消费者会丢掉正在处理的那条，直接退出会丢掉整个队列。libraryd 的做法是「停止接收 + 排空 + 超时兜底」：
running = false，消费循环的条件变成while (running || !queue.isEmpty())，队列里还有就继续取。- 主线程
join每个消费者，最多等drainTimeout。 - 超时后仍有消费者活着，
interrupt，把队列里剩下的记为abandoned。
上面的输出里，close 时队列有 26 条，排空期间又消费了 28 条（26 条加上 2 个消费者手里正在处理的），abandoned=0，耗时 83ms。这个 83ms 可以写进停机的总预算：HTTP 先停（不再有新事件），流水线再排空，最后关连接池。
三个数要进日志和指标：
pending：close 那一刻还排队的数量，反映停机时的压力。drained：排空期间成功处理的数量，正常应等于pending加上在途数量。abandoned：超时后被放弃的数量，非 0 就需要告警和人工补偿。
在 libraryd 里落地
事件类型用 record，字段不可变：
package com.plume.library.domain;
import java.time.Instant;
public record LoanEvent(long loanId, String isbn, String memberId,
Type type, Instant at) {
public enum Type { BORROWED, RETURNED, RENEWED }
public LoanEvent {
if (loanId < 0) {
throw new IllegalArgumentException("loanId 不能为负: " + loanId);
}
if (type == null || at == null) {
throw new IllegalArgumentException("type 与 at 不能为 null");
}
}
}
流水线本体：
package com.plume.library.persistence;
import com.plume.library.domain.LoanEvent;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;
public final class LoanEventPipeline implements AutoCloseable {
private static final System.Logger LOG = System.getLogger("libraryd.loan.events");
public interface LoanEventConsumer {
void accept(LoanEvent event) throws Exception;
}
public record Metrics(long accepted, long rejected, long consumed, long failed,
int queued, int maxDepth) { }
private final BlockingQueue<LoanEvent> queue;
private final List<Thread> workers = new ArrayList<>();
private final LongAdder accepted = new LongAdder();
private final LongAdder rejected = new LongAdder();
private final LongAdder consumed = new LongAdder();
private final LongAdder failed = new LongAdder();
private final AtomicInteger maxDepth = new AtomicInteger();
private volatile boolean running = true;
public LoanEventPipeline(int capacity, int workerCount,
LoanEventConsumer consumer, String name) {
if (capacity < 1 || workerCount < 1) {
throw new IllegalArgumentException("capacity 与 workerCount 必须大于 0");
}
this.queue = new ArrayBlockingQueue<>(capacity);
for (int i = 0; i < workerCount; i++) {
Thread worker = new Thread(() -> consumeLoop(consumer), name + "-" + i);
worker.setDaemon(false);
worker.start();
workers.add(worker);
}
}
private void consumeLoop(LoanEventConsumer consumer) {
while (running || !queue.isEmpty()) {
LoanEvent event;
try {
event = queue.poll(100, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
if (event == null) {
continue;
}
try {
consumer.accept(event);
consumed.increment();
} catch (Exception e) {
failed.increment();
LOG.log(System.Logger.Level.WARNING,
"event failed loan=" + event.loanId() + " type=" + event.type(), e);
}
}
}
public boolean publish(LoanEvent event, Duration timeout) throws InterruptedException {
boolean ok = queue.offer(event, timeout.toMillis(), TimeUnit.MILLISECONDS);
if (ok) {
accepted.increment();
maxDepth.accumulateAndGet(queue.size(), Math::max);
} else {
rejected.increment();
}
return ok;
}
public Metrics metrics() {
return new Metrics(accepted.sum(), rejected.sum(), consumed.sum(),
failed.sum(), queue.size(), maxDepth.get());
}
public void close(Duration drainTimeout) {
running = false;
int pending = queue.size();
long before = consumed.sum();
long deadline = System.nanoTime() + drainTimeout.toNanos();
for (Thread worker : workers) {
long remaining = deadline - System.nanoTime();
if (remaining <= 0) {
break;
}
try {
worker.join(Math.max(1, remaining / 1_000_000));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
int abandoned = queue.size();
for (Thread worker : workers) {
if (worker.isAlive()) {
worker.interrupt();
}
}
LOG.log(System.Logger.Level.INFO,
"pipeline closed pending=" + pending + " drained=" + (consumed.sum() - before)
+ " abandoned=" + abandoned);
}
@Override
public void close() {
close(Duration.ofSeconds(10));
}
}
几个需要解释的决定：
- 消费循环条件用
running || !queue.isEmpty()。只写while (running)会在 close 时立刻退出，队列里的事件全部丢失。这样写在 close 之后仍会把剩余事件处理完，自然排空。 poll(100ms)而不是take()。take会一直阻塞，close 只能靠interrupt，而中断可能打断正在执行consumer.accept的线程。用带超时的poll，线程每 100ms 检查一次running，退出更干净。- 消费者异常不退出循环。单条事件失败只记
failed并继续，否则一条脏数据就能让整个流水线停摆。需要重试的事件应该有独立的重试队列或死信表。 maxDepth只在投递成功时更新。它记录的是「队列曾经到过多深」，是容量规划的依据；queued是当前值，两者不是一回事。
装配到 AppMain，顺序与 9.1 的工作池一致：
LoanEventListener audit = new AuditEventListener(auditRepository);
LoanEventPipeline auditPipeline = new LoanEventPipeline(
config.auditQueueCapacity(), 2, audit::onEvent, "loan-audit");
LoanService loanService = new LoanService(repositories, transactions, metrics, auditPipeline);
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
http.stop();
auditPipeline.close(Duration.ofSeconds(5));
dataSource.close();
}, "libraryd-shutdown"));
HTTP 先停，保证不再有新事件进来；再排空流水线；最后关连接池——如果先关连接池，排空过程中的审计写会全部失败。
常见错误与边界
错误一：用无界队列「避免拒绝」。 new LinkedBlockingQueue<>() 让生产者永远成功，代价是队列可以涨到 OOM。上面实验里无界版本快了 4 倍，但那是因为它没做事。拒绝不是故障，是系统在保护自己。
错误二：消费者抛异常后线程退出。 如果 consumeLoop 没有 try/catch，一条事件的异常会让这个消费者线程死掉。队列还在接收，消费者却越来越少，最后变成 1 个甚至 0 个，生产者被永久阻塞。至少要有「记失败 + 继续」的兜底。
错误三：close 时先 interrupt 再排空。 中断会让 poll 抛 InterruptedException，正在处理的事件被打断，队列里剩余的全部留下。正确顺序是「置停止标志 → 等排空 → 超时才中断」。
错误四：往队列里放可变对象。 生产者放进去之后又修改了对象的字段，消费者读到的可能是改了一半的状态。LoanEvent 用 record 且字段全 final，就是为了避免这个问题。
边界：队列容量不是并发度。 容量 32 只表示最多积压 32 条，不代表 32 个消费者。消费能力由 workerCount 和单条处理时间决定：2 个消费者、每条 5ms，吞吐上限约 400 条/秒，和容量无关。
边界：多实例部署时队列是进程内的。 每个实例一条队列，A 实例的审计事件不会跑到 B 实例。需要跨实例的顺序或去重，得用数据库或消息队列，不能靠 BlockingQueue。
小结
BlockingQueue的四组操作对应四种满/空语义；请求线程只用offer(e, timeout)，不用无超时的put。ArrayBlockingQueue单锁、内存可预测；LinkedBlockingQueue双锁、1P1C 下略快；无界版本快是因为没做背压。- 容量 32、2 消费者、每条 5ms 时，300 个事件接受 34 个、拒绝 266 个，队列被打满，背压可见。
- 优雅停止是「停止接收 → 排空 → 超时中断」，输出
pending、drained、abandoned三个数。 - 消费者要捕获单条异常并继续，否则一条脏数据能让整条流水线停摆。
- libraryd 用
ArrayBlockingQueue加LoanEventPipeline，不同事件类型配不同的offer超时。
队列解决了「一件事怎么异步做」，但启动预热、并发上限、分批对齐这些协调问题还需要别的原语。下一节看三个同步器各自适合什么。
阅读导航：上一节：5.1 并发容器：ConcurrentHashMap 与 CopyOnWriteArrayList · 下一节：5.3 同步器：CountDownLatch、Semaphore 与 CyclicBarrier 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。