java - 使用 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.
    }
}

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

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

Answer 1

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

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

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

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

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

该文件还说：

…为了简化监控、调试和跟踪，所有生成的异步任务都是标记接口的实例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); }
});

Answer 2

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

下面的代码按预期打印“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");
  }
}

Answer 3

我不太确定您要达到的目标。如果您的供应商抛出异常，当您调用时，testFuture .get()您将由供应商抛出的java.util.concurrent.ExecutionException任何异常引起，您可以通过调用getCause()on来检索ExecutionException。

或者，正如您所提到的，您可以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());
    }

}

打印此结果：

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

编辑：

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

原始代码：

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

修改后的代码：

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

Answer 4

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

public class Result {

    private Integer   integer;
    private Exception exception;

    // getter setter

}

接着：

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);
}

Answer 5

只需将检查的异常包装成CompletionException

CompletableFuture使用时的异常处理要考虑的另一点completeExceptionally()是，确切的异常将在其中可用，handle()但whenComplete()在CompletionException调用join()或转发到任何下游阶段时将被包装。

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

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

考虑到这一点，最好在安全的情况下使用它并让您的异常处理程序解开CompletionExceptions.

如果你这样做了，那么在 上设置确切的（检查的）异常就没有意义了，CompletableFuture并且直接将检查的异常包装起来要简单得多CompletionException：

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

为了将此方法与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.
    }
}

输出：

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

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

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