并发调用多个 Spring 微服务 URL答案

【问题标题】：Concurrently call several Spring microservice URLs并发调用多个 Spring 微服务 URL
【发布时间】：2021-03-29 19:12:42
【问题描述】：

我有一个 Spring Boot 应用程序，它将使用 GET 方法调用多个微服务 URL。这些微服务 URL 端点都实现为@RestControllers。他们不会返回 Flux 或 Mono。

我需要我的应用程序来捕获哪些 URL没有返回 2xx HTTP 状态。

我目前正在使用以下代码来执行此操作：

List<String> failedServiceUrls = new ArrayList<>();
        for (String serviceUrl : serviceUrls.getServiceUrls()) {
            try {

                
                ResponseEntity<String> response = rest.getForEntity(serviceUrl, String.class);
                
                if (!response.getStatusCode().is2xxSuccessful()) {
                    failedServiceUrls.add(serviceUrl);
                }

            } catch (Exception e){
                failedServiceUrls.add(serviceUrl);
            }
            
        }

        // all checks are complete so send email with the failedServiceUrls.
        mail.sendEmail("Service Check Complete", failedServiceUrls);
    }

问题是每个 URL 调用响应缓慢，我必须等待一个 URL 调用完成才能进行下一个调用。

如何更改此设置以同时进行 URL 调用？在所有通话完成后，我需要发送一封电子邮件，其中包含应收集到 failedServiceUrls 中的任何错误 URL。

更新

我修改了上面的帖子，声明我只想同时拨打电话。我不在乎 rest.getForEntity 调用阻塞。

【问题讨论】：

据我所知，您可以创建几个 Completable Futures，然后您可以随时等待它们完成工作
你需要非阻塞实现，还是需要并行执行？
你想要它们同时和非阻塞还是两者兼而有之？因为 CompletableFuture 将提供并发，但不提供 nio。
答案可能取决于您对并发和 NIO 问题的很多澄清。一个简单的解决方案可能是将 NIO 的请求工厂传递给您的 RestTemplate（例如 Netty4ClientHttpRequestFactory）是 NIO。但是根据您的意思，还有很多其他选择。
我为歧义道歉。我已经更新了我的问题。

标签： java spring multithreading spring-boot spring-webflux

【解决方案1】：

在您的代码中使用执行器服务，您可以通过这种方式并行调用所有微服务：

// synchronised it as per Maciej's comment:
failedServiceUrls = Collections.synchronizedList(failedServiceUrls);
ExecutorService executorService = Executors.newFixedThreadPool(serviceUrls.getServiceUrls().size());

    List<Callable<String>> runnables = new ArrayList<>().stream().map(o -> new Callable<String>() {
      @Override
      public String call() throws Exception {
        ResponseEntity<String> response = rest.getForEntity(serviceUrl, String.class);
        // do something with the response

        if (!response.getStatusCode().is2xxSuccessful()) {
          failedServiceUrls.add(serviceUrl);
        }

        return response.getBody();
      }
    }).collect(toList());

    List<Future<String>> result = executorService.invokeAll(runnables);
    for(Future f : result) {
      String resultFromService = f.get(); // blocker, it will wait until the execution is over
    }

【讨论】：

你正在同时从多个线程修改failedServiceUrls。为避免竞争条件，必须同步访问。

【解决方案2】：

如果您只想并发调用并且不关心阻塞线程，您可以：

使用Mono#fromCallable包装阻塞服务调用
使用Flux#fromIterable 将serviceUrls.getServiceUrls() 转换为反应流
使用来自 2. 的 Flux 和来自 1 的异步服务调用同时调用和过滤带有 Flux#filterWhen 的失败服务。
使用Flux#collectList 等待所有调用完成，然后使用subscribe 中的无效网址发送电子邮件

void sendFailedUrls() {
        Flux.fromIterable(erviceUrls.getServiceUrls())
                .filterWhen(url -> responseFailed(url))
                .collectList()
                .subscribe(failedURls -> mail.sendEmail("Service Check Complete", failedURls));
    }

    Mono<Boolean> responseFailed(String url) {
        return Mono.fromCallable(() -> rest.getForEntity(url, String.class))
                .map(response -> !response.getStatusCode().is2xxSuccessful())
.subscribeOn(Schedulers.boundedElastic());

    }

使用 Reactor 阻止调用

由于底层服务调用被阻塞，它应该在专用线程池上执行。如果要实现完全并发，这个线程池的大小应该等于并发调用的数量。这就是为什么我们需要.subscribeOn(Schedulers.boundedElastic())

见：https://projectreactor.io/docs/core/release/reference/#faq.wrap-blocking

使用 WebClient 的更好解决方案

但是请注意，在使用 reactor 和 spring webflux 时应避免阻塞调用。正确的做法是将 RestTemplate 替换为 Spring 5 中完全非阻塞的 WebClient。

见：https://docs.spring.io/spring-boot/docs/2.0.3.RELEASE/reference/html/boot-features-webclient.html

【讨论】：