如何处理多个ListenableFutures？(弹簧)

Question

我正在编写一个控制器,我需要让它异步.我该如何处理清单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>中似乎不起作用,因为它是异步的,在控制器方法中,我不能为回调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>>,所以你可以在所有的期货完成时采取一个行动.

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