completablefuture join()方法中的巨大延迟

1dkrff03  于 2021-07-12  发布在  Java
关注(0)|答案(3)|浏览(773)

因此,我正在开发一个应用程序,它必须一次进行20多个http调用。每一个都需要2-3秒才能得到响应。每次调用一个(最多40秒)相当慢,所以我尝试通过completablefutures异步发送它们。这样我就可以在等待别人回复的时候打电话,理论上可以把总时间从40秒减少到4-5秒。
我做了一个非常类似的设置,我发现在这个教程https://www.codepedia.org/ama/how-to-make-parallel-calls-in-java-with-completablefuture-example.

  1. import org.codingpedia.example;
  2. import javax.inject.Inject;
  3. import java.util.List;
  4. import java.util.concurrent.CompletableFuture;
  5. import java.util.function.Supplier;
  6. import java.util.stream.Collectors;
  7. public class ParallelCallsDemoService {
  8. @Inject
  9. RestApiClient restApiClient;
  10. private ExecutorService es = Executors.newFixedThreadPool(20);
  11. public List<ToDo> getToDos(List<String> ids){
  12. List<CompletableFuture<ToDo>> futures =
  13. ids.stream()
  14. .map(id -> getToDoAsync(id))
  15. .collect(Collectors.toList());
  16. List<ToDo> result =
  17. futures.stream()
  18. .map(CompletableFuture::join)
  19. .collect(Collectors.toList());
  20. return result;
  21. }
  22. CompletableFuture<ToDo> getToDoAsync(String id){
  23. CompletableFuture<ToDo> future = CompletableFuture.supplyAsync(() -> {
  24. return restApiClient.makeSomeHttpCall(id);
  25. }, es);
  26. return future;
  27. }
  28. }

据所有人说,这似乎是工作-所有的电话都在几乎相同的时间发送,他们都在几秒钟内返回。但是在这一部分我经历了30-40秒的巨大延迟:

  1. List<ToDo> result =
  2. futures.stream()
  3. .map(CompletableFuture::join)
  4. .collect(Collectors.toList());

这使得它所需的时间与连续发送大致相同,这让我感到困惑。怎么可能我在几秒钟内就收到了所有的回复,但是加入他们却延迟了30秒?这几乎就好像(尽管表面上)它们仍然是连续制作的。为什么加入要花这么长时间?

ljo96ir5

ljo96ir51#

这里有点问题

  1. List<ToDo> result =
  2. futures.stream()
  3. .map(CompletableFuture::join)
  4. .collect(Collectors.toList());

我认为您使用的流不是平行流。因此,对map的每个调用都在等待最后一个调用完成。改变 futures.stream()futures.parallelStream() 应该有所改善。当然,如果你不使用单核的机器。

nwo49xxi

nwo49xxi2#

终于弄明白了!谢谢大家的建议。原来这跟我实现completablefutures没什么关系。当我从服务接收到响应时,我使用jaxb将java对象转换为xml字符串,以便进行日志记录。我开始研究挂起时的线程转储,并意识到实际上是线程等待的jaxb字符串转换(响应对象相当大)。我去掉了那个部分,表演立刻提高到了应有的水平。

crcmnpdw

crcmnpdw3#

我们也遇到了类似的问题。我们使用completablefuture的“.get(timeout)”方法解决了这个问题。

  1. CompletableFuture[] array = (CompletableFuture[]) futures.toArray();
  2. try {
  3. CompletableFuture.allOf(array).get(180, TimeUnit.SECONDS);
  4. } catch (InterruptedException | ExecutionException | TimeoutException e) {
  5. //log error
  6. }

根据实时结果设置超时时间。可以使用外部配置调整时间。

相关问题