10

在我使用 .Net 4.6 的 WPF 应用程序中，我有一个以高速率（每秒数百个）触发新数据点的事件，但并非一直如此。此数据显示在图表中。

我想每 50 毫秒更新一次图表，而不是在每个新数据点之后。
为了实现这一点，我Buffer(TimeSpan.FromMilliseconds(50))从 Rx 中使用，理论上效果很好。但是，如果没有创建新的数据点，我的订阅者也会每 50 毫秒被调用一次，这不是我想要的。

我创建了一个小示例应用程序来测试它：

using System;
using System.Reactive.Linq;

namespace RxTester
{
    public class Program
    {
        private static event EventHandler TheEvent;

        static void Main(string[] args)
        {
            var observable = Observable.FromEvent<EventHandler, EventArgs>(h => (s, e) => h(e), h => TheEvent += h, h => TheEvent -= h);
            var subscriber = observable.Buffer(TimeSpan.FromMilliseconds(1000))
                .Subscribe(e => Console.WriteLine($"{DateTime.Now.ToLongTimeString()}: {e.Count} elements received..."));

            var random = new Random();
            var timer = new System.Timers.Timer(2000)
                {
                    AutoReset = true,
                    Enabled = true
                };
            timer.Elapsed += (s, e) =>
                {
                    var amount = random.Next(1, 10);
                    for (int i = 0; i < amount; ++i)
                        TheEvent?.Invoke(null, null);
                };

            Console.ReadLine();

            timer.Enabled = false;
            subscriber.Dispose();
        }
    }
}

您需要添加“Rx-Linq”NuGet 包以使其运行或使用以下小提琴：https ://dotnetfiddle.net/TV5tD4

在那里，您会看到几个“收到的 0 个元素”，这是我想避免的。我知道我可以简单地检查e.Count == 0，但是当我使用多个这样的缓冲区时，这对我来说似乎不是最优的。

如果元素可用，有没有办法只创建新的缓冲元素块？
我也愿意使用其他方法来解决我的按时间批处理事件的问题 - 我已经研究过 TPL Dataflows BatchBlock，但这似乎只支持基于计数的块大小。

4

2 回答 2

4

我们可以再次使用强大的GroupByUntil方法来创建这个扩展

public static IObservable<IList<TSource>> BufferWhenAvailable<TSource>
                                          (this IObservable<TSource> source, 
                                           TimeSpan threshold)
{
    return source.Publish( sp => 
                    sp.GroupByUntil(_ => true, _ => Observable.Timer(threshold))
                      .SelectMany(i => i.ToList()));

}
于 2016-02-24T19:22:36.763 回答
3

这样做的标准方法很简单

.Buffer(period)
.Where(buffer=>buffer.Any())

如此有效地做你想要避免的事情(count==0)。然而，这张支票非常便宜，我想如果比其他成本（即调度）便宜得多。唯一关心的可能是正在发生的分配量（每 50 毫秒创建一个List<T>），然后可能会产生即将到来的 GC Gen0 压力。

于 2016-02-25T01:04:04.133 回答