如何在 Reactive Extensions 中将项目缓冲到组中？答案

【问题标题】：How do you buffer items into groups in Reactive Extensions?如何在 Reactive Extensions 中将项目缓冲到组中？
【发布时间】：2011-10-21 06:28:18
【问题描述】：

我有一个 IObservable；其中属性更改具有实体 ID 和 PropertyName。我想用它来更新数据库，但如果多个属性几乎同时更改，我只想对同一实体的所有属性进行一次更新。

如果这是一个静态 IEnumerable 并且我使用的是 LINQ，我可以简单地使用：

MyList.GroupBy(C=>C.EntityID);

但是，列表永远不会终止（永远不会调用 IObserver.OnComplete）。我想要做的是等待一段时间，比如 1 秒，然后将所有呼叫适当地分组在那一秒。

理想情况下，我会为每个 EntityID 设置单独的计数器，并且只要发现该 EntityID 的新属性更改，它们就会重置。

我不能使用 Throttle 之类的东西，因为我想处理所有属性更改，我只想一次性处理它们。

【问题讨论】：

您的问题的标题和文本似乎完全无关。你能解决这个问题吗？
已修复，虽然我也很想知道 GroupByUntil 做了什么以及它是如何工作的，因为它看起来可能会做我想要的，但当我尝试使用它进行测试时似乎并没有做任何事情不同的输入。

【解决方案1】：

给你：

MyObservable
    .Buffer(TimeSpan.FromSeconds(1.0))
    .Select(MyList =>
        MyList.GroupBy(C=>C.EntityID));

【讨论】：

【解决方案2】：

Buffer 方法似乎可以满足您的需求。给它 TimeSpan，它会将所有消息折叠到一个列表中。还有 Window 方法做类似的事情，但我不完全确定它的语义可能是什么。

【讨论】：