用Reactor抛出异常的正确方法

Question

我是一个新的项目Reactor和反应编程一般。

我目前正在编写一段类似于以下内容的代码：

Mono.just(userId)
    .map(repo::findById)
    .map(user-> {
        if(user == null){
            throw new UserNotFoundException();
        }
        return user;
    })
    // ... other mappings

这两种做法有什么实际区别吗？

Answer 1

有几种方法可以被认为是一种方便的异常抛出方式：

可以简化元素处理的方法之一是运算符handle。

下面的代码展示了如何使用它来解决我们的问题：

Mono.just(userId)
    .map(repo::findById)
    .handle((user, sink) -> {
        if(!isValid(user)){
            sink.error(new InvalidUserException());
        } else if (isSendable(user))
            sink.next(user);
        }
        else {
            //just ignore element
        }
    })

Mono.just(userId)
    .map(repo::findById)
    .concatMap(user-> {
        if(!isValid(user)){
            return Mono.error(new InvalidUserException());
        }
        return Mono.just(user);
    })

Flux.just(userId1, userId2, userId3)
    .map(repo::findById)
    .concatMap(user-> {
        if(!isValid(user)){
            return Flux.error(new InvalidUserException());
        }
        return Mono.just(user);
    })

注意！永远不要检查传入元素的可空性。Reactor项目永远不会为您发送null值，因为这违反了Reactive Streams规范（参见规则2.13）。因此，如果repo.findbyid返回null，Reactor将为您抛出NullPointerException。

从本质上讲，FlatMap被设计为从一次执行的多个子流中合并元素。这意味着flatMap下面应该有异步流，所以它们可能处理多个线程上的数据，或者可能是几个网络调用。随后，这样的期望会对实现产生很大的影响，因此FlatMap应该能够处理来自多个流(Threads)的数据（意味着使用并发数据结构），如果另一个流存在耗尽，则可以进入队列元素（意味着为每个子流的Queues)分配额外的内存），并且不违反反应流规范规则（意味着真正复杂的实现）。考虑到所有这些事实，以及我们将普通的map操作（同步）替换为使用flux/mono.error抛出异常的更方便的方式，我们不需要如此复杂的运算符，我们可以使用更简单的concatmap，它是为一次异步处理单个流而设计的，并且为了处理标量冷流进行了一些优化。

因此，当结果为空时引发异常的另一种方法是switchifempty操作符。下面的代码演示了如何使用这种方法：

Mono.just(userId)
    .flatMap(repo::findById)
    .switchIfEmpty(Mono.error(new UserNotFoundExeception()))

我们可以看到，在本例中，repo::FindByID应该将User的mono作为返回类型。因此，如果找不到User实例，结果流将为空。因此，Reactor将调用一个替代的mono，指定为switchifempty参数。

它可能被认为是可读性较差的代码或糟糕的实践（我自己的观点），但您可以使用Project Reactor按原样抛出异常（例如.map(v->throw...）)。尽管如此，在某种程度上，这样做可能会违反Reactive Streams规范（在此上下文中，从语义角度来看是违反的，因为您的运算符是订阅者链中的订阅者，因此，语义上，在lambda中抛出异常可以映射到在onnext方法中抛出异常，这违反了规范的规则2.13）。但是，由于Reactor将为您捕获抛出的异常，然后将其作为onerror信号传播到下游，因此并不禁止这样做。

1. 使用.handle运算符，以便提供复杂的元素处理
2. 使用concatmap+mono.error当我们需要在映射过程中抛出异常时，但是这种技术最适合异步元素处理的情况。
3. 使用flatmap+mono.error，当我们已经有flatmap
时
4. null作为返回类型是禁止的，因此，在下游映射中，您将获得意外的onerrornullpointerexception
而不是 nullpointerexception
5. 当调用某个特定函数的结果以空流结束时，需要发送错误信号时，请在所有情况下使用switchifempty