Java HttpClient BodyHandlers.ofPublisher 如何处理 Publisher 取消

我第一次把 BodyHandlers.ofPublisher() 接到流式接口时，最容易误判的一点是：HttpResponse 已经拿到了，并不代表响应体已经读完。这个处理器把响应体交给 Flow.Publisher>，真正的消费责任落在订阅者身上。

官方地址：https://docs.oracle.com/en/java/javase/26/docs/api/java.net.http/java/net/http/HttpResponse.BodySubscribers.html

我第一次踩坑：响应到了，不等于响应体读完了

BodyHandlers.ofPublisher() 返回的是一个把响应体转成 Publisher 的处理器。调用 sendAsync 后，响应头一到，CompletableFuture> 就可能完成；此时 response.body() 只是拿到一个等待消费的发布源。

我现在会先把边界写在代码旁边：只安排一个订阅者，并在 onSubscribe 里保存订阅。Publisher 只能订阅一次，第二次订阅会收到 IllegalStateException，不要为了“重试读取”再次调用 subscribe。

最小写法：让订阅者决定继续 request 还是 cancel

下面的写法按一个批次请求一个元素，能同时体现背压和提前停止。代码只展示消费协议，不把示例输出包装成已经在本机执行的证据。

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.Flow;
import java.util.concurrent.atomic.AtomicInteger;

HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
    .uri(URI.create("https://example.com/stream")) // 替换为实际流式接口
    .GET()
    .build();

client.sendAsync(request, HttpResponse.BodyHandlers.ofPublisher())
    .thenAccept(response -> response.body().subscribe(new Flow.Subscriber>() {
        private Flow.Subscription subscription;
        private final AtomicInteger batches = new AtomicInteger();

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription; // 保存唯一的订阅控制器
            subscription.request(1);           // 先申请一个批次，避免无界接收
        }

        @Override
        public void onNext(List items) {
            int current = batches.incrementAndGet();
            // 这里处理只读 ByteBuffer；不要把 items 当成可修改列表。
            if (current >= 8) {
                subscription.cancel();       // 业务只需要前 8 个批次
                System.out.println("cancelled after " + current + " batches");
                return;                        // 取消后不再等待 onComplete
            }
            subscription.request(1);           // 处理完当前批次再申请下一个
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("body failed: " + error.getMessage()); // 记录消费异常
        }

        @Override
        public void onComplete() {
            System.out.println("body complete: " + batches.get() + " batches"); // 正常读完
        }
    }))
    .exceptionally(error -> {
        System.err.println("request failed: " + error.getMessage()); // 处理请求阶段异常
        return null;
    });

提前停止时怎么判断：不要把 onComplete 当作取消回执

Flow.Subscription.cancel() 的语义是让发布方最终停止发送，属于尽力而为。规范明确说，取消后仍可能收到额外消息，也不保证收到 onComplete 或 onError。因此示例用本地计数和日志标记“我已经发出取消”，而不是等待某个回调来证明取消完成。

如果业务其实愿意读完整个响应，就不要在中途取消：每次处理 onNext 后继续 request(1)，让消费自然走到 onComplete。这通常更有利于释放交换资源和复用连接。确实只需要前几批时，再接受提前取消可能关闭底层 HTTP/1.1 连接或让 HTTP/2 流被重置的代价。

四个容易混淆的边界

另外，request(0) 或负数不是“暂时不取数据”的写法，而会触发 IllegalArgumentException。想暂停时保留订阅即可；想放弃当前响应才调用 cancel。

常见问题

为什么拿到 HttpResponse 后没有 body 内容？

因为 ofPublisher() 返回的是响应体发布源，不是已经拼接好的字符串或字节数组。必须对 response.body() 订阅并申请数据。

cancel 之后一定会触发 onError 吗？

不一定。取消后可能继续收到少量消息，也可能既没有 onComplete 也没有 onError；业务状态应由自己的取消分支记录。

为什么第二个订阅者读不到同一份响应？

这个 Publisher 设计为单订阅。要让多个消费者看到内容，应在第一次订阅时自行复制或转发数据，而不是再次订阅 HTTP 响应体。

我最后会按一个简单原则收口：能读完就读到终止信号；不能读完就明确取消，并把连接复用和回调不保证这两个代价写进调用方的设计里。