c# - 如何定期合并可观察对象？

Question

我试图定期合并两个传感器数据流，但我在 Rx 中无法正确执行此操作。我想出的最好的是下面的示例，但是我怀疑这是 Rx 的最佳使用。

有没有更好的方法？

我已经尝试过 Sample()，但传感器以不规则的间隔生成值，慢(>1 秒)和快(<1 秒)。 Sample() 似乎只处理快速数据。

Observable<SensorA> sensorA = ... /* hot */
Observable<SensorB> sensorB = ... /* hot */

SensorA lastKnownSensorA;
SensorB lastKnownSensorB;

sensorA.Subscribe(s => lastKnownSensorA = s);
sensorB.Subscribe(s => lastKnownSensorB = s);

var combined = Observable.Interval(TimeSpan.FromSeconds(1))
    .Where(t => _lastKnownSensorA != null)
    .Select(t => new SensorAB(lastKnownSensorA, lastKnownSensorB)

Answer 1

我认为@JonasChapuis 的答案可能就是您所追求的，但有几个问题可能会出现问题:

• CombineLatest在所有源都至少发出一个值之前不会发出值，这可能会导致更快的源中的数据丢失，直到这一点为止。这可以通过使用 StartWith 来缓解播种 null object或每个传感器流上的默认值。

• Sample如果在采样期间没有观察到新值，则不会发出值。我无法从问题中判断这是否可取，但如果不是，则有一个有趣的技巧可以使用“速度”流来解决这个问题，如下所述以创建固定频率，而不是使用 Sample 获得的最大频率.

地址CombineLatest问题，为您的传感器流确定适当的空值 - 我通常通过静态 Null 使这些可用类型上的属性 - 这使得意图非常明确。对于值类型，使用 Nullable<T>也是一个不错的选择:

Observable<SensorA> sensorA = ...  .StartWith(SensorA.Null);
Observable<SensorB> sensorB = ...  .StartWith(SensorB.Null);

注意不要犯申请 StartWith 的常见错误仅输出 CombinedLatest ……那没用!

现在，如果您需要定期结果(自然可以包括最近读数的重复)，请创建一个以所需间隔发出的“速度”流:

var pace = Observable.Interval(TimeSpan.FromSeconds(1));

然后组合如下，从结果中省略步速值:

var sensorReadings = Observable.CombineLatest(
    pace, sensorA, sensorB,
    (_, a, b) => new SensorAB(a,b));

MostRecent 也值得了解可以与 Zip 组合的运算符如果您想以特定流的速度驱动输出，则非常有效。在我演示该方法的地方查看这些答案:How to combine a slow moving observable with the most recent value of a fast moving observable以及处理多个流的更有趣的调整:How do I combine three observables such that