CompletableFuture在exceptionaly（）块发生异常时递归重新启动

Question

很难可靠地重试后台任务，该任务会向邮件服务发送请求，以获取最新的电子邮件。一旦成功接收电子邮件，应在然后accept（）阻止-持久化电子邮件中继续执行，但是如果出现异常，我必须重新运行邮件检索，直到成功尝试，成功后应持久化邮件并停止。如果我做错了，请看一看并提出建议。

   private void retrieveMailsAsync(User user) {
    CompletableFuture.supplyAsync(() -> {
        try {
            return mailService.getEmails(user.getName(), user.getPassword());
        } catch (InvalidAuthentication | TimeoutException | BadGatewayException e) {
            throw new CompletionException(e);
        }
    }).thenAccept(email -> {
        mailService.persist(email);
    }).exceptionally(ex -> {
        log.log(Level.SEVERE, "Exception retrieveMailsAsync emails, Retrying retrieveMailsAsync:: ", ex.getCause());
        retrieveMailsAsync(user);
        return null;
    });
}

另外，请看一下我是如何处理选中的异常的，将其包装到CompletionException中并重新排序-这里的主要思想是在一个exception（）块中处理所有异常（已定义的选中异常和运行时异常），而不是将它们记录在catch块中并返回null。

提前感谢大家，希望我没有做一些愚蠢的事情，或者至少Java 8已经有了可靠的解决方案。

Answer 1

我认为你可以通过以下方式实现：

public static void main(String[] args) {
    String result = call(new User().setName("name").setPassword("p")).join();
    System.out.println(result);
}

private static CompletableFuture<String> call(User user) {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> retrieveMailsAsync(user));
    return future.handleAsync((String result, Throwable ex) -> {
        // or any other Predicate that is satisfied against ex
        if(ex != null) {
            return call(user);
        } else {
            return future;
        }
    }).thenCompose(Function.identity());
}

编辑

那么，您在更改上述代码时遇到了哪些障碍，例如：

static ExecutorService service = Executors.newFixedThreadPool(1);

public static void main(String[] args) {
    call(new User().setName("name").setPassword("p"))
         // chain any other action here, like mailService.persist(email);
         .thenAcceptAsync(
            System.out::println,
            service
    );
    System.out.println("Continue main thread");
}

private static CompletableFuture<String> call(User user) {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> retrieveMailsAsync(user), service);
    return future.handleAsync((String result, Throwable ex) -> {
        // or any other Predicate that is satisfied against ex
        if(ex != null) {
            return call(user);
        } else {
            return future;
        }
    }).thenCompose(Function.identity());
}

Answer 2

我在评论中的意思是：

   private void retrieveMailsAsync(User user) {
    CompletableFuture.supplyAsync(() -> {
        while (continueQuery()) { // true for infinite retries, or some other logic
          try {
              return mailService.getEmails(user.getName(), user.getPassword());
          } catch (InvalidAuthentication | TimeoutException | BadGatewayException e) {
              log.log(Level.SEVERE, "Exception retrieveMailsAsync emails, Retrying retrieveMailsAsync: ", e);
          }
        }
        return null;
    }).thenAccept(email -> {
        mailService.persist(email);
    });
}

也就是说，你只需在提交ted runnable中重试，直到你不再有异常。