Java HttpClient 流式读取 NDJSON：ofLines、背压与连接关闭

接口返回的不是完整的整块JSON，而是一行一条订单事件，之前很多开发者遇到的问题是服务端持续写入，客户端却要等全部内容传输完才开始解析，结果内存一路涨、连接迟迟释放不了，出了异常行也很难定位。Java 11 自带的 HttpClient 可以用 BodyHandlers.ofLines() 把响应转成按行消费的流，但它只解决“怎么拿到单行数据”的问题，消费速度管控、坏数据处理和连接收尾逻辑，仍要由业务代码明确落地规则。

先确定 NDJSON 的输入边界

NDJSON（Newline Delimited JSON）把每个 JSON 对象单独放在一行，比如订单推送服务可能返回的内容格式如下：

{"id":"o-1001","state":"PAID"}
{"id":"o-1002","state":"SHIPPED"}

它和 [{...},{...}] 这类嵌套 JSON 数组完全不是一回事。数组需要等整个结构闭合后才能完整解析，而 NDJSON 收到第一行之后就可以立刻开始处理。下面的示例统一约定响应使用 UTF-8 编码、每行一个独立对象、空行直接跳过，单行超过 64 KiB 或累计接收超过 10000 行就直接停止接收。

先把这类边界规则写进代码，后面的连接管控策略才有落地依据，不然所谓的“流式处理”很容易演变成另一种没有上限的全量缓存。

用 BodyHandlers.ofLines() 搭建最小处理链

下面给出的基础客户端写法只保留最清晰的执行路径：发起请求、按行读取、解析提交，示例中的 acceptLine 是业务回调接口，实际项目里可以直接替换成批量写入器的实现。

HttpClient client = HttpClient.newBuilder()
        .connectTimeout(Duration.ofSeconds(3))
        .build();

HttpRequest request = HttpRequest.newBuilder(URI.create(endpoint))
        .timeout(Duration.ofSeconds(30))
        .header("Accept", "application/x-ndjson")
        .GET()
        .build();

HttpResponse> response = client.send(
        request, HttpResponse.BodyHandlers.ofLines());

if (response.statusCode() != 200) {
    response.body().close();
    throw new IOException("NDJSON status=" + response.statusCode());
}

try (Stream lines = response.body()) {
    AtomicInteger count = new AtomicInteger();
    lines.filter(line -> !line.isBlank()).forEach(line -> {
        if (count.incrementAndGet() > 10_000) {
            throw new IllegalStateException("too many NDJSON lines");
        }
        OrderEvent event = parseAndCheck(line);
        acceptLine(event);
    });
}

这里有两个很容易被遗漏的细节：第一，非 200 状态的响应也可能携带 body 内容，不能直接抛异常就把流丢在后台不管；第二，Stream 只是懒加载的读取视图，try 代码块执行结束时才会关闭底层响应资源。不要在外层直接把所有行收集成 List，那样就完全失去了流式处理的核心优势。

明确慢消费者场景下的背压边界

ofLines() 不会自动替业务做排队处理，也不会主动限制下游业务的处理速度。如果 acceptLine 里包含数据库写入、远程调用或磁盘IO操作，消费速度慢于网络数据到达速度时，连接库和业务侧的缓冲都可能出现无限制积压。

一个非常实用的落地做法是把解析后的行数据按批次交给有限容量的队列，让生产端在队列满时进入等待状态，这种“主动等待”就是明确的压力传递信号，不会无限制把每行解析后的对象都留在内存里：

BlockingQueue> batches = new ArrayBlockingQueue(8);
List batch = new ArrayList(100);

try (Stream lines = response.body()) {
    lines.filter(line -> !line.isBlank()).forEach(line -> {
        batch.add(parseAndCheck(line));
        if (batch.size() == 100) {
            putBatch(batches, List.copyOf(batch));
            batch.clear();
        }
    });
    if (!batch.isEmpty()) {
        putBatch(batches, List.copyOf(batch));
    }
}

队列容量设为8、批量提交大小设为100只是起始参考值，不是通用最优解。要结合单行数据大小、落库耗时和允许的延迟做压测后再调整。如果生产端等待时间持续上涨，优先检查数据库批量提交逻辑和消费者数量，不要直接把队列阈值改得更大。

分别处理解析失败、超时和取消场景

某一行数据格式非法，不代表整条连接的所有数据都失效。建议先记录当前行号和内容摘要，再按接口双方的约定选择直接跳过、放入隔离队列或终止本次同步，注意不要把完整的敏感数据 payload 直接写进错误日志。

try (Stream lines = response.body()) {
    AtomicLong lineNo = new AtomicLong();
    lines.forEach(line -> {
        long current = lineNo.incrementAndGet();
        try {
            acceptLine(parseAndCheck(line));
        } catch (RuntimeException badLine) {
            log.warn("ndjson line rejected, lineNo={}, reason={}",
                    current, badLine.getMessage());
            saveToQuarantine(current, line);
        }
    });
} catch (UncheckedIOException networkError) {
    markPartialSync(networkError.getMessage());
}

请求超时只覆盖请求发送阶段的等待时长，不等同于业务处理的总超时。如果下游处理可能出现长时间阻塞，要给批次处理器单独设置时限，同时保留“已接收行数、已提交行数、隔离行数”三个核心统计指标。主动取消任务时，要让读取流正常离开 try 代码块，底层连接才能及时回收释放。

常见误区：流式不等于无限接收

上线前的速查与验证

本地可以搭一个分段输出的测试服务验证整条链路：每 200 毫秒发送一行数据，中间插入一行非法 JSON，再让消费端故意休眠一段时间。检查日志里是否能正常打印行号、部分成功状态和最终的连接关闭动作；把响应中途截断后，确认同步结果标记为未完成，而不是被误判为执行成功。

• 请求层：连接超时和响应超时都配置明确的数值。
• 输入层：限制单行最大长度、累计字节数和总行数上限。
• 处理层：队列设置最大容量，批量提交配置失败重试或数据隔离出口。
• 收尾层：正常结束、解析异常、网络断开、任务取消四类场景都能正常关闭 body。

相关问题

NDJSON 能不能直接用 readAllBytes 读取？

数据量很小且本身有严格大小上限时可以用，但它会等完整响应返回后一次性加载全部内容到内存。持续事件流、数据导出任务和大响应场景更适合逐行流式读取。

为什么不直接打开 Stream.parallel() 做并行处理？

并行消费会改变数据提交顺序，也可能瞬间把下游服务压力拉满。先通过有限批次和可观测指标确认真实瓶颈点，再判断是否需要拆分多消费者处理。

服务端断开连接后已经处理的行怎么保证一致性？

用事件 ID 或同步游标做幂等键，系统里区分保存“已接收”和“已提交”两个状态，重试时从最后一个已确认的位置继续同步，不要盲目把整段响应重复处理一遍。

小结

Java HttpClient 的 BodyHandlers.ofLines() 非常适合把 NDJSON 转换成逐行处理链路，但可靠性保障全来自外围边界规则：输入大小有限、消费速度可控、坏行可隔离、超时规则清晰、响应流必关闭。把这些规则落地到代码和监控指标里，流式读取才不会只是把一次性内存溢出问题，换成一个长期连接占用的隐患。