IObservable序列完成前的聚合函数

Question

有没有办法在序列完成之前将聚合函数（Max、Count、....）与 Buffer 一起使用。完成后，这将产生结果，但随着流的继续，它不会给出有结果吗？

我期待有一些方法可以使缓冲区工作？

IObservable<long> source;
IObservable<IGroupedObservable<long, long>> group  = source
        .Buffer(TimeSpan.FromSeconds(5))
        .GroupBy(i => i % 3);

IObservable<long> sub = group.SelectMany(grp => grp.Max());

sub.Subscribe(l =>
{
    Console.WriteLine("working");
});

Answer 1

使用 Scan 而不是 Aggregate。 Scan 的工作方式与 Aggregate 类似，只是它在流前进时发送中间值。这对 "running totals" 很有用，这似乎正是您所要求的。

Answer 2

Rx (Min/Max/Sum/Count/Average) 中的所有 "statistical" 运算符都使用一种机制，即在订阅完成时传播计算值，这是 Scan 之间的最大区别和 Aggregate，基本上，如果您想在订阅中推送新值时收到通知，则有必要使用 Scan.

在你的情况下，如果你想保持相同的逻辑，你应该结合GroupByUntil或Window运算符，同时使用的条件可以定期创建和完成组订阅，并且将用于推送下一个值。

您可以在此处获取更多信息：http://www.introtorx.com/content/v1.0.10621.0/07_Aggregation.html#BuildYourOwn

顺便写了一篇与你想要的相关的文字。签到：http://www.codeproject.com/Tips/853256/Real-time-statistics-with-Rx-Statistical-Demo-App

IObservable序列完成前的聚合函数

Aggregate function before IObservable sequence is completed

c#

system.reactive