使用 CompletableFuture 处理 Java 8 供应商异常

Question

考虑以下代码

public class TestCompletableFuture {

    BiConsumer<Integer, Throwable> biConsumer = (x,y) -> {
        System.out.println(x);
        System.out.println(y);
    };

    public static void main(String args[]) {
        TestCompletableFuture testF = new TestCompletableFuture();
        testF.start();      
    }

    public void start() {
        Supplier<Integer> numberSupplier = new Supplier<Integer>() {
            @Override
            public Integer get() {
                return SupplyNumbers.sendNumbers();                     
            }
        };
        CompletableFuture<Integer> testFuture = CompletableFuture.supplyAsync(numberSupplier).whenComplete(biConsumer);         
    }       
}

class SupplyNumbers {
    public static Integer sendNumbers(){
        return 25; // just for working sake its not  correct.
    }
}

上面的东西工作正常。但是 sendNumbers 在我的例子中也可能抛出一个检查异常，比如:

class SupplyNumbers {
    public static Integer sendNumbers() throws Exception {
        return 25; // just for working sake its not  correct.
    }
}

现在我想在我的 biConsumer 中将此异常处理为 y。这将帮助我在单个函数 (biConsumer) 中处理结果和异常(如果有的话)。

有什么想法吗？我可以在这里使用 CompletableFuture.exceptionally(fn) 或其他任何东西吗？

Answer 1

当您想要处理已检查的异常时，使用标准功能接口(interface)的工厂方法没有帮助。当您将捕获异常的代码插入 lambda 表达式时，您会遇到问题，即 catch 子句需要 CompletableFuture 实例来设置异常，而工厂方法需要 Supplier，鸡和蛋。

您可以使用类的实例字段来允许在创建后进行修改，但最终生成的代码并不干净，而且比直接基于 Executor 的解决方案更复杂。 documentation of CompletableFuture说:

• All async methods without an explicit Executor argument are performed using the ForkJoinPool.commonPool() …

所以您知道以下代码将显示 CompletableFuture.supplyAsync(Supplier) 的标准行为，同时直接处理已检查的异常:

CompletableFuture<Integer> f=new CompletableFuture<>();
ForkJoinPool.commonPool().submit(()-> {
  try { f.complete(SupplyNumbers.sendNumbers()); }
  catch(Exception ex) { f.completeExceptionally(ex); }
});

文档还说:

… To simplify monitoring, debugging, and tracking, all generated asynchronous tasks are instances of the marker interface CompletableFuture.AsynchronousCompletionTask.

如果您想遵守此约定以使解决方案的行为更像原始的 supplyAsync 方法，请将代码更改为:

CompletableFuture<Integer> f=new CompletableFuture<>();
ForkJoinPool.commonPool().submit(
  (Runnable&CompletableFuture.AsynchronousCompletionTask)()-> {
    try { f.complete(SupplyNumbers.sendNumbers()); }
    catch(Exception ex) { f.completeExceptionally(ex); }
});