Java HttpClient 怎么把响应体按行异步消费

Java HttpClient 要把文本响应按行异步消费，最直接的做法不是先用 ofString() 收完整个正文，而是把一个 Flow.Subscriber 交给 HttpResponse.BodyHandlers.fromLineSubscriber。订阅者会在 onNext 中收到一行文本，并能通过 Flow.Subscription.request(n) 控制下一批数据的需求量。

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

如果每行处理包含写库、解析或外部调用，建议先只请求一行，等异步任务完成后再调用 request(1)。这会把下游处理能力变成明确的需求信号，避免响应行无界堆积。

先选对响应体模型

HttpClient 提供的几个文本处理器看起来相近，但完成时机和资源责任不同：

Oracle API 明确说明，ofLines() 返回 HttpResponse 时正文可能尚未完全接收，返回的 Stream 必须关闭。它适合在某个工作线程中用 try-with-resources 主动遍历。若目标是让到达的每一行直接进入响应式订阅者，fromLineSubscriber() 的责任边界更清楚。

真正逐行消费要用 fromLineSubscriber

fromLineSubscriber 是 BodySubscriber 与文本型 Flow.Subscriber 之间的适配器。HTTP 响应字节先按字符集解码，再根据行分隔规则交付字符串。没有指定分隔符时，行划分方式与 BufferedReader.readLine() 一致。

预定义 BodyHandlers 不会替你检查 HTTP 状态码。生产代码应使用自定义 BodyHandler，根据 ResponseInfo 选择正常行订阅器或其他正文处理器，避免把 404、500 的错误页当作业务数据。

实现一个一次只接收一行的订阅者

下面的原创实现把每行工作提交给独立执行器。订阅建立后只请求一行；这一行处理成功，才请求下一行。若业务处理失败，则取消上游订阅并把异常传给完成信号。

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Flow;
import java.util.function.Consumer;

final class AsyncLineSubscriber implements Flow.Subscriber {
    private final Executor executor;
    private final Consumer lineTask;
    private final CompletableFuture completion = new CompletableFuture();
    private Flow.Subscription subscription;

    AsyncLineSubscriber(Executor executor, Consumer lineTask) {
        this.executor = Objects.requireNonNull(executor);
        this.lineTask = Objects.requireNonNull(lineTask);
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = Objects.requireNonNull(subscription);
        // 先只申请一行，把积压上限交给业务处理速度控制
        subscription.request(1);
    }

    @Override
    public void onNext(String line) {
        CompletableFuture
            .runAsync(() -> lineTask.accept(line), executor)
            .whenComplete((ignored, error) -> {
                if (error != null) {
                    // 业务失败时停止继续接收，并向调用方传播异常
                    subscription.cancel();
                    completion.completeExceptionally(error);
                    return;
                }
                // 当前行处理完成后才申请下一行
                subscription.request(1);
            });
    }

    @Override
    public void onError(Throwable error) {
        // 网络、解码或上游错误统一进入完成信号
        completion.completeExceptionally(error);
    }

    @Override
    public void onComplete() {
        // 上游正常结束且所有已申请行处理完毕
        completion.complete(null);
    }

    CompletableFuture completion() {
        return completion;
    }
}

public class LineClientDemo {
    public static void main(String[] args) {
        HttpClient client = HttpClient.newHttpClient();
        ExecutorService workers = Executors.newFixedThreadPool(4);

        HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create("https://example.com/events.ndjson"))
            .GET()
            .build();

        AsyncLineSubscriber subscriber = new AsyncLineSubscriber(
            workers,
            line -> persistLine(line) // 每行进入实际解析或持久化逻辑
        );

        HttpResponse.BodyHandler handler = responseInfo -> {
            if (responseInfo.statusCode() / 100 != 2) {
                // 非 2xx 响应仍需消费或丢弃正文，才能释放交换资源
                return HttpResponse.BodySubscribers.replacing(null);
            }
            // 明确使用 UTF-8，并采用常见换行规则拆分文本
            return HttpResponse.BodySubscribers.fromLineSubscriber(
                subscriber,
                ignored -> null,
                StandardCharsets.UTF_8,
                null
            );
        };

        client.sendAsync(request, handler)
            .thenCompose(response -> {
                if (response.statusCode() / 100 != 2) {
                    // 状态码异常不进入正常行处理完成链
                    return CompletableFuture.failedFuture(
                        new IllegalStateException("HTTP " + response.statusCode())
                    );
                }
                return subscriber.completion();
            })
            .whenComplete((ignored, error) -> {
                // 无论成功失败都关闭业务线程池
                workers.shutdown();
                if (error != null) {
                    error.printStackTrace();
                }
            })
            .join();
    }

    private static void persistLine(String line) {
        // 示例只展示消费边界，真实项目在这里解析或写入存储
        System.out.println(line);
    }
}

示例里的 URL 只是占位地址，替换为你的真实接口即可。重点是 AsyncLineSubscriber 的需求管理：它不会在 onNext 一进入就立即请求更多行，而是等待异步任务完成。

每处理完一行再请求下一行

Flow.Subscription.request(1) 是这段实现的核心。它把“允许上游再交付一行”变成显式额度。只要当前行仍在执行器里处理，订阅者就不增加额度。对于每行都要写数据库或调用慢服务的场景，这比一次请求无限数量更容易控制内存和并发压力。

如果每行处理非常轻，可以把额度从 1 调整为一个小批量，但不能只提高 request(n) 而忽略执行器队列。实际观察时应记录以下指标，而不是凭感觉判断：

• 业务执行器队列长度和活跃线程数；
• 单行处理耗时的中位数与高分位数；
• HTTP 响应读取是否长期停顿；
• 异常发生后订阅是否及时取消；
• 进程堆内存是否随响应时长持续增长。

本文没有给出虚构的压测数字。合理的批量大小取决于行大小、处理耗时和下游容量，应在真实接口和真实存储上测量。

状态码、字符集与分隔符要显式决定

Oracle 文档说明，预定义 BodyHandler 通常不检查状态码。示例在 ResponseInfo 阶段先做 2xx 判断，这样错误页不会进入正常订阅者。非成功响应仍需消费或丢弃正文，避免交换资源悬空。

字符集也不能总靠猜。若接口契约明确为 UTF-8，显式传入 StandardCharsets.UTF_8 最直观。若必须遵循响应头中的 charset，可以使用 BodyHandlers.fromLineSubscriber(subscriber) 的便利重载，让它按 ofString() 的规则选择字符集。

lineSeparator 传 null 时采用类似 BufferedReader.readLine() 的常见行划分；传空字符串会抛出 IllegalArgumentException。若协议规定固定分隔符，例如只允许 "\n"，可以在四参数版本中明确传入。

完成与异常链不要重复控制

sendAsync 返回 CompletableFuture>。使用 fromLineSubscriber 时，响应体由订阅者消费，正常完成后 future 才能形成最终响应。因此业务代码应建立一条清楚的完成链：

1. onError 接收网络、解码和上游异常；
2. 异步业务任务失败时主动 cancel()，并完成异常 future；
3. onComplete 只表示上游正常结束；
4. 最外层 whenComplete 统一关闭业务执行器和记录结果。

避免在多个回调里反复关闭同一资源、重复完成 future，或在 onNext 中抛出未捕获异常。把终止责任集中起来，长时间运行的文本流才容易排查。

ofLines 什么时候更简单

如果响应是有限文件，且你已经有一个专门的工作线程顺序处理，BodyHandlers.ofLines() 会更短。它返回懒读取的 Stream，但必须用 try-with-resources 关闭：

client.sendAsync(request, HttpResponse.BodyHandlers.ofLines())
    .thenAcceptAsync(response -> {
        // Stream 必须关闭，确保 HTTP 交换相关资源能够释放
        try (var lines = response.body()) {
            lines.forEach(LineClientDemo::persistLine);
        }
    });

这种写法是“异步取得响应后，在某个线程主动遍历 Stream”，而不是通过 Flow.Subscription 控制行需求。需要细粒度背压、取消和异步业务处理时，仍应选择 fromLineSubscriber。

相关问题

fromLineSubscriber 会自动检查 200 状态码吗？

不会。预定义 BodyHandlers 不检查状态码，应在自定义 BodyHandler 中查看 ResponseInfo.statusCode()。

onNext 可以直接做耗时数据库写入吗？

不建议阻塞响应处理线程。可把工作提交到独立执行器，并在完成后再请求下一行。

request(Long.MAX_VALUE) 可以吗？

语义上可以表达近似无界需求，但会失去逐行背压的保护。只有下游足够快且已确认不会堆积时才考虑。

响应中最后一行没有换行符会丢失吗？

按 BufferedReader.readLine() 风格划分时，结束前的最后一段文本仍可作为一行交付，不要求文件必须以换行符结尾。

这是处理 SSE 的完整方案吗？

不是。SSE 还包含字段拼接、空行事件边界、重连时间和 Last-Event-ID 等协议规则。本文只解决底层文本按行异步消费。

最终可以把选择原则压缩成一句话：小响应用 ofString，主动遍历有限文本用 ofLines，需要逐行回调和背压时用 fromLineSubscriber。后者配合小额度 request(n)，才能让 HTTP 接收速度与真实业务处理能力保持一致。