Spring WebFlux 执行并行 HTTP 请求并反序列化响应

Question

我有一个<代码>列表

这些要求应该同时提出。所有请求完成后，我想有一个＜code＞列表

所以我创建了一个方法来发出HTTP请求

public Flux<JsonNode> performGetRequest(String url) {
    WebClient webClient = WebClient.create(String.format("%s%s", API_BASE_URL, url));

    return webClient.get()
            .retrieve()
            .bodyToFlux(JsonNode.class);
}

上面的方法是这样调用的

public List<CustomModel> fetch(List<String> urls) {
    return Flux.fromIterable(urls)
            .parallel()
            .runOn(Schedulers.boundedElastic())
            .flatMap(this::performGetRequest)
            .flatMap(jsonNode -> Flux.fromIterable(customDeserialize(jsonNode)))
            .sequential()
            .collectList()
            .flatMapMany(Flux::fromIterable)
            .collectList()
            .block();
}

对于每个响应，我都使用一个自定义方法来反序列化响应

private List<CustomModel> customDeserialize(final JsonNode jsonNodeResponse) {
    List<CustomModel> customModelList = new ArrayList<>();
    
    for (JsonNode block : jsonNodeResponse) {
        // deserialize the response, create an instance of CustomModel class 
        // and add it to customModelList
    }
    
    return customModelList;
}

问题是，即使我使用了< code>parallel()方法，整个过程也可能不是并行运行的。完成的时间说明我做错了。

我错过了什么吗？

Answer 1

我不能100%确定这是否是这里的问题，但我注意到在使用WebClient和ParallelFlux时，WebClient仅返回响应的Publisher（体ToMono/体ToFlux），而不是实际请求。

考虑使用Fluc. defer/Mono.defer包装远程调用，以获取已经用于请求的Publisher，例如：

.flatMap(url -> Flux.defer(() -> performGetRequest(url)))

Answer 2

问题是，即使我使用了parallel()方法，整个过程也可能不是并行运行的。完成的时间说明我做错了。

我错过了什么吗？

由于您正在调用＜code＞block＜/code＞，我将假设您正在运行一个MVC servlet应用程序，该应用程序仅将＜code＞WebClient＜/code〕用于rest调用。

如果您没有运行完整的webflux应用程序，那么您的应用程序将启动一个单独的事件循环，该循环将处理计划的所有事件。如果运行一个完整的webflux应用程序，您将获得与运行机器上的内核一样多的事件循环。

通过使用< code>parallel,反应器文件说明:

要获得ParallelFlux，您可以在任何Flux上使用并行（）运算符。就其本身而言，此方法不会并行化工作。相反，它将工作负载划分为“rails”（默认情况下，与CPU内核一样多的rails）。

为了告诉产生的ParallelFlux在哪里运行每个rails(并通过扩展来并行运行rails ),您必须使用runOn(Scheduler)。注意，对于并行工作，有一个推荐的专用调度器:Schedulers.parallel()。

您正在创建一个有界弹性调度程序，该调度程序未针对并行工作进行优化。

但我想提一下，您正在做async i/o而不是并行工作，这是非常重要的一点。当您并行运行时，您很可能不会获得任何性能提升，因为您的大部分i/o都会触发请求，然后只是等待响应。

＜code＞ParellelFlux＜/code＞将确保所有的cpu内核都被使用，但也有一些惩罚。有一个设置时间来确保所有核心都能开始工作，然后需要完成的工作不是cpu密集型的，他们只会发出1000个请求，然后所有线程都完成了，必须等待响应。

需要在内核上设置工作线程，需要将信息发送到每个内核，检索等。

当您有CPU密集型工作时，并行获得了大部分好处，其中每个事件都需要在多个内核上执行繁重的计算。但是对于async工作，常规的Flux很可能就足够了。

以下是Reactor开发人员之一Simon Baslé关于在Reactor中运行I / O工作，并行与异步

另外值得一提的是，有界弹性调度器被调整为阻塞工作，作为纯 webflux 应用程序中常规 servlet 行为的回退。

您正在servlet应用程序中运行webflow，因此您获得的好处可能不如webflow应用程序那么全面。