将一个 observable 与另一个 observable 的最新组合答案

【问题标题】：combining one observable with latest from another observable将一个 observable 与另一个 observable 的最新组合
【发布时间】：2014-12-16 14:58:13
【问题描述】：

我正在尝试组合两个值共享某个键的可观察对象。

每当第一个 observable 产生一个新值时，我想产生一个新值，并结合第二个 observable 的最新值，该选择取决于第一个 observable 的最新值。

伪代码示例：

var obs1 = Observable.Interval(TimeSpan.FromSeconds(1)).Select(x => Tuple.create(SomeKeyThatVaries, x)

var obs2 = Observable.Interval(TimeSpan.FromMilliSeconds(1)).Select(x => Tuple.create(SomeKeyThatVaries, x)

from x in obs1
  let latestFromObs2WhereKeyMatches = …
  select Tuple.create(x, latestFromObs2WhereKeyMatches)

有什么建议吗？

显然，这可以通过订阅第二个 observable 并创建一个字典来实现，该字典具有可通过键索引的最新值。但我正在寻找一种不同的方法..

使用场景：从股票报价流计算的一分钟价格柱。在这种情况下，关键是代码，字典包含具体代码的最新卖价和买价，然后用于计算。

（顺便说一句，谢谢 Dave 和 James，这是一次非常富有成果的讨论）

（抱歉格式化，在 iPad 上很难弄好..）

【问题讨论】：

如果您只想在“x”可观察对象有值时在结果可观察对象中生成通知，那么“y”可观察对象的意义何在？你的问题我不清楚。也许包括一个大理石图会有所帮助。
代码示例只是为了提示我正在尝试做的事情..我显然没有成功..
是的，猜测的代码示例通常是个坏主意。具体的大理石图会更好。
好的，我会阅读大理石图.. 同时我重写了问题..
您需要更具体。它不一定必须是图表的形式，但您至少必须指定您期望的顺序。一个具体的例子会有很大帮助 - 不是查询，而是示例输入和预期输出。

标签： c# system.reactive

【解决方案1】：

...您为什么要寻找不同的方法？听起来你对我来说是正确的。它很短，简单的代码......粗略地说，它会是这样的：

var cache = new ConcurrentDictionary<long, long>();    
obs2.Subscribe(x => cache[x.Item1] = x.Item2);    
var results = obs1.Select(x => new {
    obs1 = x.Item2,
    cache.ContainsKey(x.Item1) ? cache[x.Item1] : 0
});

归根结底，C# 是一种 OO 语言，线程安全可变集合的繁重工作已经为您完成。

可能会有花哨的 Rx 方法（感觉可能涉及连接）...但是它的可维护性如何？它的表现如何？

0.02 美元

【讨论】：

基本上我同意你的观点，但我试图在这里扩大限制。如果我或没有其他人想出一个花哨的 ;) 方法，我会将其标记为答案..
这种答案真的不像你;-)我同意你的方法是最直接的；但是，似乎查询解决方案并不是那么复杂 - 请参阅我的答案。我想他想要一个Merge-GroupBy-Scan。我认为它实际上是 2 个单独的查询：Merge-GroupBy 是第一个，将密钥移开，然后 Scan 就像 CombineLatest 一样过滤来自一个的通知一边。
@DaveSexton 我明白你为什么说它不像我——我经常被 Rx 冲昏头脑；）——但是，not always！在这里，这似乎是一个务实的解决方案。有了更多的上下文，可能会有一个更像 FRP 的解决方案的案例——但我觉得在这种情况下，它绝对是最简单的方法。尽管如此，我还是不得不为你的回答 +1，因为它肯定更像我 - 看到它以 Rx 的方式完成总是很有趣！
我同意你的博文，async/await 确实取代了 cardinality=1 的 observables。我现在很少使用只推送一个值的 observables。但对于其他一切，我真的没有更好的选择。这就是我划清界限的地方。当基数 > 1 时敲除。

【解决方案2】：

我想知道这样一个查询的目的。您介意描述一下使用场景吗？

不过，以下查询似乎可以解决您的问题。如果您已经有某种方法可以识别每个值的来源，则不需要初始预测，但是为了概括，我将它们包括在内，以与您极其抽象的提问模式保持一致。 ;-)

注意：我假设someKeyThatVaries 不是您所展示的共享数据，这就是为什么我还包括术语anotherKeyThatVaries；否则，整个查询对我来说真的没有意义。

var obs1 = Observable.Interval(TimeSpan.FromSeconds(1))
                     .Select(x => Tuple.Create(someKeyThatVaries, x));
var obs2 = Observable.Interval(TimeSpan.FromSeconds(.25))
                     .Select(x => Tuple.Create(anotherKeyThatVaries, x));

var results = obs1.Select(t => new { Key = t.Item1, Value = t.Item2, Kind = 1 })
                  .Merge(
              obs2.Select(t => new { Key = t.Item1, Value = t.Item2, Kind = 2 }))
                  .GroupBy(t => t.Key, t => new { t.Value, t.Kind })
                  .SelectMany(g =>
                    g.Scan(
                      new { X = -1L, Y = -1L, Yield = false },
                      (acc, cur) => cur.Kind == 1
                                  ? new { X = cur.Value, Y = acc.Y, Yield = true }
                                  : new { X = acc.X, Y = cur.Value, Yield = false })
                      .Where(s => s.Yield)
                      .Select(s => Tuple.Create(s.X, s.Y)));

【讨论】：