如何定期合并 observables？

Question

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

有没有更好的方法？

我尝试过 Sample()，但传感器以不规则的时间间隔产生值，慢（> 1 秒）和快（

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

Answer 2

如何使用CombineLatest() 运算符在每次产生一个值时合并传感器的最新值，然后使用Sample() 以确保每秒一次测量的最大频率？

sensorA.CombineLatest(sensorB, (a, b) => new {A=a, B=b}).Sample(TimeSpan.FromSeconds(1))