1. 引入依赖

    software.amazon.awssdk
    s3


    io.reactivex.rxjava2
    rxjava
    2.2.19

2. 创建 Publisher

private static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(10);

private Publisher createPublisher(final InputStream inputStream) {
    return new Publisher() {
        private ByteBuffer byteBuffer;

        @Override
        public void subscribe(Subscriber s) {
            s.onSubscribe(new Subscription() {
                boolean cancelled;
                @Override
                public void request(long n) {
                    if (n <= 0) {
                        s.onError(new IllegalArgumentException("The number of elements requested must be greater than 0"));
                        return;
                    }
                    if (cancelled) {
                        return;
                    }
                    try {
                        byte[] byteArray = new byte[(int) n];
                        int bytesRead = inputStream.read(byteArray);
                        if (bytesRead != -1) {
                            byteBuffer = ByteBuffer.wrap(byteArray, 0, bytesRead);
                            s.onNext(byteBuffer);
                        } else {
                            s.onComplete();
                        }
                    } catch (IOException e) {
                        s.onError(e);
                    }
                }

                @Override
                public void cancel() {
                    cancelled = true;
                }
            });
        }
    };
}

3. 执行异步上传

public void putObjectAsyncWithPublisher(String bucketName, String key, InputStream inputStream,
                                            long contentLength, String contentType, String contentEncoding) {
    S3AsyncClient s3AsyncClient = S3AsyncClient.builder().build();

    PutObjectRequest putObjectRequest = PutObjectRequest.builder()
            .bucket(bucketName)
            .key(key)
            .contentLength(contentLength)
            .contentType(contentType)
            .contentEncoding(contentEncoding)
            .build();

    Publisher requestBodyPublisher = createPublisher(inputStream);
    CompletableFuture future = s3AsyncClient.putObject(putObjectRequest, AsyncRequestBody.fromPublisher(requestBodyPublisher));

    future.whenComplete((resp, err) -> {
        if (err != null) {
            err.printStackTrace();
        } else {
            System.out.println("Async putObject Success. ETag: " + resp.eTag());
        }
    });

    future.join();
    s3AsyncClient.close();
}