Java AWS S3 SDK2如何实现S3异步批量上传下载

S3服务端本身不支持单次HTTP请求提交多个独立对象的上传/下载，AWS SDK v2也没有封装对应的“一键批量”方法，你之前了解的multipart upload是针对单个大文件拆分分片的机制，和多独立小文件的批量场景完全不匹配，不用在SDK里翻找专门的批量接口了。基于你已经在用的S3AsyncClient做受控并发调度，就能实现需要的批量上传/下载能力，小文件场景下性能足够。

• 先调整S3AsyncClient的基础配置，别用默认值
  SDK默认带的Netty异步HTTP客户端最大并发连接数只有50，批量场景很容易被打满堵死，初始化时提前调整参数适配IO密集型场景：

  S3AsyncClient s3AsyncClient = S3AsyncClient.builder()
          .httpClientBuilder(NettyNioAsyncHttpClient.builder()
                  .maxConcurrency(120) // 按服务带宽、CPU核数调整，小文件场景建议设80-200
                  .connectionMaxIdleTime(Duration.ofSeconds(30))
                  .writeTimeout(Duration.ofSeconds(60))
                  .readTimeout(Duration.ofSeconds(60)))
          .credentialsProvider(DefaultCredentialsProvider.create())
          .region(Region.of("你的桶所在区域"))
          .build();
  

• 批量请求核心是做受控并发，不要无限制循环提交异步请求，不然很容易撑爆内存、打满连接池，根据你的项目依赖选一种实现就行：
  1. 轻量场景：自定义ForkJoinPool配合并行流，不需要额外引入依赖，适合批量任务总量在千级以内的场景

     // 自定义并发线程池，不要用公共ForkJoinPool默认的CPU核数-1的配置，对IO任务来说并发太低
     ForkJoinPool batchUploadPool = new ForkJoinPool(100);
     List<FileTask> taskList = // 你的待处理任务列表，包含本地文件路径、对应S3对象key
     List<CompletableFuture<PutObjectResponse>> futures = batchUploadPool.submit(
             () -> taskList.parallelStream()
                     .map(task -> s3AsyncClient.putObject(
                             PutObjectRequest.builder()
                                     .bucket("你的桶名")
                                     .key(task.getS3Key())
                                     .build(),
                             AsyncRequestBody.fromFile(task.getLocalFile())
                     ))
                     .toList()
     ).join();
     // 等待所有任务执行完成，单独处理每个请求的异常
     CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
     

  2. 生产场景：用反应式流做背压控制，如果项目已经引入Reactor/RxJava这类反应式框架，直接用框架封装的并发控制能力，自带背压、失败隔离，不会出现OOM问题，以Reactor为例：

     Flux.fromIterable(taskList)
             .flatMap(task -> Mono.fromFuture(() -> s3AsyncClient.putObject(
                             PutObjectRequest.builder()
                                     .bucket("你的桶名")
                                     .key(task.getS3Key())
                                     .build(),
                             AsyncRequestBody.fromFile(task.getLocalFile())
                     ))
                     .onErrorResume(e -> {
                         // 单文件失败处理逻辑：打日志、加入重试队列都可以，不会影响其他文件上传
                         System.err.printf("文件%s上传失败，原因：%s%n", task.getS3Key(), e.getMessage());
                         return Mono.empty();
                     }),
                     100 // 固定并发数，根据实际场景调整
             )
             .blockLast();
     

• 批量下载的实现逻辑和上传完全一致：构造对应GetObjectRequest，用同样的受控并发逻辑提交getObject异步请求，拿到响应的ResponseInputStream<GetObjectResponse>之后写入本地文件即可。
• 小文件场景额外优化点：
  • 优先用SDK自带的AsyncRequestBody.fromFile、AsyncResponseTransformer.toFile方法处理文件读写，SDK内部做了文件句柄、内存缓冲区的复用，比自己读字节数组性能好很多
  • 单文件体积小于1MB时可以适当调高并发数，单文件大于10MB时适当降低并发，避免出口带宽被打满
  • 失败重试只针对单个失败的文件做，不要整批重跑
  • 客户端和S3桶同区域部署时，关闭path-style寻址，用默认的虚拟主机寻址方式可以减少DNS解析开销

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