Поток объекта, отправить запрос с завершаемым будущим и назначить результат для объекта - PullRequest
Новая
       12

Поток объекта, отправить запрос с завершаемым будущим и назначить результат для объекта

1 голос
/ 01 апреля 2019

У меня есть список подписавшихся объектов.Для каждого из этих подписантов мне нужно сделать запрос ReST, чтобы получить их подписывающий URL.Я пытаюсь сделать это с помощью завершаемых вариантов будущего, чтобы все запросы ReST можно было отправлять параллельно, затем мне нужно установить этот URL-адрес для каждого из подписавших, чтобы эта операция не возвращала новых подписчиков, а только обновляла те, которые я уже выполняю.

У меня есть этот код, который уже работает, но я думаю, что он может быть улучшен. 

 List<Signer> signers=......

 List<CompletableFuture> futures = signers.stream()
          .map(signer -> CompletableFuture.completedFuture(signer))
          .map(future -> future.thenCombine(       CompletableFuture.supplyAsync(()-> signatureService.getSigningUrl(future.join().getSignerId())),
                           (signer, url) -> {
                               signer.setUrl(url);
                               return url;
                           }
                   )).collect(toList());

           futures.stream()
                   .map(CompletableFuture::join)
                    .collect(toList());

Могу ли я заменить этот 

 futures.stream()
   .map(CompletableFuture::join)
   .collect(toList());

этим? 

futures.stream().forEach(CompletableFuture::join)

Я не хотел бы возвращать это, потому что это уже использовалось, устанавливая это в подписывающем лице.и мне не нравится второй collect(toList()), потому что я не пытаюсь собрать что-то в это время.

Какую другую реализацию вы бы использовали?

1 Ответ

1 голос
/ 02 апреля 2019

Нет.futures.stream().forEach(CompletableFuture::join) возвращает void, тогда как futures.stream().map(CompletableFuture::join).collect(toList()); возвращает CompletableFuture<List<?>>.

Они оба предназначены для разных целей.Но и то, и другое делает одно общее (то есть блокирует основной поток до тех пор, пока все завершаемые фьючерсы не будут завершены).

Я бы написал ваш код по-разному, используя CompletableFuture.allOf. 

    Stream<CompletableFuture<String>> streamFutures = signers.stream()
            .map(signer -> CompletableFuture.completedFuture(signer))
            .map(future -> future.thenCombine(CompletableFuture.supplyAsync(() -> signatureService.getSigningUrl(future.join().getSignerId())),
                    (signer, url) -> {
                        signer.setUrl(url);
                        return url;
                    }
            ));

    CompletableFuture<String> [] futureArr = streamFutures.toArray(size -> new CompletableFuture[size]);
    List<CompletableFuture<String>> futures = Arrays.asList(futureArr);

    CompletableFuture<Void> allFuturesVoid = CompletableFuture.allOf(futureArr);
    allFuturesVoid.join();
    CompletableFuture<List<?>> allCompletableFuture = allFuturesVoid.thenApply(future -> futures.stream().map(completableFuture -> completableFuture.join()).collect(Collectors.toList()));

Тамхороший учебник здесь https://m -hewedy.blogspot.com / 2017/02 /comptablefutureallof-that-doenst.html и здесь https://medium.com/@senanayake.kalpa/fantastic-completablefuture-allof-and-how-to-handle-errors-27e8a97144a0.

...