books

《Java 并发编程·JUC 源码与容器》14.2 批量聚合与背压

用数量阈值加时间阈值做批量窗口，用 drainTo 一次持锁取走整批并唤醒阻塞的生产者；再用有界队列和 CallerRunsPolicy 演示背压的真实形状：40 条里 7 条被拒、调用线程亲自执行任务，并给出批量失败后的二分重投边界与幂等前提。

Leeting Yan 8 分钟阅读 3927 字

承接的问题

通知扇出是一条条来的，但落库、计费、写审计日志这类动作一条条做太贵。于是有了「攒一批再处理」：攒够 50 条或者等满 100 ms 就提交一次。问题也随之而来——生产者比消费者快时，攒的批次放在哪？答案只能是有界队列，而有界队列一旦满，就必须做个选择：阻塞生产者、拒绝任务，或者让生产者自己干。

这一节把批量窗口、drainTo、有界队列与 CallerRunsPolicy 串起来，回答「背压到底是什么形状」。数据全部来自本机 JDK 25 实跑。

批量窗口需要两个触发条件

只按数量触发，低流量时最后几条会一直躺在队列里；只按时间触发，高峰期又会退化成一条一批。两个都要：

static final int MAX_BATCH = 5;
static final long WINDOW_NANOS = TimeUnit.MILLISECONDS.toNanos(50);

static void consume(ArrayBlockingQueue<Long> queue, List<Integer> sizes) throws InterruptedException {
    List<Long> batch = new ArrayList<>(MAX_BATCH);
    while (true) {                                             // 真实实现由关闭标志控制
        Long first = queue.poll(200, TimeUnit.MILLISECONDS);   // 空闲时最多等 200 ms
        if (first == null) {
            continue;                                          // 真实实现里这是空闲超时，不是退出
        }
        batch.add(first);
        long windowEnd = System.nanoTime() + WINDOW_NANOS;     // 窗口从第一条到达开始算

        while (batch.size() < MAX_BATCH) {
            int n = queue.drainTo(batch, MAX_BATCH - batch.size());
            if (n > 0) {
                continue;                                      // 一次持锁搬空能搬的
            }
            long left = windowEnd - System.nanoTime();
            if (left <= 0) {
                break;                                         // 时间阈值到了
            }
            Long next = queue.poll(left, TimeUnit.NANOSECONDS);
            if (next == null) {
                break;
            }
            batch.add(next);
        }
        sizes.add(batch.size());
        flush(batch);                                          // 一次批量写入
        batch.clear();
    }
}

static void flush(List<Long> batch) throws InterruptedException {
    Thread.sleep(40);                                          // 冒充一次批量落库
}

这段循环有三个设计点。

第一，窗口从第一条元素到达开始计时，不是在循环开头计时。否则前一批处理完、队列空着的 200 ms 也会被算进窗口，下一批的第一条刚到就被时间阈值催走，批量完全失效。

第二，先 drainTo 再 poll。drainTo 是一次加锁、循环搬运，ArrayBlockingQueue 的实现里把 count 次取元素放在同一个 lock 内完成，比循环调用 poll() 少 n-1 次加锁解锁。队列里有货时它一次就能填满批次，只有队列暂时空了才需要 poll(left) 去等剩余窗口。

第三，drainTo 顺手唤醒了被阻塞的生产者。JDK 25 的 drainTo 在归还元素之后，会对 notFull 条件按取走数量逐个 signal（只在有等待者时）。所以消费者搬走一批，阻塞在 put/offer(timeout) 上的生产者立刻就能继续。这也是「批量消费天然缓解背压」的来源：一次搬走 5 个，等于一次腾出 5 个槽位。

drainTo 也有边界：它不阻塞，队列空就返回 0；它把元素 add 到目标集合，如果目标集合抛异常，已搬走的元素不会放回队列，只是把 count 和 takeIndex 修正到已搬走的位置——也就是已经交给你的就是你的，不会再出现在队列里。

有界队列才是背压，offer 的返回值是信号

「背压」常被误解成「限速」。限速是主动控制发送速率（令牌桶），背压是下游处理不过来时，让上游的提交动作本身变慢或失败。前者需要一个计时器，后者只需要一个有界队列加一次阻塞或拒绝。

for (int i = 1; i <= TOTAL; i++) {
    long t0 = System.nanoTime();
    boolean ok = queue.offer(i, 20, TimeUnit.MILLISECONDS);   // 满则最多等 20 ms
    long cost = System.nanoTime() - t0;
    if (!ok) {
        rejected.incrementAndGet();                            // 拒绝也是一条明确信号
    } else if (cost > TimeUnit.MILLISECONDS.toNanos(1)) {
        waited.incrementAndGet();                              // 阻塞过，说明队列曾满
    }
}

本机真实输出（capacity=8、maxBatch=5、window=50 ms、消费者每批处理 40 ms；单次运行，load average 113，耗时不作基准）：

=== 批量聚合：capacity=8, maxBatch=5, window=50 ms ===
[   5.2 ms] flush batch#1 size=5 trigger=SIZE first=1 last=5
[  27.5 ms] producer item 14 REJECTED, queue=8
[  48.9 ms] producer item 15 REJECTED, queue=8
[  75.5 ms] producer item 16 REJECTED, queue=8
[  77.1 ms] flush batch#2 size=5 trigger=SIZE first=6 last=10
[  77.1 ms] producer item 17 blocked 1.4 ms, queue=4
[  99.9 ms] producer item 22 REJECTED, queue=8
[ 127.6 ms] flush batch#3 size=5 trigger=SIZE first=11 last=18
[ 127.6 ms] producer item 23 blocked 27.4 ms, queue=4
[ 152.5 ms] producer item 28 REJECTED, queue=8
[ 172.6 ms] flush batch#4 size=5 trigger=SIZE first=19 last=24
[ 172.6 ms] producer item 29 blocked 19.8 ms, queue=4
[ 198.2 ms] producer item 34 REJECTED, queue=8
[ 213.7 ms] flush batch#5 size=5 trigger=SIZE first=25 last=30
[ 213.7 ms] producer item 35 blocked 15.2 ms, queue=4
[ 235.7 ms] producer item 40 REJECTED, queue=8
[ 264.1 ms] flush batch#6 size=5 trigger=SIZE first=31 last=36
[ 359.2 ms] flush batch#7 size=3 trigger=TIME first=37 last=39
offered=40 rejected=7 blockedEvents=4 batches=7 sizes=[5, 5, 5, 5, 5, 5, 3]

几条能对上的账：

  • offered=40，rejected=7，sizes 总和 5*6+3=33，40 = 33 + 7。被拒的元素不会出现在任何一批里。
  • 每次 REJECTED 时 queue=8，说明确实撞到了容量上限；排空后 queue=4，说明消费者一次 drainTo 搬走了 4 个。
  • 最后一批是 size=3, trigger=TIME：生产结束、队列空了，时间阈值把半批催了出来。没有时间阈值，这 3 条会一直等到下次流量到来。
  • 批次里出现 first=11 last=18 size=5，中间跳过的 14、15、16 正是被拒的那几个，last 是最后一条入队的元素而不是 first+size-1。

背压的形状在这个输出里非常直白：生产者被阻塞 15–27 ms，或者直接拿到 false。队列长度是唯一的缓冲区，满了之后压力必然回到生产者身上，线程池和消费者都无法凭空变快。

CallerRunsPolicy 是一种天然背压

线程池的拒绝策略里，CallerRunsPolicy 最不像「拒绝」：它把任务还给提交者自己跑。

// JDK 25 源码摘录，已删减
public static class CallerRunsPolicy implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

这段代码的效果是：池满时，提交线程被这个任务占用，提交循环自然变慢；等它跑完，池里通常也有空位了。不需要任何限速器，背压由「谁提交谁执行」自动完成。

本机实测（pool=2、queue=2、10 个任务、每个睡 30 ms）：

=== CallerRunsPolicy：pool=2, queue=2 ===
[   4.8 ms] task  5 runs on main
[   4.9 ms] task  1 runs on pool-26
[   4.9 ms] task  2 runs on pool-27
[  35.5 ms] task  3 runs on pool-26
[  36.6 ms] task  4 runs on pool-27
[  43.3 ms] task  8 runs on main
[  67.4 ms] task  6 runs on pool-27
[  67.6 ms] task  7 runs on pool-26
[  98.5 ms] task  9 runs on pool-27
[ 106.0 ms] task 10 runs on pool-26
producer submit loop finished at 147.1 ms, terminated=true, callerRuns=2

任务 5 和任务 8 在 main 上执行，callerRuns=2。提交循环到 147.1 ms 才结束——10 个任务、2 个线程、每个 30 ms，理论下界 150 ms，生产者自己干了两个，所以没有超出多少。如果不带背压（无界队列），提交循环会在几毫秒内结束，然后 10 个任务在堆里排队，内存和延迟一起涨。

CallerRunsPolicy 的代价同样明确：

场景后果
提交者是处理用户请求的线程用户请求被迫执行一个后台任务，P99 直接抬升
提交者持有数据库连接或锁任务执行时间被拉长，可能死锁或把连接池耗干
提交者是事件循环线程（单线程）整个事件循环停摆，所有连接都卡住
池已 shutdown任务被静默丢弃，没有日志、没有异常

所以它适合「提交者本来就是后台线程、且任务可以随便推迟」的场景，例如批处理任务的生产者。面向用户的请求线程，宁可显式拒绝并返回 503，也不要在这里偷偷跑一个 30 ms 的任务。

批量失败之后的重新投递边界

一批 50 条，第 37 条写失败了，怎么办？三个选项，边界完全不同：

策略重复风险适用条件
整批重投前 36 条可能写两次下游幂等（有唯一键/去重表）
逐条重投无重复，但失败那条会反复失败批次内条目互相独立
二分重投只重投失败的一半需要定位坏条目，且下游幂等

二分是最常用的折中：失败时把批次劈成两半分别重试，直到单条仍失败就丢进死信队列。它把重投次数从 n 降到 log n，同时能精确定位毒条目：

void flushWithSplit(List<Item> batch, int depth) {
    try {
        repository.saveAll(batch);          // 一次事务写一批
        return;
    } catch (RuntimeException e) {
        if (batch.size() == 1 || depth >= 6) {
            deadLetter.addAll(batch);       // 单条仍失败：隔离，不再重试
            return;
        }
        int mid = batch.size() / 2;
        flushWithSplit(batch.subList(0, mid), depth + 1);
        flushWithSplit(batch.subList(mid, batch.size()), depth + 1);
    }
}

边界在于**「失败」的粒度必须和事务边界一致**。如果 saveAll 内部逐条自动提交，前一半成功、后一半失败，二分重投就会把成功的那一半再写一遍。要么整批一个事务，要么每条带幂等键。这是批量处理里最容易在故障时才暴露的问题。

另外，subList 是视图不是拷贝，重投过程中原 List 不能被清空或修改；flush 之后立刻 batch.clear() 是安全的，因为递归已经返回。

在 libraryd-conc 里落地

库存盘点（stocktake）是天然适合批量的场景：盘点机把每个 ISBN 的实盘数推上来，服务端按批更新 Book.stock，再写一条审计。批量窗口设为 50 条 / 100 ms，队列容量 256，满了就拒绝并让盘点机重传：

package com.plume.conc.core;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public final class StocktakeBatcher implements AutoCloseable {
    private final ArrayBlockingQueue<StocktakeEvent> queue =
            new ArrayBlockingQueue<>(256);
    private final StocktakeSink sink;
    private final List<StocktakeEvent> deadLetter = new CopyOnWriteArrayList<>();
    private final AtomicLong rejected = new AtomicLong();
    private final Thread worker;

    public StocktakeBatcher(StocktakeSink sink) {
        this.sink = sink;
        this.worker = Thread.ofPlatform().name("stocktake-batcher").start(this::loop);
    }

    public boolean submit(StocktakeEvent event) {
        if (!queue.offer(event)) {          // 有界队列满：立刻拒绝，由盘点机重传
            rejected.incrementAndGet();
            return false;
        }
        return true;
    }

    private void loop() {
        List<StocktakeEvent> batch = new ArrayList<>(50);
        try {
            while (!Thread.currentThread().isInterrupted()) {
                StocktakeEvent first = queue.poll(100, TimeUnit.MILLISECONDS);
                if (first == null) {
                    continue;
                }
                batch.add(first);
                queue.drainTo(batch, 49);
                long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(100);
                while (batch.size() < 50) {
                    long left = deadline - System.nanoTime();
                    if (left <= 0) {
                        break;
                    }
                    StocktakeEvent next = queue.poll(left, TimeUnit.NANOSECONDS);
                    if (next == null) {
                        break;
                    }
                    batch.add(next);
                }
                flushWithSplit(batch, 0);
                batch.clear();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private void flushWithSplit(List<StocktakeEvent> batch, int depth) {
        try {
            sink.saveAll(batch);                // 一个事务写一批并记审计
        } catch (RuntimeException e) {
            if (batch.size() == 1 || depth >= 6) {
                deadLetter.addAll(batch);
                return;
            }
            int mid = batch.size() / 2;
            flushWithSplit(batch.subList(0, mid), depth + 1);
            flushWithSplit(batch.subList(mid, batch.size()), depth + 1);
        }
    }

    public long rejected() {
        return rejected.get();
    }

    public List<StocktakeEvent> deadLetter() {
        return List.copyOf(deadLetter);
    }

    @Override
    public void close() {
        worker.interrupt();
    }

    public interface StocktakeSink {
        void saveAll(List<StocktakeEvent> batch);
    }
}

submit 用非阻塞的 offer 而不是 put：盘点机是外部设备，让它拿到 false 后重传，比让 HTTP 线程卡在队列上更好。如果调用方是本进程的后台线程，换成 offer(event, 50, MILLISECONDS) 让背压以延迟的形式体现，也合理。两种都是背压，区别只是压力表现成「拒绝」还是「变慢」。

这个设计的代价

批量把延迟换成了吞吐。 窗口 100 ms 意味着单条数据最多等 100 ms 才被处理。对盘点这种离线业务无所谓，对「用户点了借阅要立刻看到库存变化」就不合适。

队列长度决定了能吸收多少突发。 容量 256 在 10 倍突发下能撑一会儿，在 100 倍突发下只是把 OOM 推迟几秒。队列满之后的拒绝率才是真正要监控的指标，队列长度不是。

二分重投依赖幂等。 前面已经说过：没有幂等键或单事务，重投就是重复写。死信队列也不是垃圾桶，必须有告警和人工处理入口，否则坏数据只是被藏起来了。

什么时候不要用

不要用批量聚合去掩盖一个本来就慢的下游。批量能降低每次调用的固定开销，但如果单条处理本身就要 200 ms，攒 50 条只会让第 50 条等 10 秒。

不要在要求严格顺序的流上用多线程批量消费。批量提交本身会打乱跨批次的顺序，需要顺序就得单消费者加单分区，那时批量只是减少了事务次数。

不要把 CallerRunsPolicy 用在 Web 请求线程池的拒绝策略上。它会让 Tomcat 的工作线程去执行后台任务，表现为「偶发的接口超时」，而且极难从线程栈上看出来。

不要在没有背压信号的地方用无界队列。LinkedBlockingQueue 不传容量就是 Integer.MAX_VALUE，生产速度超过消费速度时，它的表现是内存缓慢上涨直到 OOM，而不是拒绝——这比拒绝危险得多。

小结

批量窗口要同时有数量阈值和时间阈值，窗口从第一条到达开始算；drainTo 一次持锁搬走整批，并顺带唤醒阻塞的生产者。背压的本质是有界队列加阻塞或拒绝，CallerRunsPolicy 把「谁提交谁执行」变成了天然的减速器，但只适合后台线程。批量失败后的重投必须和事务边界、幂等键一起设计，否则省下的调用次数会以重复数据的形式还回来。

阅读导航：上一节 14.1 超时、重试与熔断的并发实现 ｜ 下一节 14.3 并发限流与信号量池

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《Java 消息中间件·事件驱动架构》附录 B 架构决策记录模板
  2. 《Java 消息中间件·事件驱动架构》附录 A 术语表
  3. 《Java 消息中间件·事件驱动架构》15.3 全书验收