c# - 将 Rx 聚合保存到局部变量

Question

我正在尝试收集有关 IObservable 流的统计信息延迟。在我的例子中，这些流来自 ETW 跟踪文件的 Tx 驱动程序。

我已经编写了每秒给我这些统计数据的窗口聚合，我在处理“完整跟踪”统计数据时遇到了问题，我希望在流接收 OnComplete 时返回一些内容。下面是说明“完整跟踪”统计查询问题的代码。

流有两种很容易区分的延迟类型，>10 是一种类型的延迟，<10 是另一种类型。

struct Duration
{
    DateTime OccurenceTime;
    double Latency;
}

IObservable<Duration> observableDurations = ...

var statistics = 
    from duration in observableDurations
    group duration by duration.Latency > 10 into g
    select new
    {
        Type = g.Key ? "Latency Type A" : "Latency Type B",
        Min  = g.Min(o => o.Latency),
        Max  = g.Max(o => o.Latency),
        Avg  = g.Average(o => o.Latency),
    };
// statistics.Dump(); In LinqPad this shows correct results.

double maxLatencyTypeA = 0;
statistics.Subscribe(x =>
{
    if (x.Type == "Latency Type A")
    {
        maxLatencyTypeA = x.Max;
        // I would also save min and average into local variables
    }
    else
    {
        // I would also save the statistics for TypeB latencies into local variables
    }
});

这不会编译，因为在统计查询的定义中，g.Min、g.Max 和 g.Average 是 IObservable<double> ，但在我传递给 statistics.Suscribe() 的 lambda 表达式中我尝试将 g.Max 保存到 double这显然与 IObservable<double> 不兼容.

我可以通过更改 maxLatencyTypeA = x.Max; 在 lambda 中解决这个问题至 x.Max.Subscribe(s=>maxLatencyTypeA=s); .我的问题是，有没有一种方法可以编写更好的统计查询来生成 double 类型的查询？而不是 IObservable<double>对于这样的统计信息，我不需要为我可能想要保存的每个统计信息在统计信息订阅中嵌套更多订阅？

谢谢

Answer 1

不要使用单独的运算符来单独聚合，而是使用 Aggregate 并一次收集所有聚合。这样你就可以拥有一个包含所有结果的可观察对象。请注意，由于 float 学，我在这里计算平均值的算法会有一点漂移。如果这是一个问题，那么您还可以维护一个运行 Sum 并将其与 Count 一起使用来计算平均值。

var statistics = observableDurations
    .GroupBy(d => d.Latency > 10)
    .Select(g =>
    {
        var seed = new
        {
            Type = g.Key ? "A" : "B",
            Min = Double.MaxValue,
            Max = Double.MinValue,
            Average = 0.0,
            Count = 0
        };
        return g.Aggregate(seed, (total, duration) => new
        {
            Type = total.Type,
            Min = Math.Min(total.Min, duration.Latency),
            Max = Math.Max(total.Max, duration.Latency),
            Average = (total.Average * total.Count + duration.Latency) / (total.Count + 1),
            Count = total.Count + 1
        });
    });