RX - Group/Batch 个可观察序列中的元素突发

Question

我有一个可观察的序列。当插入第一个元素时，我想启动一个计时器并在计时器的时间跨度内批量处理后续插入的元素。然后，在序列中插入另一个元素之前，计时器不会再次启动。

所以像这样：

--------|=====timespan====|---------------|=====timespan====|-------------->
        1  2 3 4    5                     6 7            8

会产生：

[1,2,3,4,5], [6,7,8]

我尝试使用 Observable.Buffer() 和一个时间跨度，但从我的实验中，我可以看到一旦我们订阅可观察序列，定时器就会启动，并在前一个定时器一到就重新启动完成。

所以使用与前面示例相同的序列并使用具有时间跨度的 Buffer()，我会得到这样的结果：

|=====timespan====|=====timespan====|=====timespan====|=====timespan====|-->
        1  2 3 4    5                      6 7           8

这会产生这个：

[1,2,3,4], [5], [6,7], [8]

以下是我如何使用缓冲区测试此行为：

var source = Observable.Concat(Observable.Timer(TimeSpan.FromSeconds(6)).Select(o => 1),
                               Observable.Timer(TimeSpan.FromSeconds(1)).Select(o => 2),
                               Observable.Timer(TimeSpan.FromSeconds(3)).Select(o => 3),
                               Observable.Never<int>());

Console.WriteLine("{0} => Started", DateTime.Now);
source.Buffer(TimeSpan.FromSeconds(4))
      .Subscribe(i => Console.WriteLine("{0} => [{1}]", DateTime.Now, string.Join(",", i)));

输出：

4/24/2015 7:01:09 PM => Started
4/24/2015 7:01:13 PM => []
4/24/2015 7:01:17 PM => [1,2]
4/24/2015 7:01:21 PM => [3]
4/24/2015 7:01:25 PM => []
4/24/2015 7:01:29 PM => []
4/24/2015 7:01:33 PM => []

有人知道如何做到这一点吗？提前致谢！

Answer 1

试一试：

var source = Observable.Concat(Observable.Timer(TimeSpan.FromSeconds(6)).Select(o => 1),
                           Observable.Timer(TimeSpan.FromSeconds(1)).Select(o => 2),
                           Observable.Timer(TimeSpan.FromSeconds(4)).Select(o => 3),
                           Observable.Never<int>());

Console.WriteLine("{0} => Started", DateTime.Now);
source
    .GroupByUntil(x => 1, g => Observable.Timer(TimeSpan.FromSeconds(4)))
    .Select(x => x.ToArray())
    .Switch()
    .Subscribe(i => Console.WriteLine("{0} => [{1}]", DateTime.Now, string.Join(",", i)));

我不得不更改第三个计时器的测试代码持续时间，以确保该值在分组计时器之外。

RX - Group/Batch 个可观察序列中的元素突发

RX - Group/Batch bursts of elements in an observable sequence

c#

system.reactive

observable