Publish 和 SelectMany 在这个查询中做了什么？答案

【问题标题】：What are Publish and SelectMany doing in this query?Publish 和 SelectMany 在这个查询中做了什么？
【发布时间】：2022-02-25 15:55:59
【问题描述】：

我有一个（或两个）反应式方法的新问题。在我的场景中，我需要一个能够在第一个任务未完成时抑制其他发出的任务的可观察序列，并最终得到如下结果：

Observable.Interval(TimeSpan.FromMilliseconds(200))
.Select(x => Observable.FromAsync(async () =>
{
    await Task.Delay(1000);
    // Simulating long running task
    Console.WriteLine(x);
}))
.Publish(x => x.FirstAsync().SelectMany(c => c).Repeat())
.Subscribe();

我尝试了谷歌，但我真的无法解释一些事情：

首先，它是如何工作的?????
阻止可观察对象到达订阅的反应序列究竟是什么？ Replay 究竟做了什么？在这种情况下，Replay 不应该重播任务吗？或者我不知道。任何人都可以详细解释该反应式查询中的每一步吗？ Publish 对这种选择器有什么作用。 Replay 在该查询中的表现如何？如果无论如何只会发出一个元素，为什么我需要在FirstAsync 上调用SelectMany。

【问题讨论】：

如果您的代码有效会有所帮助。
关于在先前启动的任务完成之前阻止新任务启动的初始任务，您可能正在寻找 ExhaustMap 运算符。

标签： c# .net asynchronous system.reactive

【解决方案1】：

.SelectMany(c => c) 是一种扁平化/合并嵌套序列的惯用方式。您可以将其替换为.Merge()，查询的行为将是相同的。
Publish 运算符与 Func<IObservable<TSource>, IObservable<TResult>> 参数一起使用时，订阅它所链接的查询，然后保持订阅状态，直到 lambda 生成的序列完成。因此，在您的情况下，通过将内部序列x.FirstAsync().SelectMany(c => c).Replay() 包装在Publish 中，您可以延迟从链式序列（Interval+Select+FromAsync）取消订阅，直到内部序列完成。内部序列永远不会完成，因此链式序列会永远每秒产生一个冷IObservable<Unit> 子序列。您可以通过在Publish 之前拦截Do 运算符来观察这种情况：

.Do(x => Console.WriteLine($"New subsequence: {x.GetType().Name}"))

Replay 运算符类似于Publish，区别在于Replay 具有过去通知的记忆，而Publish 没有任何记忆。我猜你的意图是附加Repeat 而不是Replay。不带参数的 Replay 产生一个“可连接”的 observable，它不会自动订阅链式序列。您必须手动 Connect 它，或者将 RefCount 运算符附加到它。在您的情况下，您什么都不做，因此生成的序列永远不会发出任何东西，也永远不会完成。这是一个令人讨厌的僵局。

【讨论】：

首先感谢。这确实是我一直在寻找的答案。但关于这一点，该序列中没有僵局。你可以自己试试。它会产生我需要的东西。每次第一个可观察对象完成时，它只会产生一个值，或者类似的东西。
我需要的是：想象一下有一个东西产生了许多值，但我必须只处理一个值并在分析该值时丢弃其他值。该分析需要很长时间，因此根本不应该发出所有产生的值。理想情况下，我需要在长任务运行时禁用 observable，一旦任务完成，我可以再次启用 observable。
我还是不明白 Publish 是做什么的。根据您的回答，似乎是某种 TakeUntilComplete()？
更新。我的错。我更正了这个问题，因为你在我的代码中对 Repeat 而不是 Replay 是正确的，实际上我有 Repeat() 但在添加这个问题时无法检查。
@VincentP 在长任务运行时禁用可观察对象，这通常是不可行的。 Observables 应该被被动地观察，而不是被观察者主动控制！我想到了两种可能性：（1）每次产生值时取消订阅 observable 并在完成长任务后重新订阅；（2）阻塞发出有关新值的通知的线程。它们都不能保证 observable 会被停止，但如果你绝望并且没有其他选择，你可以尝试一下。