反应式扩展订阅呼叫等待答案

【问题标题】：Reactive Extensions Subscribe calling await反应式扩展订阅呼叫等待
【发布时间】：2023-03-04 13:01:01
【问题描述】：

我想为响应式扩展 Observable 引发的每个事件执行异步调用。我还试图让所有内容保持同步，因为我希望在处理下一个事件之前完成异步调用。

如何去做类似于以下的事情？我说类似下面的代码不会编译。

settingsChangedInMemory
    .Subscribe(async _ => {
        var settings = Extract();
        await SaveSettings(settings);
    });

我不确定它是否会改变任何东西，但我需要订阅多个 Observable。例如像这样的另一个订阅。

settingsChangedOnDisk
    .Subscribe(async _ => {
        var settings = await ReadSettings(settings);
        Apply(settings);
    });

您将如何使用响应式扩展来做到这一点？

【问题讨论】：

【解决方案1】：

怎么样：

settingsChangedInMemory
    .SelectMany(async _ => await SaveSettings(Extract()))
    .Subscribe(x => Apply(x));

永远不要将async 放在Subscribe 中，您总是希望将它放在SelectMany 中。

【讨论】：

从 RX 2.0 开始，这仍然是真的吗（早于你的回答）？上面的代码现在似乎可以编译了，这里有一个来自 RX 团队的示例 blogs.msdn.com/b/rxteam/archive/2012/03/12/… 的 Subscribe(async () => ...)
绝对正确
你介意详细说明保罗吗？他们明确声明他们支持异步订阅，而您的回复并没有真正解释为什么 Reactive 的开发人员是错误的。
async void 方法可以在任何可以调用void 方法的地方调用。因此Subscribe 支持异步方法，因为这些方法是无效的。无论如何，您不能等待一次性/订阅。如果您需要类似的内容，请先使用SelectMany 等，然后再订阅。
async 可以安全地进入Select 以及SelectMany 吗？

【解决方案2】：

您可以像这样使用 Reactive Extensions (Rx) 2.0 中发布的新 ForEachAsync 方法：

await observable
    .ForEachAsync(async x =>
    {
        Console.WriteLine(x);
        await Task.Delay(1000);
    });

ForEachAsync 返回一个 Task<T>，它在 observable 完成时完成。更多信息请参阅我的博客文章 here 或响应式扩展团队的 blog post。

【讨论】：