S3AsyncClient与AsyncResponseTransformer下载背压维护问题

核心解决方案：实现受背压控制的S3响应转换 #

AsyncResponseTransformer.toPublisher()不支持背压，会导致S3客户端主动推送全量数据到内存，这是问题根源。改用自定义Transformer将S3响应流转换为按需拉取的Flux，从上游控制数据推送节奏：

@GetMapping(path="/{filekey}")
Mono<ResponseEntity<Flux<ByteBuffer>>> downloadFile(@PathVariable("filekey") String filekey) {    
    GetObjectRequest request = GetObjectRequest.builder()
      .bucket(s3config.getBucket())
      .key(filekey)
      .build();

    // 自定义Transformer，基于InputStream实现背压控制
    AsyncResponseTransformer<GetObjectResponse, Flux<ByteBuffer>> transformer = AsyncResponseTransformer.create(
        (response, inputStream) -> {
            checkResult(response);
            // 按8KB分批读取，仅当下游请求时才拉取数据
            return Flux.using(
                () -> inputStream,
                is -> Flux.fromStream(InputStreamUtils.readByteBuffers(is, 8192)),
                InputStream::close
            );
        }
    );

    return Mono.fromFuture(s3client.getObject(request, transformer))
      .map(flux -> {
          GetObjectResponse response = transformer.response().join();
          String filename = getMetadataItem(response, "filename", filekey);            
          return ResponseEntity.ok()
            .header(HttpHeaders.CONTENT_TYPE, response.contentType())
            .header(HttpHeaders.CONTENT_LENGTH, Long.toString(response.contentLength()))
            .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"" + filename + "\"")
            .body(flux);
      });
}

关键细节： #

• Flux.using确保InputStream在流结束后正确关闭，避免资源泄漏
• 按固定大小（如8KB）分批读取数据，让下游的背压信号直接传递到S3客户端，仅当下游消费完一批数据后才读取下一批
• 彻底避免S3客户端一次性将全量文件加载到直接内存

辅助优化措施 #

1. 调整Netty直接内存参数：

   • 添加-Dio.netty.leakDetection.level=advanced：开启高级泄漏检测，定位潜在内存泄漏点
   • 添加-Dio.netty.recycler.maxCapacityPerThread=0：禁用Netty对象回收器，加快直接内存释放
   • 保持-Dio.netty.maxDirectMemory在2GB左右即可，无需盲目调大

2. 限制并发下载数：
   1核CPU节点无法支撑40个并行500MB下载，通过Reactor的并发限制控制请求量：

   @GetMapping(path="/{filekey}")
   Mono<ResponseEntity<Flux<ByteBuffer>>> downloadFile(@PathVariable("filekey") String filekey) {    
       return Mono.just(filekey)
         .transform(ConcurrencyLimiter.create(10)) // 限制10个并发下载
         .flatMap(key -> {
             // 上述自定义Transformer的下载逻辑
         });
   }
   

为什么之前的限流方法无效？ #

Flux.from(response).rateLimit(1)这类下游限流仅作用于消费端，无法影响上游的S3客户端。toPublisher()返回的是无背压的Publisher，S3会主动将全量数据推送到Netty直接缓冲区，即使下游限流，内存已经被占满，因此无法解决溢出问题。

可行性结论 #

完全可行。自定义Transformer实现的按需拉取机制，能将单连接的直接内存占用控制在几十MB以内，即使面对慢消费者也不会溢出。配合并发限制和Netty参数调整，可稳定支撑高并发下载场景。

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