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

Java 8 Supplier Exception handling with CompletableFuture

提问by ayush

Consider the following code

考虑以下代码

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.
    }
}

The above thing works fine. However sendNumberscould also throw a checked exception in my case, like:

上面的东西工作正常。但是sendNumbers，在我的情况下也可能抛出一个已检查的异常，例如：

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

Now I want to handle this exception as yin my biConsumer. This will help me in handling the result as well as exception (if any) inside a single function (biConsumer).

现在我想像y在我的biConsumer. 这将帮助我处理单个函数 ( biConsumer) 中的结果以及异常（如果有）。

Any ideas? Can I use CompletableFuture.exceptionally(fn)here or anything else?

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

回答by Holger

The factory methods using the standard functional interfaces aren't helpful when you want to handle checked exceptions. When you insert code catching the exception into the lambda expression, you have the problem that the catch clause needs the CompletableFutureinstance to set the exception while the factory method needs the Supplier, chicken-and-egg.

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

You could use an instance field of a class to allow mutation after creation, but in the end, the resulting code isn't clean and more complicated that a straight-forward Executor-based solution. The documentation of CompletableFuturesays:

您可以使用类的实例字段在创建后允许突变，但最终，生成的代码并不干净，而且比Executor基于直接的解决方案更复杂。的文档CompletableFuture说：

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

• 所有没有显式 Executor 参数的异步方法都使用ForkJoinPool.commonPool()...

So you know the following code will show the standard behavior of CompletableFuture.supplyAsync(Supplier)while handling checked exceptions straight-forward:

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

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

The documentation also says:

文档还说：

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

… 为了简化监控、调试和跟踪，所有生成的异步任务都是标记接口的实例CompletableFuture.AsynchronousCompletionTask。

If you want to adhere to this convention to make the solution even more behaving like the original supplyAsyncmethod, change the code to:

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

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

回答by assylias

You are already catching the exception in y. Maybe you did not see it because mainexited before your CompletableFuture had a chance to complete?

您已经在y. 也许你没有看到它，因为main在你的 CompletableFuture 有机会完成之前退出了？

The code below prints "null" and "Hello" as expected:

下面的代码按预期打印“null”和“Hello”：

public static void main(String args[]) throws InterruptedException {
  TestCompletableFuture testF = new TestCompletableFuture();
  testF.start();
  Thread.sleep(1000); //wait for the CompletableFuture to complete
}

public static class TestCompletableFuture {
  BiConsumer<Integer, Throwable> biConsumer = (x, y) -> {
    System.out.println(x);
    System.out.println(y);
  };
  public void start() {
    CompletableFuture.supplyAsync(SupplyNumbers::sendNumbers)
            .whenComplete(biConsumer);
  }
}

static class SupplyNumbers {
  public static Integer sendNumbers() {
    throw new RuntimeException("Hello");
  }
}

回答by Jaroslaw Pawlak

I am not quite sure what you are trying to achieve. If your supplier throws an exception, when you call testFuture .get()you will get java.util.concurrent.ExecutionExceptioncaused by any exception that was thrown by the supplier, that you can retrieve by calling getCause()on ExecutionException.

我不太确定你想要实现什么。如果当你打电话给你的供应商抛出一个异常，testFuture .get()你会得到java.util.concurrent.ExecutionException通过，是由供应商抛出的异常引起的，你可以通过调用检索getCause()上ExecutionException。

Or, just as you mentioned, you can use exceptionallyin the CompletableFuture. This code:

或者，正如您提到的，您可以exceptionally在CompletableFuture. 这段代码：

public class TestCompletableFuture {

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

    public static void main(String args[]) throws Exception {
        Supplier<Integer> numberSupplier = () -> {
            throw new RuntimeException(); // or return integer
        };

        CompletableFuture<Integer> testFuture = CompletableFuture.supplyAsync(numberSupplier)
                .whenComplete(biConsumer)
                .exceptionally(exception -> 7);

        System.out.println("result = " + testFuture.get());
    }

}

Prints this result:

打印此结果：

null
java.util.concurrent.CompletionException: java.lang.RuntimeException
result = 7

EDIT:

编辑：

If you have checked exceptions, you can simply add a try-catch.

如果你有检查过的异常，你可以简单地添加一个 try-catch。

Original code:

原始代码：

Supplier<Integer> numberSupplier = new Supplier<Integer>() {
    @Override
    public Integer get() {
        return SupplyNumbers.sendNumbers();                     
    }
};

Modified code:

修改后的代码：

Supplier<Integer> numberSupplier = new Supplier<Integer>() {
    @Override
    public Integer get() {
        try {
            return SupplyNumbers.sendNumbers();                     
        } catch (Excetpion e) {
            throw new RuntimeExcetpion(e);
        }
    }
};

回答by dieter

Perhaps you could use new Object to wrap your integer and error like this:

也许您可以使用 new Object 来包装您的整数和错误，如下所示：

public class Result {

    private Integer   integer;
    private Exception exception;

    // getter setter

}

And then:

接着：

public void start(){
    Supplier<Result> numberSupplier = new Supplier<Result>() {
        @Override
        public Result get() {
            Result r = new Result();
            try {
                r.setInteger(SupplyNumbers.sendNumbers());
            } catch (Exception e){
                r.setException(e);
            }
            return r;

        }
    };
    CompletableFuture<Result> testFuture = CompletableFuture.supplyAsync(numberSupplier).whenComplete(biConsumer);
}

回答by Didier L

Another point to take into account with exception handling in CompletableFuturewhen using completeExceptionally()is that the exact exception will be available in handle()and whenComplete()but it will be wrapped in CompletionExceptionwhen calling join()or when it is forwarded to any downstream stage.

另一点要考虑到与异常处理中CompletableFuture使用时completeExceptionally()是准确的异常将是可用handle()和whenComplete()但将包裹在CompletionException主叫时join()或当它被转发到任何下游级。

A handle()or exceptionally()applied to a downstream stage will thus see a CompletionExceptioninstead of the original one, and will have to look at its cause to find the original exception.

因此，应用于下游阶段的a handle()orexceptionally()将看到 aCompletionException而不是原始阶段，并且必须查看其原因以找到原始异常。

Moreover, any RuntimeExceptionthrown by any operation (including supplyAsync()) is also wrapped in a CompletionException, except if it is already a CompletionException.

此外，任何RuntimeException操作（包括supplyAsync()）抛出的任何内容也包含在 a 中CompletionException，除非它已经是 a CompletionException。

Considering this, it is better to play it on the safe side and have your exception handlers unwrap the CompletionExceptions.

考虑到这一点，最好在安全方面进行播放并让您的异常处理程序解开CompletionExceptions。

If you do that, there is no point anymore to set the exact (checked) exception on the CompletableFutureand it is much simpler to wrap checked exceptions in CompletionExceptiondirectly:

如果你这样做了，就没有必要在 上设置确切的（已检查的）CompletableFuture异常，CompletionException直接包装已检查的异常要简单得多：

Supplier<Integer> numberSupplier = () -> {
    try {
        return SupplyNumbers.sendNumbers();
    } catch (Exception e) {
        throw new CompletionException(e);
    }
};

To compare this approach with Holger's approach, I adapted your code with the 2 solutions (simpleWrap()is the above, customWrap()is Holger's code):

为了将此方法与Holger 的方法进行比较，我将您的代码调整为 2 个解决方案（simpleWrap()上面customWrap()是 Holger 的代码）：

public class TestCompletableFuture {

    public static void main(String args[]) {
        TestCompletableFuture testF = new TestCompletableFuture();
        System.out.println("Simple wrap");
        testF.handle(testF.simpleWrap());
        System.out.println("Custom wrap");
        testF.handle(testF.customWrap());
    }

    private void handle(CompletableFuture<Integer> future) {
        future.whenComplete((x1, y) -> {
            System.out.println("Before thenApply(): " + y);
        });
        future.thenApply(x -> x).whenComplete((x1, y) -> {
            System.out.println("After thenApply(): " + y);
        });
        try {
            future.join();
        } catch (Exception e) {
            System.out.println("Join threw " + e);
        }
        try {
            future.get();
        } catch (Exception e) {
            System.out.println("Get threw " + e);
        }
    }

    public CompletableFuture<Integer> simpleWrap() {
        Supplier<Integer> numberSupplier = () -> {
            try {
                return SupplyNumbers.sendNumbers();
            } catch (Exception e) {
                throw new CompletionException(e);
            }
        };
        return CompletableFuture.supplyAsync(numberSupplier);
    }

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

class SupplyNumbers {
    public static Integer sendNumbers() throws Exception {
        throw new Exception("test"); // just for working sake its not  correct.
    }
}

Output:

输出：

Simple wrap
After thenApply(): java.util.concurrent.CompletionException: java.lang.Exception: test
Before thenApply(): java.util.concurrent.CompletionException: java.lang.Exception: test
Join threw java.util.concurrent.CompletionException: java.lang.Exception: test
Get threw java.util.concurrent.ExecutionException: java.lang.Exception: test
Custom wrap
After thenApply(): java.util.concurrent.CompletionException: java.lang.Exception: test
Before thenApply(): java.lang.Exception: test
Join threw java.util.concurrent.CompletionException: java.lang.Exception: test
Get threw java.util.concurrent.ExecutionException: java.lang.Exception: test

As you'll notice, the only difference is that the whenComplete()sees the original exception before thenApply()in the customWrap()case. After thenApply(), and in all other cases, the original exception is wrapped.

您会注意到，唯一的区别是在案例whenComplete()之前看到了原始异常。在 之后，以及在所有其他情况下，原始异常被包装。thenApply()customWrap()thenApply()

The most surprising thing is that get()will unwrap the CompletionExceptionin the "Simple wrap" case, and replace it with an ExecutionException.

最令人惊讶的是，它get()会CompletionException在“Simple wrap”案例中解开，并将其替换为ExecutionException.