Rx：可观察值是否“可重复”？就像 IEnumerable 一样，如果不是，这段代码是如何工作的？

Question

昨天我观看了截屏视频 写作您的第一个 Rx 应用程序（在第 9 频道），Wes Dyer 展示了如何使用 反应式扩展 (Rx)。我仍然不明白的事情：

在截屏视频即将结束时，Wes Dyer 输入以下内容：

var q = from start in mouseDown
        from delta in mouseMove.StartsWith(start).Until(mouseUp)
                       .Let(mm => mm.Zip(mm.Skip(1), (prev, cur) =>
                           new { X = cur.X - prev.X, Y = cur.Y - prev.Y }))
        select delta;

简而言之，q 是一个将鼠标移动坐标增量推送给其订阅者的可观察对象。

我不明白的是 mm.Zip(mm.Skip(1), ...) 是如何工作的！？

据我所知，IObservable 不像 IEnumerable 那样是可枚举的。由于 IEnumerable 的“拉”性质，它可以一次又一次地迭代，始终产生相同的项目。 （至少对于所有行为良好的枚举应该都是这样。）IObservable 的工作方式不同。项目被推送给订阅者一次，仅此而已。在上面的示例中，鼠标移动是单个事件，如果未记录在内存中，则无法重复。

那么，.Zip 与 .Skip(1) 的组合如何才能发挥作用，因为它们处理的鼠标事件是单个、不可重复的事件呢？这个操作不是需要mm独立“看”两次吗？

---

作为参考，这里是 Observable.Zip 的方法签名：

public static IObservable<TResult> Zip <TLeft, TRight, TResult>
(
    this IObservable<TLeft>       leftSource,     //  = mm
    IObservable<TRight>           rightSource,    //  = mm.Skip(1)
    Func<TLeft, TRight, TResult>  selector
)

---

PS: 我刚刚看到还有另一个 关于 Zip 运算符的截屏视频，非常有见地。

Answer 1

这个操作不是需要mm被独立“看”两次吗？

这实际上就是您问题的答案：您可以多次订阅相同的 IObservable 序列。

mm.Skip(1) 订阅 mm 并向其自己的订阅者隐藏第一个值。
Zip 是 mm.Skip(1) 和 mm 的订阅者。由于 mm 产生的值比 mm.Skip(1) 多一个，因此 Zip 在内部始终缓冲来自 mm 的最后一个 mousemove 事件，以便与下一个未来的 mousemove 事件一起压缩它。然后选择器函数可以选择两者之间的增量。

您应该注意的另一件事是（这是问题标题的真正答案），此 Observable.FromEvent-IObservable 是一个热门 可观察，因此不可重复。但有些冷 Observables 实际上是可重复的，比如 Observable.Range(0,10)。在后一种情况下，每个订阅者将收到相同的 10 个事件，因为它们是为每个订阅者独立生成的。对于鼠标移动事件，情况并非如此（您不会获得过去的鼠标移动事件）。但因为 Zip 同时订阅左右序列，在这种情况下它可能是相同的。

PS：您还可以创建一个热/不可重复的 IEnumerable：它不需要为每个枚举器返回相同的值。例如，您可以创建一个 IEnumerable，它会等待鼠标移动事件发生，然后产生该事件。在这种情况下，枚举器总是会阻塞（设计不好），但这是可能的。 ;)

Answer 2

啊哈！ Zip screencast给了我一个重要的线索：Zip“记住”项目，以解释项目可能从一个可观察到的比另一个更早到达这一事实。我会尝试回答我的问题，如果我错了，希望有人能纠正我。

Zip 将来自两个可观察序列的输入配对，如下所示（字母和数字是“事件”）：

mm                        ----A---------B-------C------D-----------E----->
                              |         |       |      |           |
                              |         |       |      |           |
mm.Skip(1)                ----+---------1-------2------3-----------4----->
                              |         |       |      |           |
                              |         |       |      |           |
mm.Zip(mm.Skip(1), ...)   ----+--------A,1-----B,2----C,3---------D,4---->

它确实必须进行内部缓冲。在我发布的代码中，mm 是真正的、“实时”可观察的。 mm.Skip(1) 类似于从它派生的状态机。 Alex Paven 的回答简要解释了它是如何工作的。

因此， mm.Zip(mm.Skip(1), ...) 确实查看了 mm 两次，一次直接查看，一次通过 Skip( n) 过滤器。而且由于可观察量不是可重复的序列，因此它会进行内部缓冲，以考虑一个序列将比另一个序列更快地产生项目这一事实。

（我快速浏览了 .NET Reflector 的 Rx 源代码，事实上，Zip 涉及一个 Queue。）

Answer 3

项目被推送给订阅者一次，仅此而已。

是的，一个项目被推送一次，但该项目是事件“序列”之一。序列仍然是序列。这就是 Skip 起作用的原因 - 它会跳过一项，然后当下一项到来时处理它（不跳过它）。