使用Netty HttpClient的 s3AsyncClient存储的文件已损坏

后端开发 2026-07-11

我正在尝试把通过HTTP下载的文件转发给s3AsyncClient。
重要的是要保持异步，且尽量实现零拷贝，因此我使用了netty HttpClient 和 S3AsyncClient，并通过 Publisher<ByteBuffer> 传递文件。

但在运行时，对于较大的文件，存储在S3上的文件似乎被损坏，它们最终比源文件短一些，而且某些字符似乎也丢失了。

当前我的设置如下：

Flux<ByteBuffer> publisher = reactorHttpClient
                .get()
                .uri(fileUrl)
                .responseContent()
                .asByteBuffer();

 s3AsyncClient.putObject(builder -> builder.bucket(bucketName).key(fileName).build(),
                        AsyncRequestBody.fromPublisher(publisher))
                .whenComplete(...);

我的s3Client设置：

SdkAsyncHttpClient nettyHttpClient = NettyNioAsyncHttpClient.builder()
            .maxConcurrency(100)
            .build();

S3AsyncClientBuilder s3AsyncClientBuilder = S3AsyncClient.builder()
                .multipartEnabled(true)
                .httpClient(nettyHttpClient)
                .build();

解决方案

我相当确定你应该使用 multipartUpload。然而，我只有针对旧版AWS SDK 1.x、使用同步调用的示例代码。

应该是这样的：

var multipart = s3AsyncClient.createMultipartUpload(builder -> builder.bucket(...).key(...)).join();

publisher.index()
                .map(tuple -> {
                    int partNumber = Math.toIntExact(tuple.getT1() + 1);
                    ByteBuffer partBuffer = tuple.getT2();
                    return s3AsyncClient.uploadPart(builder -> builder.bucket(...)
                            .key(...)
                            .uploadId(multipart.uploadId())
                            .partNumber(partNumber)
                            .contentLength((long) partBuffer.remaining()), AsyncRequestBody.fromByteBuffersUnsafe(partBuffer));
                })

也许更好使用S3Transfer Manager

S3TransferManager transferManager = S3TransferManager.builder().s3Client(s3AsyncClient).build();
transferManager.upload(UploadRequest.builder()
                .putObjectRequest(...)
        .requestBody(AsyncRequestBody.fromPublisher(publisher)).build())