具有读写批量数据能力的生产者消费者集合答案

【问题标题】：Producer consumer collection with ability to read and write batches of data具有读写批量数据能力的生产者消费者集合
【发布时间】：2018-09-13 13:02:58
【问题描述】：

我正在寻找像BufferBlock这样的集合

但使用如下方法：

SendAsync<T>(T[])
T[] ReceiveAsync<T>()

有人可以帮忙吗？

【问题讨论】：

你检查扩展方法了吗？
@Hackerman，是的......双重检查；）如果我遗漏了什么，你能指出我吗？
为什么不用BufferBlock？顺便说一句，它确实有这样的方法。检查How to: Implement a Producer-Consumer Dataflow Pattern
至于 other 除了 Bufferblock 明确支持 pub/sub，你可以试试新的 System.Threading.Tasks.Channels 包
添加了另一个选项，BatchBlock 可能更适合您，但这取决于您的用例。

标签： c# multithreading task-parallel-library producer-consumer

【解决方案1】：

这些方法不可用，SendAsync<T> 只接受一个 T 而RecieveAsync<T> 只返回一个 T，而不是数组。

SendAsync<T>(T[])
T[] ReceiveAsync<T>()

但是有TryReceiveAll<T>(out IList<T> items)，您可以在循环中调用SendAsync<T> 将数组发送到BufferBlock 或编写您自己的扩展方法，如下所示：

public static async Task SendAllAsync<T>(this ITargetBlock<T> block, IEnumerable<T> items)
{
    foreach(var item in items)
    {
         await block.SendAsync(item)
    }
}

请注意，SendAsync 确实返回一个表示接受消息的布尔值，您可以返回一个布尔值数组，或者在其中任何一个返回 false 时返回，但这取决于您。

使用BatchBlock<T> 可能会更容易，您可以使用循环将项目作为单项发送，但分批发出项目，如果您正在构建管道，这将比使用TryRecieveAll 更容易。 BatchBlock Walkthrough 和 BatchBlock Example

【讨论】：

感谢 BatchBlock，它与我想要的很接近。理想情况下，如果集合可以异步处理批处理输入，并且可以异步读取只有最大大小的批处理，而无需外部计时器

【解决方案2】：

ReceiveAsync 和 SendAsync 可用作 ISourceBlock 和 ITargetBlockT 接口上的扩展方法。这意味着您必须将块转换为这些接口才能使用扩展方法，例如：

var buffer=new BufferBlock<string>();

var source=(ISourceBlock<string>)buffer;
var target=(ITargetBlock<string>)buffer;

await target.SendAsync("something");

这通常不是问题，因为所有 Dataflow 方法都接受接口，而不是具体类型，例如：

async Task MyProducer(ITargetBlock<string> target)
{
    ...
    await target.SendAsync(..);
    ...
    target.Complete();
}

async Task MyConsumer(ISourceBlock<string> target)
{
    ...
    var message=await target.ReceiveAsync();
    ...
}

public static async Task Main()
{
    var buffer=new BufferBlock<string>();
    MyProducer(buffer);
    await MyConsumer(buffer);
}

【讨论】：

感谢您的关注和努力。但是我关于读取和写入批量元素的问题。我知道我的问题中通过链接的文章中描述的单元素发送/接收这种方法。
@gabba 那么你应该使用 BatchBlock，而不是 BufferBlock
谢谢。将按照 JSteward 的建议将 BatchBlock 与 BufferBlock 一起使用