asynchronous - 接收 : subscribe with async function and ignore errors

Question

我想为可观察对象中的每个项目调用一个异步函数。正如回答here ，解决方案是使用 SelectMany。但是，如果异步方法抛出异常，订阅将终止。我有以下解决方案，它似乎有效:

obs.SelectMany(x => Observable
    .FromAsync(() => RunAsync())
    .Catch(Observable.Empty<string>()));

是否有更惯用的解决方案？

Answer 1

有一种标准方法可以观察 RunAsync 中发生的异常打电话，那是使用 .Materialize() .

.Materialize()方法变成一个 IObservable<T>序列为 IObservable<Notification<T>>您可以对 OnNext 进行推理的序列, OnError , 和 OnCompleted电话。

我写了这个查询:

var obs = Observable.Range(0, 10);

obs
    .SelectMany(x =>
        Observable
            .FromAsync(() => RunAsync())
            .Materialize())
    .Where(x => x.Kind != NotificationKind.OnCompleted)
    .Select(x => x.HasValue ? x.Value : (x.Exception.Message + "!"))
    .Subscribe(x => x.Dump());

有了这个支持代码:

private int counter = 0;
private Random rnd = new Random();

private System.Threading.Tasks.Task<string> RunAsync()
{
    return System.Threading.Tasks.Task.Factory.StartNew(() =>
    {
        System.Threading.Interlocked.Increment(ref counter);
        if (rnd.NextDouble() < 0.3)
        {
            throw new Exception(counter.ToString());
        }
        return counter.ToString();
    });
}

当我运行它时，我得到这样的输出:

2
4
5
1!
6
7
3!
10
8!
9

以 ! 结尾的每一行正在调用 RunAsync这导致了异常。