system.reactive - Rx Observable 能否优雅地处理运算符中的异常并继续？

Question

即，通过传递错误条件和 不是 停止整个 Observable？

我的 Observable 以用户提供的来自常见递送服务(FedEx、UPS、DHL 等)的包裹跟踪号码列表开始，在线查找预期递送日期，然后根据从今天起的天数返回这些日期(即“在 3 天内”而不是“1 月 22 日”)。问题是，如果任何单独的查找导致异常，整个流都会停止，并且不会查找其余的代码。没有能力优雅地处理，比如 UnknownTrackingCode Exception ，因此 Observable 不能保证它会查找用户提交的所有代码。

public void getDaysTillDelivery(List<String> tracking_code_list) {
        Observable o = Observable.from(tracking_code_list)
                   // LookupDeliveryDate performs network calls to UPS, FedEx, USPS web sites or APIs
                   // it might throw: UnknownTrackingCode Exception, NoResponse Exception, LostPackage Exception
                .map(tracking_code -> LookupDeliveryDate(tracking_code))
                .map(delivery_date -> CalculateDaysFromToday(delivery_date));
        o.subscribe(mySubscriber); // will handle onNext, onError, onComplete
}

由于一个错误而停止 Observable 流是设计使然:

http://reactivex.io/documentation/operators/catch.html

Handling Exceptions in Reactive Extensions without stopping sequence

https://groups.google.com/forum/#!topic/rxjava/trm2n6S4FSc

可以克服默认行为，但前提是首先消除 Rx 的许多好处:

我可以包LookupDeliveryDate所以它返回日期代替异常(例如 1899-12-31 用于 UnknownTrackingCode Exception )但这可以防止“松散耦合代码”，因为 CalculateDaysFromToday需要处理这些特殊情况

我可以用 try/catch 包围每个匿名函数和块，但这基本上阻止了我使用 lambdas

我可以用 if/thens引导代码路径，但这可能需要维护一些状态并消除确定性评估

显然，每个步骤的错误处理会阻止在 Subscriber 中整合所有错误处理。

编写我自己的错误处理运算符是可能的，但文档很少

有没有更好的方法来处理这个问题？

Answer 1

如果出现错误，您到底希望发生什么？你只是想把那个条目扔掉还是想让下游的东西用它做点什么？

如果您希望下游采取某些行动，那么您实际上是将错误转化为数据(有点像您返回 1899-12-31 的哨兵值来表示错误的示例)。根据定义，此策略意味着下游的所有内容都需要了解数据流可能包含错误而不是数据，并且必须对其进行修改以处理它。

但是，与其产生魔法值，不如将 Observable流成Either的流值。该值要么是日期，要么是错误。下游的所有内容都收到此 Either对象并可以询问它是否有值或错误。如果它有一个值，他们可以产生一个新的 Either对象及其计算结果。如果它有一个错误并且他们不能对它做任何事情，他们可以自己产生一个错误。

我不知道 Java 语法，但这就是它在 c# 中的样子:

Observable.From(tracking_code_list)
    .Select(t =>
        {
            try { return Either.From(LookupDeliveryDate(t)); }
            catch (Exception e)
            {
                return Either.FromError<Date>(e);
            }
        })
     .Select(dateEither =>
        {
            return dateEither.HasValue ?
                Either.From(CalculateDaysFromToday(dateEither.Value)) :
                Either.FromError<int>(dateEither.Error);
        })
     .Subscribe(value =>
        {
            if (value.HasValue) mySubscriber.OnValue(value.Value);
            else mySubscribe.OnError(value.Error);
        });

您的另一个选择是“处理”/在错误发生时抑制错误。根据您的需要，这可能就足够了。在这种情况下，只需 LookupDeliveryDate返回魔法日期而不是异常，然后添加 .filter在它们到达 CalculateDaysFromToay 之前过滤掉神奇的日期.