如何限制 Spring-Webflux 中的并行执行？答案

【问题标题】：How to limit parallel execution in Spring-Webflux?如何限制 Spring-Webflux 中的并行执行？
【发布时间】：2021-09-25 05:54:02
【问题描述】：

我的 rest 客户端实现是使用 Project Reactor 中的 Webflux，类似于以下内容：

Flux<Response1> request1(String uri) {
  return webClient
          .get()
          .uri(uri)
          .retrieve()
          .bodyToMono(Responses1.class)
          .map(r -> response1ToList(r))
          .flatMapMany(Flux::fromIterable);
}

Flux<Response2> request2(Response1 response1) {
  uri = f(response1);
  return webClient
          .get()
          .uri(uri)
          .retrieve
          .bodyToMono(Responses2.class)
          .map(r -> response1ToList(r))
          .flatMapMany(Flux::fromIterable);
}

Flux<Response2> res2 = res1.flatMap(request2).subscribe();

初始请求 (request1) 返回用于发送一系列请求 (request2) 的元素列表。

我的问题是request2s都是并行发送的，这对服务器来说太重了。

有没有办法限制同时执行的request2的数量？

【问题讨论】：

标签： java spring-webflux project-reactor

【解决方案1】：

您可以通过传入并发因子来控制flatMap() 订阅的内部流的数量。

Flux<Response1> request1(String uri) {
  return webClient
          .get()
          .uri(uri)
          .retrieve()
          .bodyToMono(Responses1.class)
          .map(r -> response1ToList(r))
          .flatMapMany(Flux::fromIterable);
}

Flux<Response2> request2(Response1 response1) {
  uri = f(response1);
  return webClient
          .get()
          .uri(uri)
          .retrieve
          .bodyToMono(Responses2.class)
          .map(r -> response1ToList(r))
          .flatMapMany(Flux::fromIterable);
}

Flux<Response2> res2 = res1.flatMap(request2, concurrencyFactor).subscribe();

如果你想一个一个地做，你可以使用concatMap()。

【讨论】：