java - 任务链中的 CompletableFuture 错误处理

Question

我完全迷失了如何使用 CompletableFutures 进行错误处理。我需要的是让多个任务异步运行。这些任务由多个步骤组成，如下例所示：

从数据库接收数据 -> 将此数据用于请求 -> 执行另一个请求 -> 更新数据库记录

现在每一步都可能导致异常，即未找到数据库记录或不正确的数据、请求失败、错误响应或更新数据库失败等。我想处理这些异常以记录错误并停止任务，甚至可能恢复任务。

现在我构建了一个新项目来使用 CompletableFutures 来模拟这个过程。我使用了以下代码：

public static Integer randomError() {
        Random rd = new Random();
        if(rd.nextBoolean()) {
            try {
                throw new Exception("RANDOM ERROR");
            } catch (Exception e) {
                e.printStackTrace();
            }
        } else {
            return rd.nextInt();
        }
        return 0;
    }


ExecutorService ex = Executors.newFixedThreadPool(64);
        System.out.println("Main thread: " + Thread.currentThread());
        //Starting tasks
        List<CompletableFuture> listTasks = new ArrayList<CompletableFuture>();
        List<String> listErrors = new ArrayList<String>();
        System.out.println("Starting threads...");
        for (int i = 0; i < 10; i++) {
            int counter = i;
            //Add tasks to TaskQueue (taskList)
            listTasks.add(
                CompletableFuture.supplyAsync(()->{
                    //Simulate step 1
                    return 0;
                },ex).thenApplyAsync(x -> {
                    //Simulate step 2
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return x + 1;
                }, ex).thenApplyAsync(x -> {
                    //Simulate step 3 with a potential error
                    randomError();
                    return x + 1;
                }, ex).thenApplyAsync(x -> {
                    //On error this shouldnt be executed?
                    //Simulate tep 4
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return x + 1;
                }, ex).thenAcceptAsync( x -> {
                      //Simulate COMPLETION step 5 
//                    listTasks.remove(counter);
                }, ex).exceptionally(e -> {
                    listErrors.add("ERROR: " + counter);
                    System.out.println(e);
                    return null;
                })
            );
        }
        System.out.println("Done");

现在这段代码创建了 10 个任务，每个任务由 5 个步骤组成。现在，当第 3 步产生异常时，第 4 步仍会执行。为什么？在我的串行监视器中，我看到抛出的错误，但 CompletableFuture 仍然可以正常完成。当我这样做的时候1 / 0;。这会产生一个错误，被.exceptionally(). 那是如何捕获的而不是自定义抛出的异常？

我想要的是错误，停止链条并去做.exceptionally()处理错误。

标签： javaexceptioncompletable-future