Java 8可完全并行执行方法

Question

我有3种方法需要并行运行，因为它们彼此独立，并在最后合并每种方法的结果，并将其作为响应发送。我还需要处理异常。

在不同的帖子中，我找到了下面的代码并进行了相应的修改。

public Response getResponse() {
    Response resultClass = new Response();
   try {
    CompletableFuture<Optional<ClassA>> classAFuture
        = CompletableFuture.supplyAsync(() -> service.getClassA() );
    CompletableFuture<ClassB> classBFuture
        = CompletableFuture.supplyAsync(() -> {
             try {
                   return service.getClassB(); 
              }
              catch (Exception e) {
                   throw new CompletionException(e);
              }
     });
    CompletableFuture<ClassC> classCFuture
        = CompletableFuture.supplyAsync(() -> { 
            try {
                return service.getClassC();
            } catch (Exception e) {
                throw new CompletionException(e);
            }
    });

   CompletableFuture<Response> responseFuture =
    CompletableFuture.allOf(classAFuture, classBFuture, classCFuture)
         .thenApplyAsync(dummy -> {
            if (classAFuture.join().isPresent() {
               ClassA classA = classAFuture.join();
               classA.setClassB(classBFuture.join());
               classA.setClassC(classCFuture.join());
               response.setClassA(classA)
             }
            return response;
         });
   responseFuture.join();
  } catch (CompletionExecution e) {
    throw e;
  }
  return response;
}

上述各项是否应正确并行运行？我知道这需要更多的时间，我想确保我做对了。

Answer 1

这个想法是正确的，但这一切都可以用更少的代码来完成：

  public Response getResponse() {
    CompletableFuture<Optional<ClassA>> classAFuture = CompletableFuture.supplyAsync(() -> service.getClassA());
    CompletableFuture<ClassB> classBFuture = CompletableFuture.supplyAsync(() -> service.getClassB());
    CompletableFuture<ClassC> classCFuture = CompletableFuture.supplyAsync(() -> service.getClassC());
    
    try {
      return CompletableFuture.allOf(classAFuture, classBFuture, classCFuture)
        .thenApply(() -> {
          Response response = new Response();
          Optional<ClassA> maybeA = classAFuture.get();
          if (maybeA.isPresent()) {
            ClassA classA = maybeA.get();
            classA.setClassB(classBFuture.get());
            classA.setClassC(classCFuture.get());
            response.setClassA(classA);
          }
          return response;
        }).get();
    } catch (ExecutionException e) { // Ususally the exception is wrapped to ExecutionException by java concurrency framework itself
        Throwable cause = e.getCause();
        if (cause != null) {
          throw cause;
        } else {
          throw e;
        }
    }
  }

主要事情：

1. 您不需要将异常包装到CompletionException。
2. 您不需要使用thenApplyAsync。只需thenApply是一回事，除非您想非常具体地说明要使用的线程类型。检查此项以获取更多信息https://stackoverflow.com/a/47489654/3020903
3. 您不需要connect（）任何东西。当CompletableFuture.all完成时，您可以非常确定所有提供的作业都已完成，然后调用get（）将只返回值。

至于你能保证作业A、B和C将并行运行吗。是和否。如果有足够的系统资源并行运行，它将并行运行。你已经尽力要求他们并行运行。也许在某个时候，您还希望提供自定义线程池以获得更多控制，但这是另一天的主题。

Answer 2

如果您想并行运行方法，您应该使用ExecutorService。试试这样的方法：

ExecutorService myExecutor = Executors.newFixedThreadPool(3);
        List<Future<Object>> futures = myExecutor.invokeAll(
            Arrays.asList(
                () -> service.getClassA(),
                () -> service.getClassB(),
                () -> service.getClassC(),
            )
        );
        myExecutor.shutdown();