Java Flow.Publisher 背压怎么传到数据库批处理：Subscription 请求量与取消边界

批量导入服务最容易出现一种假象：上游消息还在持续到达，数据库连接池却已经排队。此时给线程池再加线程，通常只会把等待藏得更深。Java 9 引入的 java.util.concurrent.Flow 把这件事拆成了一个可观察的契约：Subscriber 通过 Subscription.request(long) 声明自己还能接收多少项，消费完成后再申请下一批。

把数据库一次能稳定处理的批量大小作为请求量，例如 request(20)；不要在订阅时直接请求 Long.MAX_VALUE。批次失败或任务关闭时调用 cancel()，并把取消状态传到数据库写入层。

先把 Flow 的背压责任分清

Flow.Publisher 负责生产并通知订阅者，Flow.Subscriber 负责接收，真正把两者连起来的是 Flow.Subscription。订阅成功后，Publisher 会通过 onSubscribe 把 Subscription 交给 Subscriber；在 Subscriber 调用 request 之前，它不应该假设自己会收到无限数据。

这里的“20”只是一次需求量，不代表 Publisher 一定已经把 20 条数据放进内存，也不代表数据库已经提交。它表达的是消费方当前愿意再接收的额度，Publisher 仍可能因为资源不足、错误或完成状态而少发。

图中只保留本节的三个真实节点：Flow.Publisher 产生数据，Subscription 传递控制，request(20) 表示消费方新增 20 项需求量。

让数据库批次成为请求节拍

下面的 Subscriber 不把每条记录立即写库，而是先在内存里收集到 20 条。批次提交成功后，再调用 request(20) 申请下一批；这样上游的未完成需求量不会随着运行时间无限增长。

final class BatchSubscriber implements Flow.Subscriber {
    private final int batchSize = 20;
    private final List batch = new ArrayList(batchSize);
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(batchSize);
    }

    @Override
    public void onNext(Record item) {
        batch.add(item);
        if (batch.size() == batchSize) {
            writeBatch(batch);
            batch.clear();
            subscription.request(batchSize);
        }
    }

    @Override
    public void onError(Throwable error) {
        closeBatch();
    }

    @Override
    public void onComplete() {
        writeRemaining(batch);
        closeBatch();
    }
}

示例里的 writeBatch 必须是可观测的批处理动作：成功才补请求，失败走 onError 或主动取消。真实项目中还要处理不足一整批时的 onComplete，否则最后几条记录会停留在内存里。

第二张图绑定这段代码的三个节点：onNext 把记录放入批次，数据库批次成功后再申请额度，cancel用于终止后续消费。

request 的边界不能靠经验猜

request(long n) 接收的是新增需求量。按照 Java SE 25 的接口契约，传入小于等于 0 的值会导致 Subscriber 收到带有 IllegalArgumentException 的 onError。因此批量大小应在配置加载时就校验，避免把空批次误转成 request(0)。

不要把 Long.MAX_VALUE 当成“性能更好”的默认值。它相当于近似无限需求，会让背压失去意义；只有下游确实有独立的有界队列、并且已经在别处完成流量控制时，才有理由采用这种策略。

取消是状态变化，不是瞬时橡皮擦

cancel() 的语义是尽力让订阅停止，官方文档明确允许调用后仍收到额外消息。因此取消分支不能只依赖“再也不会进 onNext”这个假设。可以用一个关闭标志拒绝新批次，并让数据库写入检查任务上下文。

private final AtomicBoolean cancelled = new AtomicBoolean();

void stop() {
    if (cancelled.compareAndSet(false, true)) {
        subscription.cancel();
        closeBatch();
    }
}

public void onNext(Record item) {
    if (cancelled.get()) {
        return;
    }
    batch.add(item);
    // 达到 batchSize 后写库；取消状态在写入前再次检查
}

停止顺序的重点是幂等：先改变本地状态，再通知 Subscription，最后关闭批次资源。这样即便在途消息晚到，也不会重新开启写库流程。若数据库驱动提供取消或超时接口，还应把同一个任务的关闭信号继续向下传递。

用三条测试证明背压真的生效

第一条测试记录每次 request 的参数，确认初始值和成功提交后的补请求都等于批次大小；第二条让 writeBatch 抛错，确认失败后没有新的 request；第三条在批次中间调用 cancel，确认剩余在途项不会开启新的数据库批次。

如果需要验证上游是否尊重请求量，可以使用一个测试 Publisher 记录已发出的项目数，并在每次 onNext 后检查“已发送 - 已完成”不超过当前需求量。这个断言比只看总处理条数更能抓住提前生产的问题。

常见问题与核对方式

request(20) 会阻塞当前线程等待 20 条数据吗？

不会。它只是增加未满足需求量；数据何时到达、是否异步以及资源不足时如何处理，由 Publisher 的实现决定。

取消后为什么还会收到 onNext？

取消是尽力停止，已经排队或正在派发的消息可能继续到达。Subscriber 应用本地关闭状态保护数据库批次。

数据库批次应该多大？

从连接池、事务日志、单条写入成本和失败重试代价一起估算。先用 20 这类可解释的小值压测，再根据提交耗时和内存曲线调整。

把请求量和下游承载力绑定起来

Flow 的价值不在于换一种回调写法，而在于把“还能接多少”变成调用方可见的控制信号。让 request(batchSize) 跟随数据库成功提交，让错误停止补请求，让 cancel() 同时关闭本地批次和底层资源，背压才会从接口契约落到真实的数据路径。