队列设了上限，内存还是涨上去了
一个攒批写库的消费者：队列容量 100，生产者用
put，满了就阻塞。下游数据库慢了半秒，队列里一条积压都没有，生产者也没被阻塞，进程里却堆着 5000 条还没写进去的数据。队列的容量只管住了数据经过的四个位置中的一个。
「生产者 → 有界队列 → 消费者攒批 → 写入下游」是服务端最常见的异步结构：日志落盘、埋点上报、批量写库、给下游做削峰。它有五个边界经常被漏掉：一批最多多大、取出队列之后的数据有多少、多久必须刷出一次、一批里有一条失败怎么办、关闭时没写完的数据去哪。
本文在 JDK 25.0.4 上用闩锁和预先装好的队列逐个构造这些场景。实验见文末。
一、「取到空为止」的批没有上限
常见写法是先 take() 一条，再把队列里现有的全部取出来：
batch.add(queue.take());
queue.drainTo(batch); // 有多少取多少平时队列里只有几条，批很小。下游慢过一阵之后队列攒满，下一批就是整个积压。实测容量 1000 的队列积压 1000 条时：
| 写法 | 批数 | 最大一批 |
|---|---|---|
drainTo(batch) | 1 | 1000 条 |
drainTo(batch, 99) | 10 | 100 条 |
批的大小等于队列容量，会撞上下游的各种上限：一条 SQL 的最大包大小、单个事务的行数与锁持有时间、HTTP 请求体限制。而它恰好出现在下游刚恢复、最经不起大请求的时候。drainTo 有带上限的重载，批上限应该按下游能接受的大小来定，与队列容量无关。批量写库本身的取值见百万行数据导入。
二、队列有界，不等于在途有界
为了不让写入拖慢消费，一种写法是消费者取到一批后交给另一个线程异步写入，自己立刻回去取下一批：
writer.submit(() -> sink.write(batch)); // writer 是 Executors.newSingleThreadExecutor()newSingleThreadExecutor 的任务队列是无界的。消费者永远不会被下游拖住，于是有界队列永远不满，生产者永远不被阻塞。实测队列容量 100，下游卡住 500ms：
| 生产者放入 | 队列里 | 已取出、未写完 | 生产者 | |
|---|---|---|---|---|
| 不限在途 | 5000 条（全部） | 0 | 5000 | 没有被阻塞 |
| 在途额度 200 | 350 条 | 100 | 200 | 被阻塞 |
要限制的量是在途：已经离开队列、下游还没确认写入的数据。两种做法：
- 同步写：消费者写完一批再取下一批，在途最多就是一批。要并行写入时，开多个这样的消费者；
- 异步写加额度：交出一批之前先从
Semaphore申请这么多许可，写入完成（无论成功失败）后归还。上表第二行就是 200 个许可的结果：350 = 在途 200 + 队列 100 + 消费者手里等额度的一批 50。
监控也要跟着改。只看队列长度，上面第一行显示的是 0，看起来一切正常；应该同时暴露在途数量和最老一条数据的等待时间。线程池自身的队列是否有界，见线程池参数怎么定。
三、按时间刷出：截止时间属于这一批
只按大小刷出，流量小的时候数据会一直留在批里，所以还要一个时间条件。下面的写法看起来有，其实没有：
while (batch.size() < 50) {
Item next = queue.poll(100, MILLISECONDS); // 每次都重新等 100ms
if (next == null) break;
batch.add(next);
}只要下一条在 100ms 内到达，循环就继续。实测数据每 80ms 到一条：
poll(100ms) 都重新计时，只要下一条在 100ms 内到达，批就不会刷出，实测第一条等了 864ms；改成从批里第一条开始计算截止时间，每次只等剩余的时间，第一条等了 102ms| 写法 | 第一批 | 第一条等了 |
|---|---|---|
每次 poll(100ms) 重新计时 | 10 条 | 864ms |
| 按批的截止时间等待 | 2 条 | 102ms |
正确的写法是在取到批里第一条时算出截止时间，之后每次只等剩余的时间：
long deadline = System.nanoTime() + MILLISECONDS.toNanos(100);
while (batch.size() < 50) {
long remaining = deadline - System.nanoTime();
if (remaining <= 0) break;
Item next = queue.poll(remaining, NANOSECONDS);
if (next == null) break;
batch.add(next);
}用 System.nanoTime() 而不是 currentTimeMillis()：后者会随系统时钟校准跳变。同样的「循环里每次重新计时」问题也出现在带超时的等待里，见超时了、取消了，任务其实还在跑。
四、一批里有一条写不进去
100 条数据里第 38 条是坏数据（下游会因为它拒绝整批），批大小 50。三种处理方式的结果：
| 处理 | 写入 | 好数据丢失 | 消费者线程 | 之后的生产者 |
|---|---|---|---|---|
| 不捕获，异常逃出循环 | 0 | 49 条已取出未写入，另 50 条留在队列里没人处理 | 已退出 | 放满队列后被阻塞 |
| 捕获后只记日志 | 50 | 49 条（与坏数据同批） | 存活 | 正常 |
| 整批失败后逐条重试，坏数据进死信 | 99 | 0 | 存活 | 正常 |
第一行是最隐蔽的。消费者线程因为未捕获异常退出，没有任何组件会重启它；生产者把队列放满后阻塞在 put 上，表现为上游接口全部卡住，而问题出在很久以前的一条坏数据。消费循环的最外层要捕获异常并继续，同时暴露消费者是否存活的指标。
第二行说明「数据一出队列，队列就不再为它负责」。批是一个整体失败的单位，但里面的数据未必都有问题。区分两类失败：
- 整批都会失败（下游不可用、超时）：保留这一批，退避后重试，重试有上限；超过上限进死信或落盘，不能无限重试堵住后面的数据；
- 个别数据有问题：拆开处理。逐条重试最简单，批很大时可以二分。写不进去的进死信，留下原始内容和失败原因。
重试的前提是写入幂等，否则「超时但其实写成功了」的批会被重复写入。
五、关闭：先停上游，再排空
消费者已经取出 20 条、还没写入时收到关闭信号，队列里还有 80 条：
| 关闭方式 | 写入 | 丢失 |
|---|---|---|
| 中断后直接退出 | 0 | 100 |
| 中断后排空队列并写完 | 100 | 0 |
关闭的顺序是：上游停止放入 → 消费者把手里的批和队列里剩下的写完 → 确认消费者线程已经结束 → 关闭下游连接。顺序反了（先关连接）或者少了第一步（上游还在放入，排空没有终点），都会丢数据或者关不掉。给排空设一个期限，超时后把剩余条数记下来，而不是无限等。后台保存的如果不是队列里的数据而是一份内存状态，问题换成「保存的是哪一版」，见自动保存显示成功了，最后一次修改却没存上。
用「毒丸」（一个特殊对象表示结束）通知消费者时，有两个细节：
- 2 个消费者只放 1 颗毒丸，实测有 1 个消费者停不下来。每个消费者需要一颗，或者让消费者取到毒丸后再放回去；
- 队列已满时
offer(毒丸)返回 false，毒丸没有进队列。要用put，或者检查返回值并重试。
这里的「不丢」只针对正常关闭。数据在内存队列里时进程被杀，一定会丢；不能接受的话，生产者应该把数据交给有持久化与确认机制的消息队列，见 Kafka 不丢、不重与 Exactly Once。
六、一份参数清单
设计这类消费者时要写下来的数字和行为：
| 项 | 要回答的问题 |
|---|---|
| 队列容量 | 满了之后生产者是阻塞、丢弃还是报错 |
| 批上限 | 下游单次能接受的大小 |
| 最长等待 | 从批里第一条数据算起 |
| 在途上限 | 已取出、未确认的数据最多多少 |
| 失败处理 | 整批重试的次数与退避；个别失败的拆分方式；死信去向 |
| 关闭 | 停上游的方式、排空期限、超时后剩余数据的去向 |
| 监控 | 队列长度、在途数量、最老数据的等待时间、消费者是否存活、死信数量 |
配套实验
- codesphere-labs/java/bounded-batch-consumer：JDK 25.0.4 上批的上限、队列之外的在途量、两种按时间刷出的写法、三种批内失败处理、两种关闭方式与毒丸（验证记录）
参考资料
- JDK 25 API：BlockingQueue（
drainTo、offer、poll、put） - JDK 25 API：Executors.newSingleThreadExecutor（无界队列）
- JDK 25 API：Semaphore
- JDK 25 API：System.nanoTime