在 Observable 链中执行基于任务的方法 => IObservable<IObservable<Unit>>答案

【问题标题】：Executing Task based methods in Observable chain => IObservable<IObservable<Unit>>在 Observable 链中执行基于任务的方法 => IObservable<IObservable<Unit>>
【发布时间】：2022-01-28 05:04:10
【问题描述】：

我有很多反应性的代码，但需要调用基于任务的方法。

例如，在这个 sn -p PracticeIdChanged 中是一个 IObservable。当 PracticeIdChanged 触发时，我希望系统通过重新加载一些内容来做出反应，并具有如下所示的代码：

 PracticeIdChanged.Subscribe(async x => {
            SelectedChargeInfo.Item = null;
            await LoadAsync().ConfigureAwait(false);
        });

虽然它似乎工作正常，但我收到有关在订阅中执行异步代码的警告。此外，我认为这是一种代码异味，因为我正在混合两个单独的线程模型，我认为这可能会在以后咬我。

像这样的重构工作（甚至）没有像 .Merge()、.Switch() 或 .Concat() 这样的组合方法

PracticeIdChanged
            .Do(_ => SelectedChargeInfo.Item = null)
            .Select(_ => LoadAsync().ToObservable())
            .Subscribe();

当 PracticeIdChanged 触发 LoadAsync 方法时执行

Select 产生一个看起来很奇怪的 IObservable。这可以吗，还是需要一些组合功能，如 .Merge 或 .Switch()

在很多地方，我使用 SelectMany 来执行基于任务的方法，但它需要返回任务，这需要更改上面示例中基于任务的方法的签名，而我不想这样做。

【问题讨论】：

标签： c# system.reactive rx.net

【解决方案1】：

这取决于您希望从结果序列中获得什么样的通知，以及在出现错误时您想要什么样的行为。在您的示例中，您将.Subscribe() 传递给序列，而没有传递任何处理程序（onNext/onError/onCompleted），这表明您对收到任何通知都不感兴趣。您不关心异步操作的完成，所有这些操作基本上都会被解雇和遗忘。此外，一个异步操作的失败不会对其余操作造成任何影响：已经开始的异步操作将继续运行（它们不会被取消），并且不会阻碍开始新的异步操作。最后，源序列 (PracticeIdChanged) 的失败将导致未处理的异常，这将使进程崩溃。如果这是您想要的行为，那么您当前的设置就是您所需要的。

为了比较，让我们考虑一下这个设置：

await PracticeIdChanged
    .Do(_ => SelectedChargeInfo.Item = null)
    .Select(_ => Observable.FromAsync(ct => LoadAsync(ct)))
    .Merge()
    .DefaultIfEmpty();

此设置假定LoadAsync 方法具有CancellationToken 参数。结果序列为awaited。当所有LoadAsync 操作完成，或者其中任何一个操作失败，或者源序列失败时，await 将完成。在失败的情况下，所有当前正在运行的异步操作都会收到一个取消信号，以便它们可以快速退出。 await 将不会等待它们完成。只有发生的第一个错误会作为异常传播。可以通过将 await 包装在 try/catch 块中来处理此异常。不可能出现无法捕获的、进程崩溃的、未处理的异常。

链末尾的DefaultIfEmpty 的目的是防止InvalidOperationException，以防源序列发出零元素。这是空可观察序列的这种奇怪“功能”的解决方法，在同步或异步等待时抛出。

【讨论】：