在尝试数据流之后,我遇到了一个新问题。我想限制所有块的输入队列。我的生产块(ActionBlock)正在非常快速地创建5000个元素,并将它们发布到广播块中。因此,如果我将广播块的有限容量设置为 100,他会丢弃大量数据。但我更喜欢生产块在我的缓冲区块的输入队列中等待新插槽。
有什么办法可以解决这个问题吗?
这正是BufferBlock
的目的。如果您设置了它BoundedCapacity
并且它已满,它将推迟接收任何消息,直到有人使用它们。这意味着例如Post()
将阻塞,SendAsync()
将返回未完成的Task
。
编辑:没有内置块可以发送到多个目标并且永远不会丢弃数据。但是您可以轻松地从ActionBlock
和发送循环中构建一个:
static ITargetBlock<T> CreateMultipleTargetsBlock<T>(
IEnumerable<ITargetBlock<T>> targets, int boundedCapacity)
{
var targetsList = targets.ToList();
var block = new ActionBlock<T>(
async item =>
{
foreach (var target in targetsList)
{
await target.SendAsync(item);
}
},
new ExecutionDataflowBlockOptions { BoundedCapacity = boundedCapacity });
// TODO: propagate completion from block to targets
return block;
}
此代码假定您不需要克隆每个目标的数据,并且目标列表永远不会更改。修改代码应该相当简单。