无法从 CompletableFuture 中的异常传播自定义异常答案

【问题标题】：Unable to propagate custom exception from exceptionally in CompletableFuture无法从 CompletableFuture 中的异常传播自定义异常
【发布时间】：2020-01-17 16:52:56
【问题描述】：

           public void executeAsync(Task task){
             CompletableFuture.runAsync(
                () -> {
                    task.execute(CompletableFuture.completedFuture(null), executor)
                            .exceptionally(
                                    ex -> {
                                        log.error(ex.getMessage(), ex);
                                        log.debug(
                                                "exception from executor" + ex.getStackTrace());
                                        throw new RunTimeException(
                                                ex);
                                    })
                            .join();
                },
                executor);
     }


    public static Task execute(Task... tasks) {
    return (future, executor) ->
            tasks.length == 0
                    ? future
                    : CompletableFuture.allOf(
                            Arrays.stream(tasks)
                                    .map(t -> t.execute(future, executor))
                                    .toArray(CompletableFuture[]::new));
}

在上面的代码中，我可以看到其中一项任务的 executeAsync() 方法的异常记录，但是对于此方法的调用函数无法捕获从它抛出的异常。

任何指针都会在这方面有所帮助，似乎没有什么对我有用。我需要捕获任何任务引发的任何异常并相应地更新数据库。

【问题讨论】：

这应该如何工作？该任务是异步执行的，因此抛出的Exception不能绑定到外部catch。
请说明您想要达到的目标。
如果任何任务失败，我想在上层捕获异常，这是最好的方法（如果上层不能捕获它，那么最好的解决方案是什么找出是否发生任何故障？
@AlexeiKaigorodov ：已经解释了上面的场景，在这种情况下需要指针来处理错误

标签： java multithreading asynchronous threadpoolexecutor

【解决方案1】：

据我所知，唯一能够处理层级错误的异步库是 DF4J（由我开发）。一层定义为Dataflow对象：

 Dataflow upper = new Dataflow();  
 Dataflow nested = new Dataflow(upper);

Dataflow 是一个由异步过程和嵌套Dataflows 组成的图。当异步过程引发异常时，它会传播到父数据流。我们只能监控上层数据流的所有错误。

最接近CompletableFuture 的类是AsyncFunc。 AsyncFuncs 可能有异步参数来从其他AsyncFuncs 或CompletableFutures 获取值，但在本例中我们不使用它们。

对于每种 AsyncFunc，建议声明自己的类：

class StringToInt extends AsyncFunc<Integer> {
    String argumnet;

    // in constructor, link this async function to the parent dataflow
    public StringToInt(Dataflow df, String argumnet) {
        super(df);
        this.argumnet = argumnet;
    }

    @Override
    protected Integer callAction() throws Throwable {
        return Integer.valueOf(argumnet); // can throw NumberFormatException
    }
}

这个类可以如下使用：

    // start good asyncFunc tied to upper dataflow
    new StringToInt(upper, "10").start();
    // start bad asyncFunc tied to nested dataflow
    new StringToInt(nested, "not an integer").start();
    try {
        // here wait all started async functions to finish
        upper.blockingAwait(100);
        Assert.fail("exception expected");
    } catch (CompletionException e) {
       // here we catch first error from all async functions
        Throwable cause = e.getCause();
        cause.printStackTrace();
        Assert.assertEquals(NumberFormatException.class, cause.getClass());
    }

建议使用从 Github 克隆的最新版本的 DF4J。

【讨论】：