如何在CompletableFuture处理程序之外获取异常？

Question

我有以下情况，我试图看看是否有解决方案：

• 两个Spring服务调用必须并行进行（一个是现有的服务调用/逻辑，另一个是新添加的服务调用）

然而，一条快乐的道路应该是直截了当的，当涉及到服务发出的错误时，应该遵守以下规则：

>

如果其中只有一个失败，则通过另一个服务（异步）记录错误，并且API只返回成功服务的结果--这可以通过相应的@Async线程来完成。

@Service
public class Serv1 interface ServInf {
 @Async("customPool")
 public CompletableFuture<List<Obj>> getSomething(int id) {
   // The service ensures that the list is never null, but it can be empty
   return CompletableFuture.completedFuture(/* calling an external RESTful API */);
 }
}

@Service
public class Serv2 interface ServInf {
 @Async("customPool")
 public CompletableFuture<List<Obj>> getSomething(int id) {
   // The service ensures that the list is never null, but it can be empty
   return CompletableFuture.completedFuture(/* calling another external RESTful API */);
     }
 }

@RestController
public class MyController {

 /** Typical service @Autowired's */

 @GetMapping(/* ... */)
 public WrapperObj getById(String id) {

     CompletableFuture<List<String>> service1Result =
             service1.getSomething(id)
                     .thenApply(result -> {
                         if (result == null) { return null; }
                         return result.stream().map(Obj::getName).collect(Collectors.toList());
                     })
                     .handle((result, exception) -> {
                         if (exception != null) {
                             // Call another asynchronous logging service which should be easy
                             return null;
                         } else {
                             return result;
                         }
                     });

     CompletableFuture<List<String>> service2Result =
             service2.getSomething(id)
                     .thenApply(result -> {
                         if (result == null) { return null; }
                         return result.stream().map(Obj::getName).collect(Collectors.toList());
                     })
                     .handle((result, exception) -> {
                         if (exception != null) {
                             // Call another asynchronous logging service which should be easy
                             return null;
                         } else {
                             return result;
                         }
                     });

     // Blocking till we get the results from both services
     List<String> result1 = service1Result.get();
     List<String> result2 = service2Result.get();

     /** Where to get the exceptions thrown by the services if both fail
     if (result1 == null && result2 == null) {
         /** Signal that the API needs to fail as a whole */
         throw new CustomException( /** where to get the messages? */);
     }

     /** merge and return the result */
 }
}

我的问题是，因为这些服务返回某个对象的列表，即使我使用CompletableFuture。handle（）并检查是否存在异常，我无法返回异常本身以捕获并让Spring Advice类处理它（链接以返回列表）。

我想到的一件事是使用AtomicReference，以便捕获异常，并在handle（）中设置它们，并在未来完成后使用它们，例如。

AtomicReference<Throwable> ce1 = new AtomicReference<>();
AtomicReference<Throwable> ce2 = new AtomicReference<>();

.handle((result, exception) -> {
    if (exception != null) {
        ce1.set(exception);
        return null; // This signals that there was a failure
    } else {
        return result;
    }
});

List<String> result1 = service1Result.get();
List<String> result2 = service2Result.get();

/** Where to get the exceptions thrown by the services if both fail
if (result1 == null && result2 == null) {
    /** Signal that the API needs to fail as a whole */
    throw new CustomException(/** do logic to capture ce1.get().getMessage() + ce2.get().getMessage() */);
}

首先，在这种多线程异步调用中，这听起来像是一个可行的解决方案吗？

其次，这看起来很混乱，所以我想知道是否有一种更优雅的方法可以在Spring异步池之外捕获这些异常，并在主线程中处理它，例如，组合异常信息并将其抛出到Spring Advice异常处理程序。




Answer 1

可完成的期货交易处理起来相当麻烦，但在国际海事组织（IMO）看来，这将是一种更具功能性和反应性的方法。

我们需要从https://stackoverflow.com/a/30026710/1225328:

static<T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> com) {
    return CompletableFuture.allOf(com.toArray(new CompletableFuture<?>[0]))
            .thenApply(v -> com.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList())
            );
}

然后，我使用可选来表示操作的状态，但是trymonad会更适合（所以如果您的代码库中有这样的实用程序，请使用一个-Java还没有在本地带来一个）：

CompletableFuture<Optional<List<Object>>> future1 = service1.getSomething().thenApply(Optional::of).exceptionally(e -> {
    // log e
    return Optional.empty();
});
CompletableFuture<Optional<List<Object>>> future2 = service2.getSomething().thenApply(Optional::of).exceptionally(e -> {
    // log e
    return Optional.empty();
});

现在等待两个未来，并在获得结果后进行处理：

CompletableFuture<List<Object>> mergedResults = sequence(Arrays.asList(future1, future2)).thenApply(results -> {
    Optional<List<Object>> result1 = results.get(0);
    Optional<List<Object>> result2 = results.get(1);
    if (result1.isEmpty() && result2.isEmpty()) {
        throw new CustomException(...);
    }
    // https://stackoverflow.com/a/18687790/1225328:
    return Stream.of(
            result1.map(Stream::of).orElseGet(Stream::empty),
            result2.map(Stream::of).orElseGet(Stream::empty)
    ).collect(Collectors.toList());
});

然后理想情况下，您可以直接返回mergedResults，让框架为您处理，这样您就不会阻塞任何线程，或者您可以。get（）（这将阻止线程），如果抛出了CustomException（或任何其他异常）（可在e.getCause（）中访问），它将抛出ExecutionException）。

如果您已经在使用ProjectReactor（或同等产品），这看起来会更简单，但想法大致相同。