CODEWITHSUNDEEP
javaspringmultithreadingspring-bootspring-webflux
user8297969
user8297969

Reputation: 455

Concurrently call several Spring microservice URLs

I have a Spring Boot application that will call several microservice URLs using the GET method. These microservice URL endpoints are all implemented as @RestControllers. They don't return Flux or Mono.

I need my application to capture which URLs are not returning 2xx HTTP status.

I'm currently using the following code to do this:

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);
    }

The problem is that each URL call is slow to respond and I have to wait for one URL call to complete prior to making the next one.

How can I change this to make the URLs calls be made concurrently? After all call have completed, I need to send an email with any URLs that have an error that should be collected in failedServiceUrls.

Update

I revised the above post to state that I just want the calls to be made concurrently. I don't care that rest.getForEntity call blocks.

Upvotes: 0

Views: 692

Answers (2)

Maciej Mościcki
Maciej Mościcki

Reputation: 766

If you just want to make calls concurrently and you don't care about blocking threads you can:

  1. wrap the blocking service call using Mono#fromCallable
  2. transform serviceUrls.getServiceUrls() into a reactive stream using Flux#fromIterable
  3. Concurrently call and filter failed services with Flux#filterWhen using Flux from 2. and asynchronous service call from 1.
  4. Wait for all calls to complete using Flux#collectList and send email with invalid urls in 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());

    }

Blocking calls with Reactor

Since the underlying service call is blocking it should be executed on a dedicated thread pool. Size of this thread pool should be equal to the number of concurrent calls if you want to achieve full concurrency. That's why we need .subscribeOn(Schedulers.boundedElastic())

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

Better solution using WebClient

Note however, that blocking calls should be avoided when using reactor and spring webflux. The correct way to do this would be to replace RestTemplate with WebClient from Spring 5 which is fully non-blocking.

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

Upvotes: 0

leozin
leozin

Reputation: 387

Using the executor service in your code, you can call all microservices in parallel this way:

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

Upvotes: 2

Related Questions