java - 如何处理多个 ListenableFuture？ ( Spring )

Question

我正在编写一个 Controller ，我需要使其异步。我如何处理 ListenableFuture 的列表？因为我有一个 URL 列表，需要一一发送 GET 请求，那么最好的解决方案是什么？

@RequestMapping(value = "/repositories", method = RequestMethod.GET)
    private void getUsername(@RequestParam(value = "username") String username) {
        System.out.println(username);
        List<ListenableFuture> futureList = githubRestAsync.getRepositoryLanguages(username);
        System.out.println(futureList.size());
}

在我使用的服务中List<ListanbleFuture>这似乎不起作用，因为它是异步的，在 Controller 方法中我不能有 futureList 的大小运行 for loop在其上进行回调。

public List<ListenableFuture> getRepositoryLanguages(String username){
      return getRepositoryLanguages(username, getUserRepositoriesFuture(username));
    }

private ListenableFuture getUserRepositoriesFuture(String username) throws HttpClientErrorException {
        HttpEntity entity = new HttpEntity(httpHeaders);
        ListenableFuture future = restTemplate.exchange(githubUsersUrl + username + "/repos", HttpMethod.GET, entity, String.class);
        return future;
    }
private List<ListenableFuture> getRepositoryLanguages(final String username, ListenableFuture<ResponseEntity<String>> future) {
        final List<ListenableFuture> futures = new ArrayList<>();
        future.addCallback(new ListenableFutureCallback<ResponseEntity<String>>() {
            @Override
            public void onSuccess(ResponseEntity<String> response) {
                ObjectMapper mapper = new ObjectMapper();
                try {
                    repositories = mapper.readValue(response.getBody(), new TypeReference<List<Repositories>>() {
                    });
                    HttpEntity entity = new HttpEntity(httpHeaders);
                    System.out.println("Repo size: " + repositories.size());
                    for (int i = 0; i < repositories.size(); i++) {
                        futures.add(restTemplate.exchange(githubReposUrl + username + "/" + repositories.get(i).getName() + "/languages", HttpMethod.GET, entity, String.class));
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("FAILURE in getRepositoryLanguages: " + throwable.getMessage());
            }
        });

        return futures;
    }

我应该使用类似 ListenableFuture<List> 的东西吗？而不是List<ListenableFuture> ？

Answer 1

您似乎有一个 List<ListenableFuture<Result>> ，但你想要 ListenableFuture<List<Result>> ，这样您就可以在所有 future 完成时采取一项操作。

public static <T> ListenableFuture<List<T>> allOf(final List<? extends ListenableFuture<? extends T>> futures) {
    // we will return this ListenableFuture, and modify it from within callbacks on each input future
    final SettableListenableFuture<List<T>> groupFuture = new SettableListenableFuture<>();

    // use a defensive shallow copy of the futures list, to avoid errors that could be caused by
    // someone inserting/removing a future from `futures` list after they call this method
    final List<? extends ListenableFuture<? extends T>> futuresCopy = new ArrayList<>(futures);

    // Count the number of completed futures with an AtomicInt (to avoid race conditions)
    final AtomicInteger resultCount = new AtomicInteger(0);
    for (int i = 0; i < futuresCopy.size(); i++) {
        futuresCopy.get(i).addCallback(new ListenableFutureCallback<T>() {
            @Override
            public void onSuccess(final T result) {
                int thisCount = resultCount.incrementAndGet();

                // if this is the last result, build the ArrayList and complete the GroupFuture
                if (thisCount == futuresCopy.size()) {
                   List<T> resultList = new ArrayList<T>(futuresCopy.size());
                    try {
                        for (ListenableFuture<? extends T> future : futuresCopy) {
                            resultList.add(future.get());
                        }
                        groupFuture.set(resultList);
                    } catch (Exception e) {
                        // this should never happen, but future.get() forces us to deal with this exception.
                        groupFuture.setException(e);
                    }
                }
            }

            @Override
            public void onFailure(final Throwable throwable) {
                groupFuture.setException(throwable);

                // if one future fails, don't waste effort on the others
                for (ListenableFuture future : futuresCopy) {
                    future.cancel(true);
                }
            }
        });
    }

    return groupFuture;
}