流式传输HTTP响应文件至请求时遭遇IOException问题排查

开发者阿泽

2026-6-16

问题描述

我正在编写一个程序，需要从一个API获取文件并发送至另一个API。为避免将整个文件存入内存的byte[]，同时减少昂贵的文件系统读写操作，我尝试将字节流直接从响应流式传输至请求中。该方案有时可行，但偶尔会抛出如下异常：

// Application Exception ... //
Caused by: java.io.UncheckedIOException: java.io.IOException: closed
    at de.trinext.lib.nf.api.util.MultiPartBodyPublisher$PartsIterator.hasNext(MultiPartBodyPublisher.java:121)
    at java.net.http/jdk.internal.net.http.RequestPublishers$IterablePublisher$ByteBufferIterator.hasNext(RequestPublishers.java:135)
    at java.net.http/jdk.internal.net.http.PullPublisher$Subscription$PullTask.run(PullPublisher.java:120)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$CompleteRestartableTask.run(SequentialScheduler.java:149)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$SchedulableTask.run(SequentialScheduler.java:207)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(SequentialScheduler.java:280)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(SequentialScheduler.java:233)
    at java.net.http/jdk.internal.net.http.PullPublisher$Subscription.request(PullPublisher.java:138)
    at java.net.http/jdk.internal.net.http.Stream$RequestSubscriber.trySend(Stream.java:1132)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$LockingRestartableTask.run(SequentialScheduler.java:182)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$CompleteRestartableTask.run(SequentialScheduler.java:149)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$TryEndDeferredCompleter.complete(SequentialScheduler.java:324)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$CompleteRestartableTask.run(SequentialScheduler.java:151)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$SchedulableTask.run(SequentialScheduler.java:207)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(SequentialScheduler.java:280)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(SequentialScheduler.java:233)
    at java.net.http/jdk.internal.net.http.Stream.signalWindowUpdate(Stream.java:967)
    at java.net.http/jdk.internal.net.http.WindowController.increaseStreamWindow(WindowController.java:275)
    at java.net.http/jdk.internal.net.http.Stream.incoming_windowUpdate(Stream.java:714)
    at java.net.http/jdk.internal.net.http.Stream.otherFrame(Stream.java:508)
    at java.net.http/jdk.internal.net.http.Stream.incoming(Stream.java:502)
    at java.net.http/jdk.internal.net.http.Http2Connection.processFrame(Http2Connection.java:1051)
    at java.net.http/jdk.internal.net.http.frame.FramesDecoder.decode(FramesDecoder.java:155)
    at java.net.http/jdk.internal.net.http.Http2Connection$FramesController.processReceivedData(Http2Connection.java:310)
    at java.net.http/jdk.internal.net.http.Http2Connection.asyncReceive(Http2Connection.java:859)
    at java.net.http/jdk.internal.net.http.Http2Connection$Http2TubeSubscriber.processQueue(Http2Connection.java:1756)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$LockingRestartableTask.run(SequentialScheduler.java:182)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$CompleteRestartableTask.run(SequentialScheduler.java:149)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$SchedulableTask.run(SequentialScheduler.java:207)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(SequentialScheduler.java:280)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(SequentialScheduler.java:233)
    at java.net.http/jdk.internal.net.http.Http2Connection$Http2TubeSubscriber.runOrSchedule(Http2Connection.java:1773)
    at java.net.http/jdk.internal.net.http.Http2Connection$Http2TubeSubscriber.onNext(Http2Connection.java:1800)
    at java.net.http/jdk.internal.net.http.Http2Connection$Http2TubeSubscriber.onNext(Http2Connection.java:1734)
    at java.net.http/jdk.internal.net.http.common.SSLTube$DelegateWrapper.onNext(SSLTube.java:210)
    at java.net.http/jdk.internal.net.http.common.SSLTube$SSLSubscriberWrapper.onNext(SSLTube.java:492)
    at java.net.http/jdk.internal.net.http.common.SSLTube$SSLSubscriberWrapper.onNext(SSLTube.java:295)
    at java.net.http/jdk.internal.net.http.common.SubscriberWrapper$DownstreamPusher.run1(SubscriberWrapper.java:316)
    at java.net.http/jdk.internal.net.http.common.SubscriberWrapper$DownstreamPusher.run(SubscriberWrapper.java:259)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$LockingRestartableTask.run(SequentialScheduler.java:182)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$CompleteRestartableTask.run(SequentialScheduler.java:149)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$SchedulableTask.run(SequentialScheduler.java:207)
    ... 3 more
Caused by: java.io.IOException: closed
    at java.net.http/jdk.internal.net.http.ResponseSubscribers$HttpResponseInputStream.current(ResponseSubscribers.java:464)
    at java.net.http/jdk.internal.net.http.ResponseSubscribers$HttpResponseInputStream.read(ResponseSubscribers.java:508)
    at java.base/java.io.InputStream.read(InputStream.java:220)
    at de.trinext.lib.nf.api.util.MultiPartBodyPublisher$PartsIterator.computeNext(MultiPartBodyPublisher.java:175)
    at de.trinext.lib.nf.api.util.MultiPartBodyPublisher$PartsIterator.hasNext(MultiPartBodyPublisher.java:119)
    ... 53 more
Caused by: java.io.IOException: fixed content-length: 2861131, bytes received: 1539620
    at java.net.http/jdk.internal.net.http.common.Utils.wrapWithExtraDetail(Utils.java:391)
    at java.net.http/jdk.internal.net.http.Http1Response$BodyReader.onReadError(Http1Response.java:676)
    at java.net.http/jdk.internal.net.http.Http1AsyncReceiver.checkForErrors(Http1AsyncReceiver.java:302)
    at java.net.http/jdk.internal.net.http.Http1AsyncReceiver.flush(Http1AsyncReceiver.java:268)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$LockingRestartableTask.run(SequentialScheduler.java:182)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$CompleteRestartableTask.run(SequentialScheduler.java:149)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$SchedulableTask.run(SequentialScheduler.java:207)
    at java.net.http/jdk.internal.net.http.HttpClientImpl$DelegatingExecutor.execute(HttpClientImpl.java:177)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(SequentialScheduler.java:282)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(SequentialScheduler.java:251)
    at java.net.http/jdk.internal.net.http.Http1AsyncReceiver.onReadError(Http1AsyncReceiver.java:516)
    at java.net.http/jdk.internal.net.http.Http1AsyncReceiver$Http1TubeSubscriber.onError(Http1AsyncReceiver.java:595)
    at java.net.http/jdk.internal.net.http.common.SSLTube$DelegateWrapper.onError(SSLTube.java:257)
    at java.net.http/jdk.internal.net.http.common.SSLTube$SSLSubscriberWrapper.complete(SSLTube.java:441)
    at java.net.http/jdk.internal.net.http.common.SSLTube$SSLSubscriberWrapper.onErrorImpl(SSLTube.java:511)
    at java.net.http/jdk.internal.net.http.common.SSLTube$SSLSubscriberWrapper.onError(SSLTube.java:525)
    at java.net.http/jdk.internal.net.http.common.SubscriberWrapper$DownstreamPusher.run1(SubscriberWrapper.java:294)
    at java.net.http/jdk.internal.net.http.common.SubscriberWrapper$DownstreamPusher.run(SubscriberWrapper.java:259)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$LockingRestartableTask.run(SequentialScheduler.java:182)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$CompleteRestartableTask.run(SequentialScheduler.java:149)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$SchedulableTask.run(SequentialScheduler.java:207)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(SequentialScheduler.java:280)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(SequentialScheduler.java:233)
    at java.net.http/jdk.internal.net.http.common.SubscriberWrapper.errorCommon(SubscriberWrapper.java:421)
    at java.net.http/jdk.internal.net.http.common.SSLFlowDelegate$Reader.errorCommon(SSLFlowDelegate.java:384)
    at java.net.http/jdk.internal.net.http.common.SSLFlowDelegate$Reader.processData(SSLFlowDelegate.java:517)
    at java.net.http/jdk.internal.net.http.common.SSLFlowDelegate$Reader$ReaderDownstreamPusher.run(SSLFlowDelegate.java:283)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$LockingRestartableTask.run(SequentialScheduler.java:182)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$CompleteRestartableTask.run(SequentialScheduler.java:149)
    at java.net.http/jdk.internal.net.http.common.SequentialScheduler$SchedulableTask.run(SequentialScheduler.java:207)
    at java.base/java.util.concurrent.ThreadPerTaskExecutor$TaskRunner.run(ThreadPerTaskExecutor.java:314)
    at java.base/java.lang.VirtualThread.run(VirtualThread.java:329)
Caused by: java.io.IOException: BUFFER_UNDERFLOW with EOF, 8570 bytes non decrypted.
    at java.net.http/jdk.internal.net.http.common.SSLFlowDelegate$Reader.processData(SSLFlowDelegate.java:487)
    ... 6 more

程序流程：

获取文件
将HttpResponse.BodyHandlers.ofInputStream()得到的InputStream保存至变量
将此InputStream作为Supplier传入MultiPartBodyPublisher实现
使用java.net.http.HttpRequest.Builder.POST(BodyPublisher)发送请求

解决方案

问题根源

从异常栈能看出核心问题是源API的响应流被提前关闭，导致读取时抛出closed异常。深层原因包括：

SSL解密失败：底层出现BUFFER_UNDERFLOW with EOF，说明SSL传输过程中连接意外中断，未接收完声明的Content-Length。
流生命周期不匹配：HttpResponseInputStream绑定原始HTTP连接，若源API连接因网络波动、超时或服务器主动断开，流会被强制关闭，而下游的MultiPartBodyPublisher还在尝试读取。
无重试机制：流式传输对网络稳定性要求高，单次请求遇到波动就会失败，没有重试逻辑兜底。

修复方案

1. 添加超时与重试策略

为源API请求配置合理超时时间，并针对网络类异常添加重试逻辑：

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

int maxRetries = 3;
HttpResponse<InputStream> response = null;
for (int i = 0; i < maxRetries; i++) {
    try {
        response = client.send(
                HttpRequest.newBuilder(URI.create(sourceApiUrl))
                        .GET()
                        .build(),
                HttpResponse.BodyHandlers.ofInputStream()
        );
        if (response.statusCode() == 200) break;
    } catch (IOException | InterruptedException e) {
        if (i == maxRetries - 1) throw e;
        Thread.sleep(1000 * (i + 1)); // 指数退避重试
    }
}

2. 绑定流与响应生命周期

HttpResponseInputStream依赖原始HTTP连接，需保证读取流过程中连接不被提前释放，保留HttpResponse引用直到读取完成：

try (HttpResponse<InputStream> response = client.send(request, HttpResponse.BodyHandlers.ofInputStream())) {
    InputStream inputStream = response.body();
    // 在此执行MultiPartBodyPublisher和POST请求逻辑
}

3. 优化MultiPartBodyPublisher异常处理

在自定义的MultiPartBodyPublisher实现中添加异常捕获，确保流异常时能正确释放资源：

@Override
protected ByteBuffer computeNext() {
    try {
        byte[] buffer = new byte[8192];
        int read = inputStream.read(buffer);
        if (read == -1) {
            inputStream.close();
            return endOfData();
        }
        return ByteBuffer.wrap(buffer, 0, read);
    } catch (IOException e) {
        try {
            inputStream.close();
        } catch (IOException ex) {
            // 忽略关闭异常
        }
        throw new UncheckedIOException(e);
    }
}

4. 可选：添加中间缓存

若网络稳定性极差，可临时用内存或磁盘缓存中转（小文件适用）：

ByteArrayOutputStream baos = new ByteArrayOutputStream();
byte[] buffer = new byte[8192];
int read;
while ((read = inputStream.read(buffer)) != -1) {
    baos.write(buffer, 0, read);
}
InputStream cachedStream = new ByteArrayInputStream(baos.toByteArray());
// 用cachedStream创建MultiPartBodyPublisher

内容的提问来源于stack exchange，提问作者xtay2