Java 8 Completable Futures 所有不同的数据类型

Posted 2023-02-16

【中文标题】Java 8 Completable Futures 所有不同的数据类型

【英文标题】：Java 8 Completable Futures allOf different data types

【发布时间】：2017-07-18 06:53:57

【问题描述】：

我有 3 个 CompletableFuture，所有 3 个都返回不同的数据类型。

我希望创建一个结果对象，它是所有 3 个期货返回的结果的组合。

所以我当前的工作代码如下所示：

public ClassD getResultClassD() 

    ClassD resultClass = new ClassD();
    CompletableFuture<ClassA> classAFuture = CompletableFuture.supplyAsync(() -> service.getClassA() );
    CompletableFuture<ClassB> classBFuture = CompletableFuture.supplyAsync(() -> service.getClassB() );
    CompletableFuture<ClassC> classCFuture = CompletableFuture.supplyAsync(() -> service.getClassC() );

    CompletableFuture.allOf(classAFuture, classBFuture, classCFuture)
                     .thenAcceptAsync(it -> 
                        ClassA classA = classAFuture.join();
                        if (classA != null) 
                            resultClass.setClassA(classA);
                        

                        ClassB classB = classBFuture.join();
                        if (classB != null) 
                            resultClass.setClassB(classB);
                        

                        ClassC classC = classCFuture.join();
                        if (classC != null) 
                            resultClass.setClassC(classC);
                        

                     );

    return resultClass;

我的问题是：

我的假设是，因为我使用的是allOf 和thenAcceptAsync，所以这个调用将是非阻塞的。我的理解对吗？

这是处理返回不同结果类型的多个期货的正确方法吗？

在thenAcceptAsync内构造ClassD对象对吗？

在 thenAcceptAsync lambda 中使用 join 或 getNow 方法是否合适？

【参考方案1】：

您的尝试正朝着正确的方向前进，但并不正确。您的方法getResultClassD() 返回一个已经实例化的ClassD 类型的对象，任意线程将在该对象上调用修改方法，而getResultClassD() 的调用者不会注意到。这可能会导致竞争条件，如果修改方法本身不是线程安全的，那么调用者将永远不会知道 ClassD 实例何时真正可以使用。

正确的解决方案是：

public CompletableFuture<ClassD> getResultClassD() 

    CompletableFuture<ClassA> classAFuture
        = CompletableFuture.supplyAsync(() -> service.getClassA() );
    CompletableFuture<ClassB> classBFuture
        = CompletableFuture.supplyAsync(() -> service.getClassB() );
    CompletableFuture<ClassC> classCFuture
        = CompletableFuture.supplyAsync(() -> service.getClassC() );

    return CompletableFuture.allOf(classAFuture, classBFuture, classCFuture)
         .thenApplyAsync(dummy -> 
            ClassD resultClass = new ClassD();

            ClassA classA = classAFuture.join();
            if (classA != null) 
                resultClass.setClassA(classA);
            

            ClassB classB = classBFuture.join();
            if (classB != null) 
                resultClass.setClassB(classB);
            

            ClassC classC = classCFuture.join();
            if (classC != null) 
                resultClass.setClassC(classC);
            

            return resultClass;
         );

现在，getResultClassD() 的调用者可以在操作完成后使用返回的CompletableFuture 查询进度状态或链相关操作，或者使用join() 检索结果。

要解决其他问题，是的，此操作是异步的，在 lambda 表达式中使用 join() 是合适的。 join 的创建正是因为 Future.get() 被声明为抛出检查异常，这使得在这些 lambda 表达式中的使用变得不必要地困难。

请注意，null 测试仅在这些 service.getClassX() 可以实际返回 null 时才有用。如果其中一个服务调用因异常而失败，则整个操作（由CompletableFuture<ClassD> 表示）将异常完成。

【讨论】：

感谢您的详细回复。我对您的回答的唯一跟进是 thenApplyAsync 的返回类型为 CompletableFuture，这将如何在这里工作以及如何调用此方法并使用结果

不，这是allOf的返回类型，即CompletableFuture<Void>，这就是为什么传递thenApplyAsync的函数接收Void作为输入（上面的dummy参数，而不是@987654341 @，你也可以写成(Void dummy) ->)。然后，该函数将Void 输入（实际上忽略它）转换为ClassD 结果，因此thenApplyAsync 的结果将是CompletableFuture<ClassD>。

@Holger 我走的路线与您类似，但我在服务调用中使用了 Optional.ofNullable，因此您可以使用 cCFuture.join().ifPresent(class::SetStuff)

@Ash：几乎总是有不止一种方法可以做到……

@Holger 感谢您对虚拟参数的解释。有道理

【参考方案2】：

我的做法与@Holger 在他的回答中所做的类似，但将服务调用包装在 Optional 中，这会导致 thenApplyAsync 阶段中的代码更简洁

CompletableFuture<Optional<ClassA>> classAFuture
    = CompletableFuture.supplyAsync(() -> Optional.ofNullable(service.getClassA())));

CompletableFuture<Optional<ClassB>> classBFuture
    = CompletableFuture.supplyAsync(() -> Optional.ofNullable(service.getClassB()));

CompletableFuture<Optional<ClassC>> classCFuture
    = CompletableFuture.supplyAsync(() -> Optional.ofNullable(service.getClassC()));

return CompletableFuture.allOf(classAFuture, classBFuture, classCFuture)
     .thenApplyAsync(dummy -> 
        ClassD resultClass = new ClassD();

        classAFuture.join().ifPresent(resultClass::setClassA)
        classBFuture.join().ifPresent(resultClass::setClassB)
        classCFuture.join().ifPresent(resultClass::setClassC)

        return resultClass;
     );

【讨论】：

是的，使用可选项是正确的思考方式

【参考方案3】：

我之前遇到过类似的事情，并创建了一个简短的演示来展示我是如何解决这个问题的。

与@Holger 类似的概念，只是我使用了一个函数来组合每个单独的未来。

https://github.com/te21wals/CompletableFuturesDemo

基本上：

    public class CombindFunctionImpl implement CombindFunction 
    public ABCData combind (ClassA a, ClassB b, ClassC c) 
        return new ABCData(a, b, c);
   

...

    public class FutureProvider 
public CompletableFuture<ClassA> retrieveClassA() 
    return CompletableFuture.supplyAsync(() -> 
        try 
            Thread.sleep(1000L);
         catch (InterruptedException e) 
            e.printStackTrace();
        
        return new ClassA();
    );


public CompletableFuture<ClassB> retrieveClassB() 
    return CompletableFuture.supplyAsync(() -> 
        try 
            Thread.sleep(2000L);
         catch (InterruptedException e) 
            e.printStackTrace();
        
        return new ClassB();
    );

public CompletableFuture<ClassC> retrieveClassC() 
    return CompletableFuture.supplyAsync(() -> 
        try 
            Thread.sleep(3000L);
         catch (InterruptedException e) 
            e.printStackTrace();
        
        return new ClassC();
    );

......

public static void main (String[] args)
    CompletableFuture<ClassA> classAfuture = futureProvider.retrieveClassA();
    CompletableFuture<ClassB> classBfuture = futureProvider.retrieveClassB();
    CompletableFuture<ClassC> classCfuture = futureProvider.retrieveClassC();

    System.out.println("starting completable futures ...");
    long startTime = System.nanoTime();

    ABCData ABCData = CompletableFuture.allOf(classAfuture, classBfuture, classCfuture)
            .thenApplyAsync(ignored ->
                    combineFunction.combind(
                            classAfuture.join(),
                            classBfuture.join(),
                            classCfuture.join())
            ).join();

    long endTime = System.nanoTime();
    long duration = (endTime - startTime);
    System.out.println("completable futures are complete...");
    System.out.println("duration:\t" + Duration.ofNanos(duration).toString());
    System.out.println("result:\t" + ABCData);

【参考方案4】：

如果您不想声明尽可能多的变量，另一种处理方法是使用 thenCombine 或 thenCombineAsync 将您的未来链接在一起。

public CompletableFuture<ClassD> getResultClassD()

  return CompletableFuture.supplyAsync(ClassD::new)
    .thenCombine(CompletableFuture.supplyAsync(service::getClassA), (d, a) -> 
      d.setClassA(a);
      return d;
    )
    .thenCombine(CompletableFuture.supplyAsync(service::getClassB), (d, b) -> 
      d.setClassB(b);
      return d;
    )
    .thenCombine(CompletableFuture.supplyAsync(service::getClassC), (d, c) -> 
      d.setClassC(c);
      return d;
    );

getter 仍然会被异步触发并按顺序执行结果。这基本上是获得相同结果的另一种语法选项。

【讨论】：

如果只有 2 或 3 个期货要组合，这可能是一个好方法。但是，您可能应该从 CompletableFuture.completedFuture(new ClassD()) 开始，因为实例化可能不值得异步运行。事实上，您甚至可以在第一个 future 上将其实例化为 thenApply()。