我离开 Java 已经有几年了,最近很兴奋地看到新的 java.net.http.HttpClient 和 AWS Java SDK 2.0 支持非阻塞异步操作。几年前,我在会议演讲中听说过反应式编程的概念,但实践机会不多。
我有一个问题,似乎很适合尝试这种编程风格:基本上,我想通过 HTTP 下载大量文件(比如 10,000 个)并将它们写回到 S3。
我已经使用 failsafe 实现了针对非阻塞异步 http GET 的重试,并且可以轻松地通过 S3 异步客户端将其与上传组合起来(请参见下面的草图)。
然而,我不确定如何正确限制程序的内存使用:如果文件下载速度比写回到 S3 的速度快,就没有机制施加背压并防止内存不足异常。
我熟悉一些传统的阻塞解决方案来解决这个问题 - 例如使用信号量来限制并发下载数量,或将下载内容写入某个有界阻塞队列,S3上传线程将从中拉取。然而,如果我要使用这样的阻塞机制来应用反压力,则会让我质疑首先使用非阻塞IO的优势。
是否有更具代表性的“响应式”方法来实现相同的目标?
我有一个问题,似乎很适合尝试这种编程风格:基本上,我想通过 HTTP 下载大量文件(比如 10,000 个)并将它们写回到 S3。
我已经使用 failsafe 实现了针对非阻塞异步 http GET 的重试,并且可以轻松地通过 S3 异步客户端将其与上传组合起来(请参见下面的草图)。
然而,我不确定如何正确限制程序的内存使用:如果文件下载速度比写回到 S3 的速度快,就没有机制施加背压并防止内存不足异常。
我熟悉一些传统的阻塞解决方案来解决这个问题 - 例如使用信号量来限制并发下载数量,或将下载内容写入某个有界阻塞队列,S3上传线程将从中拉取。然而,如果我要使用这样的阻塞机制来应用反压力,则会让我质疑首先使用非阻塞IO的优势。
是否有更具代表性的“响应式”方法来实现相同的目标?
import net.jodah.failsafe.Failsafe;
import net.jodah.failsafe.RetryPolicy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.core.async.AsyncRequestBody;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.services.s3.model.PutObjectResponse;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
public class BackupClient {
private static final Logger LOGGER = LoggerFactory.getLogger(BackupClient.class);
private final HttpClient httpClient = HttpClient.newBuilder().build();
private final S3AsyncClient s3AsyncClient = S3AsyncClient.create();
public runBackup(List<URI> filesToBackup) {
List<CompletableFuture<PutObjectResponse>> futures = filesToBackup.stream()
.map(backupClient::submitBackup)
.collect(Collectors.toList());
futures.forEach(CompletableFuture::join);
}
private CompletableFuture<PutObjectResponse> submitBackup(URI uri) {
return sendAsyncWithRetries(uri, HttpResponse.BodyHandlers.ofString())
.thenCompose(httpResponse -> s3AsyncClient.putObject(PutObjectRequest.builder()
.bucket("my-bucket")
.key(uri.toASCIIString())
.build(), AsyncRequestBody.fromString(httpResponse.body())));
}
private <T> CompletableFuture<HttpResponse<T>> sendAsyncWithRetries(URI uri, HttpResponse.BodyHandler<T> handler) {
final HttpRequest request = HttpRequest.newBuilder()
.uri(uri)
.timeout(Duration.ofMinutes(2))
.GET()
.build();
final var retryPolicy = new RetryPolicy<HttpResponse<T>>()
.withMaxRetries(4)
.withDelay(Duration.ofSeconds(1))
.handleResultIf(response -> 200 != response.statusCode());
return Failsafe.with(retryPolicy)
.getStageAsync(context -> {
if (context.getAttemptCount() > 0) {
LOGGER.error("Retry " + context.getAttemptCount() + " for " + uri);
}
return this.httpClient.sendAsync(request, handler);
});
}
}