6

我有一堆事件进来,我必须毫无损失地执行所有这些事件,但我想确保它们在适当的时间段被缓冲和消耗。有人有解决方案吗?

我在 Rx 中找不到任何可以在不丢失事件的情况下做到这一点的操作员(Throttle - 丢失事件)。我也考虑过缓冲、延迟等...找不到好的解决方案。

我试图在中间放一个计时器,但不知何故它根本不起作用:

GetInitSequence()
            .IntervalThrottle(TimeSpan.FromSeconds(5))
            .Subscribe(
                item =>
                    {
                        Console.WriteLine(DateTime.Now);
                        // Process item
                    }
            );

public static IObservable<T> IntervalThrottle<T>(this IObservable<T> source, TimeSpan dueTime)
    {
        return Observable.Create<T>(o =>
            {
                return source.Subscribe(x =>
                    {
                        new Timer(state => 
                            o.OnNext((T)state), x, dueTime, TimeSpan.FromMilliseconds(-1));
                    }, o.OnError, o.OnCompleted);
        });
    }
4

4 回答 4

13

这个问题不是 100% 清楚的,所以我在做一些假设。

Observable.Delay不是您想要的,因为这会在每个事件到达时产生延迟,而不是为处理创建均匀的时间间隔。

Observable.Buffer不是您想要的,因为这将导致每个给定时间间隔内的所有事件都传递给您,而不是一次一个。

因此,我相信您正在寻找一种解决方案,该解决方案可以创建某种节拍器,并在每个节拍时为您提供一个事件。这可以使用节拍器简单地构建并将连接到您的源:Observable.IntervalZip

var source = GetInitSequence();
var trigger = Observable.Interval(TimeSpan.FromSeconds(5));    
var triggeredSource = source.Zip(trigger, (s,_) => s); 
triggeredSource.Subscribe(item => Console.WriteLine(DateTime.Now));

这将每 5 秒触发一次(在上面的示例中),并按顺序为您提供原始项目。

此解决方案的唯一问题是,如果您在(例如)10 秒内没有更多源元素,则当源元素到达时,它们将立即被发送出去,因为一些“触发”事件正坐在那里等待它们. 该场景的大理石图:

source:  -a-b-c----------------------d-e-f-g
trigger: ----o----o----o----o----o----o----o
result:  ----a----b----c-------------d-e-f-g

这是一个非常合理的问题。这里已经有两个问题可以解决:

Rx IObservable 缓冲以消除突发事件

一种以均匀间隔推送缓冲事件的方法

提供的解决方案是主要Drain扩展方法和次要Buffered扩展。我已将这些修改得更简单(不需要Drain,只需使用Concat)。用法是:

var bufferedSource = source.StepInterval(TimeSpan.FromSeconds(5));

扩展方法StepInterval

public static IObservable<T> StepInterval<T>(this IObservable<T> source, TimeSpan minDelay)
{
    return source.Select(x => 
        Observable.Empty<T>()
            .Delay(minDelay)
            .StartWith(x)
    ).Concat();
}
于 2012-07-01T21:55:07.900 回答
1

我知道这可能太简单了,但这行得通吗?

var intervaled = source.Do(x => { Thread.Sleep(100); });

基本上,这只是在值之间设置了最小延迟。太简单了?

于 2012-07-02T01:23:31.857 回答
1

按照 Enigmativity 的回答,如果您只想将所有值延迟一个 TimeSpan,我不明白为什么Delay不是您想要的运算符

  GetInitSequence()
        .Delay(TimeSpan.FromSeconds(5)) //ideally pass an IScheduler here
        .Subscribe(
            item =>
                {
                    Console.WriteLine(DateTime.Now);
                    // Process item
                }
        );
于 2013-06-24T08:14:40.737 回答
0

Observable.Buffer怎么样?这应该将 1s 窗口中的所有事件作为单个事件返回。

var xs = Observable.Interval(TimeSpan.FromMilliseconds(100));
var bufferdStream = xs.Buffer(TimeSpan.FromSeconds(5));
bufferdStream.Subscribe(item => { Console.WriteLine("Number of events in window: {0}", item.Count); });

这可能是你问的不是很清楚。你的代码应该做什么?看起来你只是通过为每个事件创建一个计时器来延迟。它还破坏了 observable 的语义,因为下一个和完成可能发生在下一个之前。

请注意,这也仅在使用的计时器上同样准确。通常,计时器最多可精确到 16 毫秒。

编辑:

您的示例变为,并且 item 包含窗口中的所有事件:

GetInitSequence()
            .Buffer(TimeSpan.FromSeconds(5))
            .Subscribe(
                item =>
                    {
                        Console.WriteLine(DateTime.Now);
                        // Process item
                    }
            );
于 2012-07-01T18:03:44.667 回答