基于WebFlux实时压缩S3多文件的流式实现问题

问题背景 #

需要从S3批量读取对象，实时生成Zip/TGZ文件并以Flux<ByteBuffer>返回，文件体积可达数百MB甚至数GB，无法将所有内容加载到内存或磁盘。当前用PipedInputStream/PipedOutputStream能处理小文件，但处理1GB级别的大文件时崩溃。

错误分析 #

从堆栈跟踪来看，核心问题有两个：

1. AWS连接池资源耗尽：

   Unable to execute HTTP request: Acquire operation took longer than the configured maximum time. This indicates that a request cannot get a connection from the pool within the specified maximum time.
   并发读取S3对象时，连接池配置不足导致请求超时。

2. PipedInputStream/PipedOutputStream的局限性：
   这对类依赖线程间管道通信，大流量场景下易出现背压处理不当，引发内存溢出或阻塞。

解决方案 #

一、基于WebFlux的流式Zip实现 #

1. 自定义流式Zip生成逻辑

结合ZipOutputStream与WebFlux背压机制，避免一次性加载所有数据：

public Flux<ByteBuffer> streamZipFromS3(List<S3ObjectSummary> objects, S3AsyncClient s3Client) {
    return Flux.create(sink -> {
        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
             ZipOutputStream zos = new ZipOutputStream(baos)) {

            // 逐个处理S3对象，避免并发写入ZipOutputStream（非线程安全）
            for (S3ObjectSummary obj : objects) {
                zos.putNextEntry(new ZipEntry(obj.getKey()));

                // 流式读取S3对象并写入Zip流
                s3Client.getObject(GetObjectRequest.builder().bucket(obj.getBucketName()).key(obj.getKey()).build(),
                        AsyncResponseTransformer.toBytes())
                        .thenAccept(response -> {
                            try {
                                zos.write(response.asByteArray());
                                zos.closeEntry();
                                // 发送生成的Zip片段
                                sink.next(ByteBuffer.wrap(baos.toByteArray()));
                                baos.reset();
                            } catch (IOException e) {
                                sink.error(e);
                            }
                        })
                        .join();
            }

            zos.finish();
            sink.next(ByteBuffer.wrap(baos.toByteArray()));
            sink.complete();
        } catch (IOException e) {
            sink.error(e);
        }
    });
}

2. 响应式优化实现

用Flux.concatMap逐个处理对象，全流程贴合响应式流：

public Flux<ByteBuffer> reactiveStreamZip(List<S3ObjectSummary> objects, S3AsyncClient s3Client) {
    return Flux.fromIterable(objects)
            .concatMap(obj -> {
                return Mono.fromCallable(() -> new ZipEntry(obj.getKey()))
                        .flatMapMany(entry -> {
                            // 读取S3对象的数据流
                            Flux<ByteBuffer> s3Data = s3Client.getObject(
                                            GetObjectRequest.builder().bucket(obj.getBucketName()).key(obj.getKey()).build(),
                                            AsyncResponseTransformer.toPublisher())
                                    .flatMapMany(Flux::from);

                            // 依次发送Zip条目头、文件数据、条目结束标记
                            return Flux.concat(
                                    Flux.just(createEntryHeader(entry)),
                                    s3Data,
                                    Flux.just(createEntryFooter())
                            );
                        });
            })
            // 发送Zip文件结束标记
            .concatWith(Flux.just(createZipFooter()));
}

// 生成Zip条目头的ByteBuffer
private ByteBuffer createEntryHeader(ZipEntry entry) throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    try (ZipOutputStream zos = new ZipOutputStream(baos)) {
        zos.putNextEntry(entry);
    }
    return ByteBuffer.wrap(baos.toByteArray());
}

// 生成Zip条目结束标记的ByteBuffer
private ByteBuffer createEntryFooter() throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    try (ZipOutputStream zos = new ZipOutputStream(baos)) {
        zos.closeEntry();
    }
    return ByteBuffer.wrap(baos.toByteArray());
}

// 生成Zip文件结束标记的ByteBuffer
private ByteBuffer createZipFooter() throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    try (ZipOutputStream zos = new ZipOutputStream(baos)) {
        zos.finish();
    }
    return ByteBuffer.wrap(baos.toByteArray());
}

二、TGZ格式的流式实现（更适合大文件） #

TGZ（Tar+Gzip）天生适合流式处理，无需提前知晓文件大小，结合Apache Commons Compress实现：

public Flux<ByteBuffer> streamTgzFromS3(List<S3ObjectSummary> objects, S3AsyncClient s3Client) {
    return Flux.create(sink -> {
        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
             GzipCompressorOutputStream gzos = new GzipCompressorOutputStream(baos);
             TarArchiveOutputStream taos = new TarArchiveOutputStream(gzos)) {

            for (S3ObjectSummary obj : objects) {
                TarArchiveEntry entry = new TarArchiveEntry(obj.getKey());
                entry.setSize(obj.getSize()); // 复用S3返回的文件大小
                taos.putArchiveEntry(entry);

                // 流式读取S3对象并写入Tar流
                s3Client.getObject(GetObjectRequest.builder().bucket(obj.getBucketName()).key(obj.getKey()).build(),
                        AsyncResponseTransformer.toBytes())
                        .thenAccept(response -> {
                            try {
                                taos.write(response.asByteArray());
                                taos.closeArchiveEntry();
                                sink.next(ByteBuffer.wrap(baos.toByteArray()));
                                baos.reset();
                            } catch (IOException e) {
                                sink.error(e);
                            }
                        })
                        .join();
            }

            taos.finish();
            sink.next(ByteBuffer.wrap(baos.toByteArray()));
            sink.complete();
        } catch (IOException e) {
            sink.error(e);
        }
    });
}

TGZ优势：

• Tar格式支持边写边处理，无需提前统计所有文件大小
• Gzip压缩可流式执行，内存占用更低

三、AWS客户端配置优化 #

解决连接池超时问题，调整S3AsyncClient的连接池参数：

S3AsyncClient s3Client = S3AsyncClient.builder()
        .httpClientBuilder(NettyNioAsyncHttpClient.builder()
                .maxConcurrency(50) // 增加最大并发连接数
                .connectionTimeout(Duration.ofSeconds(30)) // 延长连接超时时间
                .connectionAcquisitionTimeout(Duration.ofSeconds(60)) // 延长连接获取超时时间
                .build())
        .build();

根据实际请求量调整maxConcurrency，避免超出系统文件描述符限制。

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