如何按指定字节拆分InputStream并分块提交至BodyPublisher？

当然可以实现你要的这种输入流，直接继承JDK自带的FilterInputStream就能快速封装，核心逻辑就是跟踪已读取的字节数，达到指定上限后返回流结束信号（-1）。

自定义SubsetInputStream实现 #

import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;

public class SubsetInputStream extends FilterInputStream {
    private final long maxBytesToRead;
    private long bytesReadSoFar;

    public SubsetInputStream(InputStream in, long chunkSize) {
        super(in);
        this.maxBytesToRead = chunkSize;
        this.bytesReadSoFar = 0;
    }

    @Override
    public int read() throws IOException {
        if (bytesReadSoFar >= maxBytesToRead) {
            return -1; // 达到字节上限，返回流结束
        }
        int result = super.read();
        if (result != -1) {
            bytesReadSoFar++;
        }
        return result;
    }

    @Override
    public int read(byte[] b) throws IOException {
        return read(b, 0, b.length);
    }

    @Override
    public int read(byte[] b, int off, int len) throws IOException {
        if (bytesReadSoFar >= maxBytesToRead) {
            return -1;
        }
        // 计算剩余可读取的字节数，避免超过chunkSize
        int bytesAvailable = (int) Math.min(len, maxBytesToRead - bytesReadSoFar);
        int bytesRead = super.read(b, off, bytesAvailable);
        if (bytesRead != -1) {
            bytesReadSoFar += bytesRead;
        }
        return bytesRead;
    }

    // 可选：重写skip方法，避免跳过超出chunkSize的字节
    @Override
    public long skip(long n) throws IOException {
        long skipBytes = Math.min(n, maxBytesToRead - bytesReadSoFar);
        long skipped = super.skip(skipBytes);
        bytesReadSoFar += skipped;
        return skipped;
    }
}

匹配需求的使用示例 #

InputStream is = // 你的原始输入流
long numBytes = // 总字节数
long chunkSize = // 每个分块的大小
long bytesRead = 0;

while (bytesRead < numBytes) {
    SubsetInputStream subsetIs = new SubsetInputStream(is, chunkSize);
    // 传入BodyPublisher构建请求
    HttpRequest request = HttpRequest.newBuilder()
            .uri(/* 你的请求地址 */)
            .POST(HttpRequest.BodyPublishers.ofInputStream(() -> subsetIs))
            .build();
    // 发送请求并处理响应
    // ...
    
    // 更新已读取字节数：如果能拿到实际发送的字节数，用实际值更准确；否则用chunkSize
    bytesRead += chunkSize;
}

关键注意点 #

• 原始InputStream不能被其他线程或代码同时读取，否则会破坏分块的位置连续性。
• 如果最后一块剩余字节不足chunkSize，SubsetInputStream会自动读取剩余所有字节后返回流结束，无需额外处理。
• 因为FilterInputStream直接复用原始流的读写指针，每次创建SubsetInputStream时，原始流的位置会自动延续上一次的结束位置。

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