.net - Reactive Extensions 中的 BufferWithTime 会重叠调用 OnNext 吗？

Question

我正在使用 Rx 的 BufferWithTime() 来批量处理消息。如果我的 OnNext 方法的完成时间比所使用的时间间隔长，我是否可以同时调用两个 OnNext 方法？

换句话说，在 BufferWithTime() 调用中指定的时间间隔会以绝对值计算，还是会在我的 OnNext 方法返回后充当“异步 sleep ”？

Answer 1

不，请调用 OnNext将排队。下面的代码在缓冲时间过去时写入跟踪，以及对订阅者的 OnNext 调用的开始/结束(其中它休眠了缓冲时间的 4 倍):

Observable.Interval(TimeSpan.FromMilliseconds(50), Scheduler.TaskPool)
    .Take(50)
    .BufferWithTime(TimeSpan.FromMilliseconds(100))
    .Do(_ => Trace.WriteLine("Buffer elapsed"))
    .ObserveOn(Scheduler.TaskPool)
    .Subscribe(_ =>
    {
        Trace.WriteLine("Begin OnNext");
        Thread.Sleep(200);
        Trace.WriteLine("End OnNext");
    });

输出如下。您可以看到 Begin/End OnNext 永远不会重叠，即使是 Buffer elapsed在 OnNext 期间出现两次称呼:

Buffer elapsed
Begin OnNext
Buffer elapsed
Buffer elapsed
End OnNext
Begin OnNext
Buffer elapsed
Buffer elapsed
Buffer elapsed
End OnNext
Begin OnNext
End OnNext
Begin OnNext
End OnNext
Begin OnNext
End OnNext
Begin OnNext
End OnNext