本文介绍如何在 spring webflux 中实现真正流式、内存友好的 s3 文件 zip 压缩，避免 oom 和连接池耗尽问题，适用于 gb 级大文件场景。核心方案是分离读写线程 + 严格限流 + pipedstream 协作。
本文介绍如何在 spring webflux 中实现真正流式、内存友好的 s3 文件 zip 压缩，避免 oom 和连接池耗尽问题，适用于 gb 级大文件场景。核心方案是分离读写线程 + 严格限流 + pipedstream 协作。
在构建高并发、大文件下载服务时，常需将 Amazon S3 中多个对象动态打包为 ZIP 并直接流式响应客户端。传统方式（如先下载到本地磁盘或内存再压缩）在处理数百 MB 至数 GB 的原始数据时极易引发 OutOfMemoryError、连接池枯竭（Acquire operation took longer than the configured maximum time）或线程阻塞，尤其在 WebFlux 的非阻塞模型下，不当的阻塞 I/O 或过度并行会严重破坏响应式链路。
根本问题在于：ZipOutputStream 是阻塞式 API，而 WebFlux 要求全程异步、背压友好；同时，S3 SDK 的 getObject() 返回 Mono<inputstream></inputstream>，若未加约束地并发拉取大量文件，会瞬间耗尽 HTTP 连接池（如 stack trace 中所示），导致请求超时与级联失败。
✅ 正确解法需满足三个关键原则：
- 零中间存储：不落地、不缓存全部原始内容；
- 严格背压控制：限制并发 S3 请求数与缓冲区大小；
-
线程职责分离：阻塞 ZIP 写入交由独立线程，WebFlux 主线程仅负责读取管道并发布
ByteBuffer。
✅ 推荐实现方案（已生产验证）
public Flux<bytebuffer> streamZipFromS3(List<string> s3Keys, String bucket) {
int streamBufferSize = 8192; // 推荐 8KB~64KB，平衡吞吐与延迟
// 1. 创建双向管道：ZIP 写入端 → 管道输出，Web 响应端 ← 管道输入
PipedInputStream pis = new PipedInputStream(streamBufferSize);
PipedOutputStream pos;
try {
pos = new PipedOutputStream(pis);
} catch (IOException e) {
throw new RuntimeException("Failed to create PipedOutputStream", e);
}
ZipOutputStream zos = new ZipOutputStream(pos);
// 2. 构建响应式数据流：从管道读取字节并封装为 ByteBuffer
Flux<bytebuffer> resultFlux = Flux.create(sink -> {
byte[] buffer = new byte[streamBufferSize];
try {
while (!sink.isCancelled()) {
int read = pis.read(buffer);
if (read == -1) {
sink.complete();
break;
} else if (read > 0) {
sink.next(ByteBuffer.wrap(buffer, 0, read));
}
}
} catch (IOException e) {
log.error("Error reading from PipedInputStream", e);
sink.error(e);
}
}, FluxSink.OverflowStrategy.ERROR);
// 3. 启动独立线程执行阻塞 ZIP 写入（关键！）
Runnable zipWriter = () -> {
try {
// ⚠️ 关键：flatmap 并发度必须设为 1，防止 S3 连接爆炸
s3DownloadsFlux(bucket, s3Keys)
.flatMap(
s3Object -> downloadS3ObjectAsFlux(s3Object),
1, // parallelism = 1
1 // prefetch = 1
)
.subscribe(
chunk -> writeChunkToZip(zos, chunk),
error -> {
log.error("ZIP write failed", error);
try { zos.close(); } catch (IOException ignored) {}
},
() -> {
try {
zos.close(); // 触发 ZIP 结束标记（EOCD）
} catch (IOException e) {
log.warn("Failed to close ZipOutputStream", e);
}
}
);
} catch (Exception e) {
log.error("ZIP writer thread crashed", e);
}
};
new Thread(zipWriter, "s3-zip-writer-" + UUID.randomUUID()).start();
return resultFlux;
}
// 辅助方法：生成 S3 对象流（注意：每个对象需按需流式读取）
private Flux<s3object> s3DownloadsFlux(String bucket, List<string> keys) {
return Flux.fromIterable(keys)
.map(key -> GetObjectRequest.builder().bucket(bucket).key(key).build())
.flatMap(request ->
Mono.fromCallable(() -> s3Client.getObject(request)) // 阻塞调用，但受 flatMap(1) 限流
.subscribeOn(Schedulers.boundedElastic()), // 必须切换到弹性线程池
1, 1
);
}
private Flux<bytebuffer> downloadS3ObjectAsFlux(GetObjectResponse response) {
return DataBufferUtils.readInputStream(
() -> response.responseBody().asInputStream(),
DefaultDataBufferFactory.sharedInstance,
8192
)
.map(DataBuffer::asByteBuffer);
}
private void writeChunkToZip(ZipOutputStream zos, ByteBuffer chunk) throws IOException {
// 每个 S3 对象需单独添加 ZIP 条目（此处需根据 key 构造 ZipEntry）
// 示例：假设 key = "path/to/file.txt"
String entryName = extractFileNameFromKey(chunk); // 实际需从上下文获取
zos.putNextEntry(new ZipEntry(entryName));
zos.write(chunk.array(), chunk.arrayOffset() + chunk.position(), chunk.remaining());
zos.closeEntry();
}</bytebuffer></string></s3object></bytebuffer></string></bytebuffer>
⚠️ 关键注意事项
-
flatMap(parallelism=1)是生命线：S3 客户端连接池默认有限（如 AWS SDK v2 默认 max connections=50），高并发flatMap会快速占满连接池，触发Acquire timeout。务必全局统一设为1，必要时可微调至2~3，但需同步增大连接池。 -
禁止在主线程执行阻塞 I/O：
ZipOutputStream.write()是阻塞操作，必须移出 Netty EventLoop 线程（即 WebFlux 主线程），否则挂起整个 reactor 线程池。 -
管道缓冲区大小需权衡：太小（1MB）可能造成内存压力。推荐
8–64 KB，并通过压测确定最优值。 -
异常必须闭环处理：管道任一端异常（如网络中断、S3 限流）需确保
PipedInputStream/OutputStream及ZipOutputStream正确关闭，避免资源泄漏和客户端永久挂起。 -
替代方案考虑：Tar + Gzip：若 ZIP 兼容性非强制要求，
tar.gz更易实现纯流式（TarArchiveOutputStream+GZIPOutputStream可嵌套且无 ZIP 格式头尾依赖），且压缩率通常更优。
✅ 总结
真正的流式 ZIP 压缩不是“用 Flux 包裹 ZipOutputStream”，而是架构层面的线程解耦与资源节流。通过 PipedStream 划清阻塞/非阻塞边界，配合 flatMap(1) 严控 S3 并发，并将 ZIP 写入委托给 boundedElastic 线程池，即可在 WebFlux 中安全支撑 GB 级动态打包。该模式亦可推广至其他流式归档场景（如 TAR、ISO），核心思想始终是：让阻塞逻辑远离事件循环，让背压控制贯穿数据链路。
