设置 BlockingCollection 中项目的最大处理量答案

【问题标题】：Set a maximum processing of items in a BlockingCollection设置 BlockingCollection 中项目的最大处理量
【发布时间】：2021-04-03 15:46:54
【问题描述】：

我有一个blockingcollection的下载，我想有一个我可以做的最大并发下载数，也就是最大并发await downloadService.download(release)，但是BlockingCollection的项目数应该基本上是无限的，所以假设我可以在 BlockingCollection 中有 49945 次下载，但我应该同时下载最多 5 次，每当下载完成时，它会再下载一次，如果可能的话。这是我的实际代码：

BlockingCollection<Download> sendQueue = new BlockingCollection<Download>(new ConcurrentQueue<Download>());
while (true)
            {
                var release = sendQueue.Take();
                ThreadPool.QueueUserWorkItem(async rlZ =>
                {
                    //do whatever you have to do
                    await downloadService.download(release);
                }, release);
            }

【问题讨论】：

一旦任何线程完成需要减少计数并将新项目添加到队列中，您需要保留 5 个项目的计数器。 docs.microsoft.com/en-us/dotnet/api/…
@mjwills 这不是限制 BlockingCollection 可以拥有的元素数量，而不是我想要对 BlockingCollection 的元素执行的并发操作的数量吗？
ThreadPool.QueueUserWorkItem 和 BlockingCollection 不是最适合在 I/O 密集型工作（例如从 Web 下载内容）时使用的工具。你可以在这里看到更好的方法：How to limit the amount of concurrent async I/O operations?

标签： c# multithreading blockingcollection

【解决方案1】：

设置 BlockingCollection 中项目的最大处理量

您不能在 BlockingCollection 中设置它，因为 BlockingCollection 不进行任何处理 - 它只是一个集合。

如果您想限制同时处理的处理器数量，则需要将其放入处理代码本身。限制异步代码的标准方法是使用SemaphoreSlim。

但是，在您的情况下，我建议您使用 n 消费者。这使代码更干净。

var consumers = new Task[]
{
  Task.Run(ConsumerAsync),
  Task.Run(ConsumerAsync),
  Task.Run(ConsumerAsync),
  Task.Run(ConsumerAsync),
  Task.Run(ConsumerAsync),
};
await Task.WhenAll(consumers);

static async Task ConsumerAsync()
{
  foreach (var release in sendQueue.GetConsumingEnumerable())
    await downloadService.download(release);
}

提示：如果您使用像 Channel 这样的异步兼容队列而不是 BlockingCollection，那么您可以使用异步枚举而不是消费枚举并删除 Task.Run 调用。

【讨论】：

我读过 Channel 并且它非常棒！我处于一个位置，我有一个生产者将无限和无限数量的下载添加到队列（现在是频道）和 n 个消费者（同时最多 5 个），我目前真的很困惑，我必须创建一个带有 while true 循环的单个任务（消费者），我 TryRead 每个下载项目？使用 SemaphoreSlim 和 WaitToReadAsync 创建 n 个任务控制我有 5 个并且不再有通道？听起来很简单，我只想处理无限数量的网络下载，但最多同时处理 5 个。
@TePi：对于频道，使用与我发布的代码类似的结构。只需将GetConsumingEnumerable 替换为ReadAllAsync 并将foreach 更改为await foreach。然后删除Task.Run 调用。你最终会得到五个异步消费者。