java 使用 AsyncRestTemplate 多次制作 API 并等待所有完成

Make API multiple times with AsyncRestTemplate and wait for all to complete

提问by TV Nath

I have to make Rest API invocation using RestTemplate multiple time with different parameters. API is same but it is the parameter that is getting changed. Number of times is also variable. I want to use AsyncRestTemplate but my main Thread should wait until all API calls have been successfully completed. I also want to work with responses that each API call returned. Currently I am using RestTemplate. In basic form it is as following.

我必须使用 RestTemplate 多次使用不同的参数进行 Rest API 调用。API 是相同的，但参数正在发生变化。次数也是可变的。我想使用 AsyncRestTemplate 但我的主线程应该等到所有 API 调用都成功完成。我还想处理每个 API 调用返回的响应。目前我正在使用 RestTemplate。基本形式如下。

List<String> listOfResponses = new ArrayList<String>();
for (Integer studentId : studentIdsList) {
    String respBody;
    try {
        ResponseEntity<String> responseEntity = restTemplate.exchange(url, method, requestEntity, String.class);
    } catch (Exception ex) {
        throw new ApplicationException("Exception while making Rest call.", ex);
    }
    respBody = requestEntity.getBody();
    listOfResponses.add(respBody);          
}

How can I implement AsyncRestTemplate in this situation?

在这种情况下如何实现 AsyncRestTemplate？

回答by Didier L

The main idea when using AsyncRestTemplate(or any asynchronous API, in fact), is to send all you requests in a first time, keeping the corresponding futures, then process all responses in a second time. You can simply do this with 2 loops:

使用AsyncRestTemplate（或任何异步 API，实际上）时的主要思想是在第一次发送所有请求，保留相应的期货，然后在第二次处理所有响应。你可以简单地用 2 个循环来做到这一点：

List<ListenableFuture<ResponseEntity<String>>> responseFutures = new ArrayList<>();
for (Integer studentId : studentIdsList) {
    // FIXME studentId is not used
    ListenableFuture<ResponseEntity<String>> responseEntityFuture = restTemplate.exchange(url, method, requestEntity, String.class);
    responseFutures.add(responseEntityFuture);
}
// now all requests were send, so we can process the responses
List<String> listOfResponses = new ArrayList<>();
for (ListenableFuture<ResponseEntity<String>> future: responseFutures) {
    try {
        String respBody = future.get().getBody();
        listOfResponses.add(respBody);
    } catch (Exception ex) {
        throw new ApplicationException("Exception while making Rest call.", ex);
    }
}

Note: if you need to pair the responses with the original requests, you can replace the list of futures with a map or a list of request+response objects.

注意：如果您需要将响应与原始请求配对，您可以将期货列表替换为地图或请求+响应对象列表。

I also noted that studentIdis not used in your question.

我还注意到studentId您的问题中没有使用。

回答by Danylo Zatorsky

You could use Java 8 Stream API, if that's feasible for you:

如果对您来说可行，您可以使用 Java 8 Stream API：

List<String> listOfResponses = studentIdsList.stream()
    .parrallel()
    .map({studentId ->
        ResponseEntity<String> responseEntity = restTemplate.exchange(url, method, studentId, String.class);
        return responseEntity.getBody();
    })
    .collect(Collectors.toList());

This code will basically perform 2 things:

这段代码基本上会执行两件事：

1. Performs requests in parallel;
2. Collect the results of the requests into a List.

1. 并行执行请求；
2. 将请求的结果收集到一个列表中。

UPDATE: Agree with @Didier L - this solution may not work properly when you need to do a lot of requests. Here is an updated version:

更新：同意@Didier L - 当您需要执行大量请求时，此解决方案可能无法正常工作。这是一个更新的版本：

List<String> listOfResponses  = studentIdsList.stream()
                .map(studentId -> asyncRestTemplate.exchange(url, method, studentId, String.class)
                .collect(Collectors.toList()).stream()
                .map(this::retrieveResult)
                .collect(Collectors.toList());

    /**
     * Retrieves results of each request by blocking the main thread. Note that the actual request was performed on the previous step when
     * calling asyncRestTemplate.exchange(url, method, studentId, String.class)
     */
    private String retrieveResult(ListenableFuture<ResponseEntity<String>> listenableFuture) {
        try {
            return listenableFuture.get().getBody();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

回答by Sneh

Here is another solution I would like to suggest which uses Spring's RestTemplate rather than AsyncRestTemplate. It is also using Java 8 CompletableFuture.

这是我想建议的另一个解决方案，它使用 Spring 的 RestTemplate 而不是 AsyncRestTemplate。它还使用 Java 8 CompletableFuture。

public void sendRequestsAsync(List<Integer> studentList) {
    List<CompletableFuture<Void>> completableFutures = new ArrayList<>(studentList.size()); //List to hold all the completable futures
    List<String> responses = new ArrayList<>(); //List for responses
    ExecutorService yourOwnExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    for (Integer studentId : studentList) { //Iterate student list
        CompletableFuture<Void> requestCompletableFuture = CompletableFuture
                .supplyAsync(
                        () -> restTemplate.exchange("URL/" + studentId, HttpMethod.GET, null, String.class),
                        yourOwnExecutor
                )//Supply the task you wanna run, in your case http request
                .thenApply((responseEntity) -> {
                    responses.add(responseEntity.getBody());
                    return responseEntity;
                })//now you can add response body to responses
                .thenAccept((responseEntity) -> {
                    doSomeFinalStuffWithResponse(responseEntity);
                })//here you can do more stuff with responseEntity (if you need to)
                .exceptionally(ex -> {
                    System.out.println(ex);
                    return null;
                });//do something here if an exception occurs in the execution;

        completableFutures.add(requestCompletableFuture);
    }

    try {
        CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()])).get(); //Now block till all of them are executed by building another completablefuture with others.
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (ExecutionException e) {
        e.printStackTrace();
    }
}

I like this solution more because I can chain as much business logic I want and don't have to depend on Spring's internals for Async Sending. Obviously you can clean up the code more, I haven't paid much attention into that for now.

我更喜欢这个解决方案，因为我可以链接尽可能多的业务逻辑，而且不必依赖 Spring 的内部来进行异步发送。显然你可以更多地清理代码，我现在没有太多关注。