BlockingCollection 替代方案？

Question

我们正在使用 BlockingCollection 在实时应用程序中实现生产者-消费者模式：

BlockingCollection<T> collection = new BlockingCollection<T>();
CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();

// Starting up consumer
Task.Run(() => consumer(this.cancellationTokenSource.Token));
…
void Producer(T item)
{
    collection.Add(item);
}
…
void consumer()
{
    while (true)
    {
           var item = this.blockingCollection.Take(token);
           process (item);
    }
}

可以肯定的是，这是实际生产代码的非常简化的版本。 有时，当应用程序负载较重时，我们会观察到消费部分落后于生产部分。由于应用程序逻辑非常复杂，它涉及通过网络与其他应用程序以及SQL数据库的交互。许多地方都可能出现延误；它们可能发生在对 process() 的调用中，这原则上可以解释为什么消费部分可能很慢。

除了上述所有考虑因素之外，使用 BlockingCollection 是否有一些固有的东西可以解释这种现象？ .Net中是否有更有效的选择来实现生产者-消费者模式？

Answer 1

首先，BockingCollection不是生产者/消费者场景的最佳选择。至少有两个更好的选项（数据流，频道），选择取决于实际应用程序方案 - 问题所缺少。

也可以使用 async流和iasyncenmerable。

异步流

在这种情况下，生产者可以是异步迭代器。消费者将收到iAsyncencencenterable，并在其完成之前进行迭代。它还可以产生自己的IASYNCENCENUMETARE输出，可以将其传递给管道中的下一个方法：

生产者可以是：

public static async IAsyncEnumerable<Message> ProducerAsync(CancellationToken token)
{
    while(!token.IsCancellationRequested)
    {
        var msg=await Task.Run(()=>SomeHeavyWork());
        yield return msg;
    }
}

和消费者：

async Task ConsumeAsync(IAsyncEnumerable<Message> source)
{
    await foreach(var msg in source)
    {
        await consumeMessage(msg);
    }
}

在这种情况下，没有缓冲，并且生产者在消费者之前不能发出新的消息消耗当前的。消费者可以通过并行处理。最后， system.linq.async 为我们提供linq操作编写例如：

    List<OtherMsg> results=await ProducerAsync(cts.Token)
                                 .Select(msg=>consumeAndReturn(msg))
                                 .ToListAsync();

dataFlow-action block

dataFlow 可以使用块来构建整个处理管道，每个块从上一个接收消息（数据），对其进行处理并将其传递到下一个块。大多数块具有输入，并且在适当的输出缓冲区。每个块都使用单个工作任务，但可以配置为更多。但是，应用程序代码不必处理任务。

在最简单的情况下，单个 action block 可以处理一个或多个生产者发布给它的消息，用作消费者：

async Task ConsumeAsync<Message>(Message message)
{
    //Do something with the message
}

...
ExecutionDataflowBlockOptions _options= new () {
    MaxDegreeOfParallelism=4,
    BoundedCapacity=5
};

ActionBlock<Message> _block=new ActionBlock(ConsumeAsync,_options);

async Task ProduceAsync(CancellationToken token)
{
    while(!token.IsCancellationRequested)
    {
        var msg=await produceNewMessageAsync();
        await _block.SendAsync(msg);
    }
   _block.Complete();
   await _block.Completion;
}

在此示例中除了目前正在处理的人之外。

BufferBlock作为生产者/消费者队列

BufferBlock是一个不活动块，该块由其他块用作缓冲区。它可以用作异步生产者/消费者集合，如如何：实现生产者 - 消费者数据流模式。在这种情况下，代码需要明确接收消息。线程取决于开发人员。 ：

static void Produce(ITargetBlock<byte[]> target)
{
    var rand = new Random();

    for (int i = 0; i < 100; ++ i)
    {
        var buffer = new byte[1024];
        rand.NextBytes(buffer);
        target.Post(buffer);
    }

    target.Complete();
}

static async Task<int> ConsumeAsync(ISourceBlock<byte[]> source)
{
    int bytesProcessed = 0;

    while (await source.OutputAvailableAsync())
    {
        byte[] data = await source.ReceiveAsync();
        bytesProcessed += data.Length;
    }

    return bytesProcessed;
}

static async Task Main()
{
    var buffer = new BufferBlock<byte[]>();
    var consumerTask = ConsumeAsync(buffer);
    Produce(buffer);

    var bytesProcessed = await consumerTask;

    Console.WriteLine($"Processed {bytesProcessed:#,#} bytes.");
}

.NET 6中的平行消费者

可以通过使用等待和 aceiveallasync ：

static async Task<int> ConsumeAsync(IReceivableSourceBlock<byte[]> source)
{
    int bytesProcessed = 0;

    await foreach(var data in source.ReceiveAllAsync())
    {
        bytesProcessed += data.Length;
    }

    return bytesProcessed;
}

并同时使用 parallel.foreachAsyAsyAsync来简化 消费者。 ：

static async Task ConsumeAsync(IReceivableSourceBlock<byte[]> source)
{
    var msgs=source.ReceiveAllAsync();

    await Parallel.ForEachAsync(msgs,
        new ParallelOptions { MaxDegreeOfParallelism = 4},
        msg=>ConsumeMsgAsync(msg));

}

默认情况下， Parallel.ForeachAsync 将使用核心

channels

频道与GO的频道相似。它们是专门为生产者/消费者场景构建的，并允许在比DataFlow库较低的级别上创建管道。如果今天构建了数据流库，则将其建立在频道之上。

只能通过其读者或作家界面直接访问通道。这是有意的，可以轻松地对方法进行管道。一个非常常见的模式是生产方法来创建其拥有的通道并仅返回ChannelReader。消费方法接受该读者作为输入。这样，生产者可以控制频道的寿命，而不必担心其他生产商是否正在写它。

使用频道，生产者看起来像这样：

ChannelReader<Message> Producer(CancellationToken token)
{
    var channel=Channel.CreateBounded(5);
    var writer=channel.Writer;

    _ = Task.Run(()=>{
        while(!token.IsCancellationRequested)
        {
           ...
           await writer.SendAsync(msg);
        }
    },token)
    .ContinueWith(t=>writer.TryComplete(t.Exception));

    return channel.Reader;
}

不寻常的 .continuewith（t =＆gt; writer.triter.trycomplete（t.exception））; 用于向作者发出信号。这也将表示读者也可以完成。这样，从一种方法传播到另一种方法。 任何例外都会传播

writer.trycomplete（t.exception））都不会阻止或执行任何重要工作，因此 。这意味着无需在工作任务上使用等待，这将通过重新造成任何例外来使代码复杂化。

一种消费方法只需要 ChannelReader 作为源。

async Task ConsumerAsync(ChannelReader<Message> source)
{
    await Parallel.ForEachAsync(source.ReadAllAsync(),
        new ParallelOptions { MaxDegreeOfParallelism = 4},
        msg=>consumeMessageAsync(msg)
    );
}

一种方法可以从一个通道读取并使用生产者模式将新数据发布到另一个数据：

ChannelReader<OtherMessage> ConsumerAsync(ChannelReader<Message> source)
{
    var channel=Channel.CreateBounded<OtherMessage>();
    var writer=channel.Writer;

    await Parallel.ForEachAsync(source.ReadAllAsync(),
        new ParallelOptions { MaxDegreeOfParallelism = 4},
        async msg=>{
            var newMsg=await consumeMessageAsync(msg);
            await writer.SendAsync(newMsg);
        })
       .ContinueWith(t=>writer.TryComplete(t.Exception));
}

Answer 2

您可以查看使用 dataFlow库。我不确定它是否比阻止票据更具性能。正如其他人所说的那样，不能保证您的消费速度比农产品更快，因此总是有可能落后。