并发调用多个Spring微服务URL

如何解决并发调用多个Spring微服务URL

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

我需要我的应用程序捕获哪些网址不返回 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 中收集的任何错误网址。

更新

我修改了上面的帖子，声明我只想同时进行调用。我不在乎 rest.getForEntity 调用块。

解决方法

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

// 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
    }

,

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

1. 使用 Mono#fromCallable 包装阻塞服务调用
2. 使用 serviceUrls.getServiceUrls() 将 Flux#fromIterable 转换为响应式流
3. 使用来自 2. 的 Flux 和来自 1 的异步服务调用通过 Flux#filterWhen 同时调用和过滤失败的服务。
4. 使用 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 时应避免阻塞调用。正确的做法是用完全非阻塞的 Spring 5 中的 RestTemplate 替换 WebClient。

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