Flow request 怎样实现反压

在 Java Flow 里，request(n) 表示订阅者愿意再接收多少个数据项，不是让线程睡眠一段时间，也不是直接指定多少字节。没有调用它，Publisher 通常不会发送 onNext；调用过大的值，又可能让处理速度跟不上的消费者积压资源。

官方文档：https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/util/concurrent/Flow.html

排查反压时可以记住一条规则：先在 onSubscribe 建立一个有限窗口，消费到阈值后再补充窗口；如果已经无法继续消费，就取消订阅。下面只围绕 Java SE 25 的 Flow 契约和 HttpClient 响应体订阅展开。

先看 request 到底控制什么

最常见的现象是订阅成功了，却一直没有 onNext。这不一定是线程或网络故障，可能只是 Subscriber 没有向 Subscription 表达需求。Oracle 文档把 Subscription 定义为 Publisher 与 Subscriber 之间的控制连接，只有请求后的数据项才会交付。

这里的“项”由 Publisher 的泛型决定。对 Flow.Publisher 来说，一个需求通常对应一次字符串回调；对 HTTP BodySubscriber 来说，需求对应的是一次 List 回调，而不是一个固定字节数。这个区分是后面选择窗口大小的前提。

把需求窗口写成可复查的规则

request(n) 会把 未满足需求 增加 n。n 小于等于零时，订阅者会收到带有 IllegalArgumentException 的 onError；Long.MAX_VALUE 可以视为近似无限需求。连续调用不是覆盖，而是继续增加需求，所以补请求时要根据已经消费的数量设计窗口。

import java.util.concurrent.Flow;

final class WindowSubscriber implements Flow.Subscriber {
    private final int windowSize;
    private int remaining;
    private Flow.Subscription subscription;

    WindowSubscriber(int windowSize) {
        if (windowSize 

这段代码使用固定窗口，逻辑容易复查，但每个窗口耗尽后才补一次，吞吐和延迟取决于窗口大小。若希望减少“需求完全见底”的空档，可以在消费到半窗时补充半窗，让未满足需求大致保持在半窗到整窗之间；核心仍是先消费再补充，不能在每次回调无条件叠加大额 request。

在 BodySubscriber 中按窗口补请求

HttpClient 的自定义 HttpResponse.BodySubscriber 本质上是 Flow.Subscriber>。因此它的 onNext 收到的是有序的 ByteBuffer 列表，列表和其中的只读缓冲区交给订阅者后不会再被 HTTP Client 使用。这里的 request 窗口控制的是列表回调数量，不能把它当成响应体字节上限。

import java.net.http.HttpResponse;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Flow;

final class CountingBodySubscriber implements HttpResponse.BodySubscriber {
    private final CompletableFuture result = new CompletableFuture();
    private Flow.Subscription subscription;
    private long bytes;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        // 先建立一个有限需求；这里的单位是 List 回调
        this.subscription = subscription;
        subscription.request(8);
    }

    @Override
    public void onNext(List buffers) {
        // ByteBuffer 只在当前回调中统计，不把它保存到异步任务后继续使用
        for (ByteBuffer buffer : buffers) {
            bytes += buffer.remaining();
        }
        // 消费完当前数据后补充下一批需求，避免响应流停在半路
        subscription.request(8);
    }

    @Override
    public void onError(Throwable error) {
        // 失败时完成阶段异常结束，同时不再请求后续数据
        result.completeExceptionally(error);
    }

    @Override
    public void onComplete() {
        // 正常结束时交付统计结果
        result.complete(bytes);
    }

    @Override
    public java.util.concurrent.CompletionStage getBody() {
        return result;
    }
}

实际项目中如果订阅者无法继续处理，应该调用保存的 subscription.cancel()，而不是既不 request 也不 cancel。Oracle 对 BodySubscriber 的提醒是：继续请求直到 onComplete 或 onError，或者明确取消；在响应体尚未耗尽时取消，可能导致底层 HTTP 连接关闭，后续请求无法复用该连接。

遇到停滞、积压和提前结束怎么判断

request(1) 是不是最安全？它最容易观察和控制，但每项一次请求可能增加协调开销；适合单步处理或成本很高的消费逻辑，不是所有场景的默认最优值。

可以在 onNext 开头补 request 吗？可以表达新的需求，但要先确认当前项已经被接收且处理逻辑不会因需求叠加而超过缓冲能力。通常把补请求放在本次消费边界之后，更容易解释“还剩多少未完成工作”。

request(n) 能限制 HTTP 响应的字节数吗？不能直接这样理解。对 BodySubscriber，n 约束的是下一批 Flow 项；若需要字节容量边界，应结合专门的 BodySubscriber 能力或在消费逻辑中按 ByteBuffer 剩余量处理。

总结一下：request 是需求计数，窗口大小是资源和处理能力之间的取舍；反压是否闭环，要同时看到初始化请求、消费后的补请求，以及异常或无法继续消费时的完成/取消路径。

Flow.Subscription 官方文档：https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/util/concurrent/Flow.Subscription.html

BodySubscriber 官方文档：https://docs.oracle.com/en/java/javase/25/docs/api/java.net.http/java/net/http/HttpResponse.BodySubscriber.html