无法在 CompletableFutures 中传播异常的自定义异常

Question

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(( 方法的异常日志记录，但是对于调用此方法的函数无法捕获从中抛出的异常。

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

Answer 1

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

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

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

最接近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。