12

如果发生错误，有没有办法让一个可观察的序列恢复执行序列中的下一个元素？从这篇文章看来，您需要在 Catch() 中指定一个新的可观察序列来恢复执行，但是如果您只需要继续处理序列中的下一个元素怎么办？有没有办法做到这一点？

更新：场景如下：我有一堆需要处理的元素。该处理由一系列步骤组成。我已将步骤分解为我想编写的任务。我按照此处发布的 ToObservable() 指南 将任务转换为可观察对象以进行组合。所以基本上我正在做这样的事情 -

foreach(element in collection)
{
   var result = from aResult in DoAAsync(element).ToObservable() 
         from bResult in DoBAsync(aResult).ToObservable() 
         from cResult in DoCAsync(bResult).ToObservable() 
         select cResult;
   result.subscribe( register on next and error handlers here)
 }

或者我可以这样：

var result = 
        from element in collection.ToObservable() 
        from aResult in DoAAsync(element).ToObservable() 
         from bResult in DoBAsync(aResult).ToObservable() 
         from cResult in DoCAsync(bResult).ToObservable() 
         select cResult;

即使假设其中一个元素的处理引发异常，这里继续处理其他元素的最佳方式是什么。我希望能够记录错误并继续前进。

4

2 回答 2

12

James 和 Richard 都提出了一些好的观点，但我认为他们没有为您提供解决问题的最佳方法。

詹姆斯建议使用.Catch(Observable.Never<Unit>()). 当他说“将......允许流继续”时他错了，因为一旦你遇到异常，流必须结束——这就是理查德在提到观察者和可观察者之间的合同时指出的。

此外，Never以这种方式使用会导致您的 observables 永远无法完成。

简而言之，这.Catch(Observable.Empty<Unit>())是将序列从以错误结尾的序列更改为以完成结尾的序列的正确方法。

您已经找到了使用SelectMany处理源集合的每个值以便捕获每个异常的正确想法，但是您遇到了一些问题。

您使用任务 (TPL) 只是为了将函数调用转换为 observable。这会强制您的 observable 使用任务池线程，这意味着该SelectMany语句可能会以不确定的顺序产生值。

您还隐藏了处理数据的实际调用，使重构和维护更加困难。

我认为您最好创建一个允许跳过异常的扩展方法。这里是：

public static IObservable<R> SelectAndSkipOnException<T, R>(
    this IObservable<T> source, Func<T, R> selector)
{
    return
        source
            .Select(t =>
                Observable.Start(() => selector(t)).Catch(Observable.Empty<R>()))
            .Merge();
}

使用此方法，您现在可以简单地执行此操作：

var result =
    collection.ToObservable()
        .SelectAndSkipOnException(t =>
        {
            var a = DoA(t);
            var b = DoB(a);
            var c = DoC(b);
            return c;
        });

这段代码要简单得多，但它隐藏了异常。如果你想在让你的序列继续的同时保持异常，那么你需要做一些额外的时髦。向扩展方法添加几个重载Materialize可以保持错误。

public static IObservable<Notification<R>> Materialize<T, R>(
    this IObservable<T> source, Func<T, R> selector)
{
    return source.Select(t => Notification.CreateOnNext(t)).Materialize(selector);
}

public static IObservable<Notification<R>> Materialize<T, R>(
    this IObservable<Notification<T>> source, Func<T, R> selector)
{
    Func<Notification<T>, Notification<R>> f = nt =>
    {
        if (nt.Kind == NotificationKind.OnNext)
        {
            try
            {
                return Notification.CreateOnNext<R>(selector(nt.Value));
            }
            catch (Exception ex)
            {
                ex.Data["Value"] = nt.Value;
                ex.Data["Selector"] = selector;
                return Notification.CreateOnError<R>(ex);
            }
        }
        else
        {
            if (nt.Kind == NotificationKind.OnError)
            {
                return Notification.CreateOnError<R>(nt.Exception);
            }
            else
            {
                return Notification.CreateOnCompleted<R>();
            }
        }
    };
    return source.Select(nt => f(nt));
}

这些方法允许你这样写：

var result =
    collection
        .ToObservable()
        .Materialize(t =>
        {
            var a = DoA(t);
            var b = DoB(a);
            var c = DoC(b);
            return c;
        })
        .Do(nt =>
        {
            if (nt.Kind == NotificationKind.OnError)
            {
                /* Process the error in `nt.Exception` */
            }
        })
        .Where(nt => nt.Kind != NotificationKind.OnError)
        .Dematerialize();

您甚至可以链接这些Materialize方法并使用ex.Data["Value"]&ex.Data["Selector"]来获取抛出错误的值和选择器函数。

我希望这有帮助。

于 2011-07-26T04:52:21.023 回答
1

IObservable和IObserveris之间的合同OnNext*(OnCompelted|OnError)?得到所有运营商的支持，即使不是由来源。

您唯一的选择是使用 重新订阅源Retry，但如果源返回每个描述的IObservable 实例，您将看不到任何新值。

您能否提供有关您的方案的更多信息？也许还有另一种看待它的方式。

编辑：根据您更新的反馈，听起来您只需要Catch：

var result = 
    from element in collection.ToObservable() 
    from aResult in DoAAsync(element).ToObservable().Log().Catch(Observable.Empty<TA>())
    from bResult in DoBAsync(aResult).ToObservable().Log().Catch(Observable.Empty<TB>()) 
    from cResult in DoCAsync(bResult).ToObservable().Log().Catch(Observable.Empty<TC>())
    select cResult;

这将错误替换为Empty不会触发下一个序列的错误（因为它SelectMany在幕后使用。

于 2011-05-19T07:39:27.523 回答