CompletableFuture 在 exceptiony() 块异常时递归重启

Question

很难对后台任务进行可靠的重试，该任务会向邮件服务发送请求以获取最新的电子邮件。成功接收电子邮件后，应在thenAccept() 块中继续执行 - 保留电子邮件，但是如果发生异常，我必须重新运行邮件检索，直到成功尝试，成功后应该保留邮件并停止。如果我做错了，请看看并提出建议。

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

PS 也请看看我是如何处理检查异常，将其包装到 CompletionException 并重新抛出 - 这里的主要思想是在一个 exceptionally() 块中处理所有异常（定义检查和运行时）而不是记录它们在 catch 块中并返回 null。

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

Answer 1

我在评论中的意思是：

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

也就是说，您只需在 submitted 可运行项中重试，直到不再出现异常。

Answer 2

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

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