使用 Reactive Extensions 逐组缓冲，嵌套订阅答案

【问题标题】：Buffer group by groups with Reactive Extensions, nested subscribe使用 Reactive Extensions 逐组缓冲，嵌套订阅
【发布时间】：2017-02-28 16:24:52
【问题描述】：

我有一个事件源，它生成属于某些组的事件。我想缓冲这些组并将这些组（分批）发送到存储。到目前为止，我有这个：

eventSource
    .GroupBy(event => event.GroupingKey)
    .Select(group => new { group.Key, Events = group })
    .Subscribe(group => group.Events
                            .Buffer(TimeSpan.FromSeconds(60), 100)
                            .Subscribe(list => SendToStorage(list)));

所以有一个对组中事件的嵌套订阅。不知何故，我认为有更好的方法，但我还没有弄清楚。

【问题讨论】：

标签： c# .net system.reactive reactive

【解决方案1】：

解决办法如下：

eventSource
    .GroupBy(e => e.GroupingKey)
    .SelectMany(group => group.Buffer(TimeSpan.FromSeconds(60), 100))
    .Subscribe(list => SendToStorage(list));

以下是一些可以帮助您“减少”的一般规则：

1) 嵌套订阅通常在嵌套订阅之前使用Selecting 固定所有内容，然后是Merge，然后是嵌套订阅。所以应用它，你会得到这个：

eventSource
    .GroupBy(e => e.GroupingKey)
    .Select(group => new { group.Key, Events = group })
    .Select(group => group.Events.Buffer(TimeSpan.FromSeconds(60), 100)) //outer subscription selector
    .Merge()
    .Subscribe(list => SendToStorage(list));

2）您显然可以组合两个连续的选择（并且由于您没有对匿名对象做任何事情，因此可以将其删除）：

eventSource
    .GroupBy(e => e.GroupingKey)
    .Select(group => group.Buffer(TimeSpan.FromSeconds(60), 100)) 
    .Merge()
    .Subscribe(list => SendToStorage(list));

3) 最后，Select 后跟 Merge 可以简化为 SelectMany：

eventSource
    .GroupBy(e => e.GroupingKey)
    .SelectMany(group => group.Buffer(TimeSpan.FromSeconds(60), 100))
    .Subscribe(list => SendToStorage(list));

【讨论】：

编写了一些单元测试来验证它是否可以继续工作并且像一个魅力一样工作。谢谢！

【解决方案2】：

这是一种方法

(from g in eventSource.GroupByUntil(e => e.GroupingKey,
                                    g => g.Buffer(TimeSpan.FromSeconds(60), 100))
 from b in g.ToList()
 select b).Subscribe(SendToStorage);

【讨论】：