c# - 如何在 Reactive Extensions 中将项目缓冲到组中？

Question

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

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

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

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

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

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

score 8 · Accepted Answer

干得好：

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

score 2 · Accepted Answer

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

2 回答 2