multithreading - 并行收集处理的应用程序设计

Question

我正在试验 System.Collections.Concurrent 命名空间，但我在实现我的设计时遇到了问题。

1. 我的输入队列（ConcurrentQueue）正在从一个线程很好地填充，该线程在启动时执行一些 I/O 以读取和解析。
2. 接下来，我在输入队列上启动 Parallel.ForEach()。我正在对每个项目进行一些 I/O 绑定工作。
3. 为 ForEach() 中处理的每个项目创建一个日志项目，并将其放入结果队列中。

我想做的是开始记录我开始阅读输入，因为我可能无法将所有日志项都放在内存中。等待项目进入结果队列的最佳方式是什么？是否有我应该查看的设计模式或示例？

Answer 1

我认为您正在寻找的模式是生产者/消费者模式。更具体地说，您可以围绕 TPL 和 BlockingCollection 构建生产者/消费者实现。

您要阅读的主要概念是：

1. 任务，
2. 阻塞集合，
3. TaskFactory.ContinueWhenAll（将允许您在一组任务/线程完成运行时执行一些操作）。
4. BlockingCollection 中的边界和阻塞。这允许您为输出集合设置最大大小（出于内存原因），如果达到您指定的最大大小，生产者线程将等待消费者拾取元素。
5. BlockingCollection.CompleteAdding和BlockingCollection.IsCompleted可用于同步生产者和消费者（生产者可以说何时完成，消费者可以检查并继续运行，直到生产者完成）。

我链接的第二篇文章中有一个更完整的示例。

在您的情况下，我认为您希望消费者从结果队列中取出东西并尽快处理它们（将它们写入日志存储或类似存储）。

因此，您转储日志项的最终集合应该是 a BlockingCollection，而不是 a ConcurrentQueue。