无扩展功能：批量处理事件+每批次之间增加延迟 [英] Reactive Extensions: Process events in batches + add delay between every batch

问题描述

我有一个在某些点上引发1000事件几乎在同一时间的应用程序。我想这样做是为了批事件的50项块，并开始处理它们每10秒。有没有必要等一批启动一批新的处理之前完成

I have an application which at some points raises 1000 events almost at the same time. What I would like to do is to batch the events to chunks of 50 items and start processing them every 10 seconds. There's no need to wait for a batch to complete before starting a new batch processing.

例如：

10:00:00: 10000 new events received
10:00:00: StartProcessing (events.Take(50))
10:00:10: StartProcessing (events.Skip(50).Take(50))
10:00:15: StartProcessing (events.Skip(100).Take(50))

任何想法如何实现这一目标？我想反应的扩展是要走的路，但其他的解决方案是可以接受的太

Any ideas how to achieve this? I suppose Reactive Extensions is the way to go but other solutions are acceptable too.

我试图从这里开始：

        var bufferedItems = eventAsObservable
            .Buffer(15)
            .Delay(TimeSpan.FromSeconds(5)

但注意到，像我希望的，而是所有批次同时启动，虽然5秒拖延迟迟没有工作。

But noticed that the delay didn't work as I hoped for and instead all the batches started simultaneously, though 5 seconds delayed.

我还测试了窗口，方法，但我没有发现任何行为上的差异，我想在窗口的时间跨度实际上意味着采取一切该发生在接下来的10秒事件：

I also tested the Window-method, but I didn't notice any difference in behavior. I suppose the TimeSpan in Window actually means that "take every event which happens in the next 10 seconds:

        var bufferedItems = eventAsObservable
            .Window(TimeSpan.FromSeconds(10), 5)
            .SelectMany(x => x)
            .Subscribe(DoProcessing);

我使用的接收。 - 主要2.0.20304-β

I'm using the Rx-Main 2.0.20304-beta.

推荐答案

如果您不想睡觉线程，你可以这样做：

If you'd prefer not to sleep threads, you can do this:

var tick = Observable.Interval(TimeSpan.FromSeconds(5));

eventAsObservable
.Buffer(50)
.Zip(tick, (res, _) => res)
.Subscribe(DoProcessing);

这篇关于无扩展功能：批量处理事件+每批次之间增加延迟的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！