并行编程：动态队列的进程列表答案

【问题标题】：Parallel Programming : Process List of Dynamic Queues并行编程：动态队列的进程列表
【发布时间】：2020-01-15 21:56:13
【问题描述】：

我有一本包含不同类型项目的字典。通过向现有条目添加新条目和项目来更改字典。

并行处理记录的有效方法是什么。当字典记录 1 有 1 项，字典记录 2 有 200 项时。当一个新项目被添加到字典记录 1 时，它应该创建一个新线程并处理它，如果旧记录已经被处理，而不是等待整个批次完成。

Dictionary<ItemType, Queue<Item>> ItemsTypes = new Dictionary<Guid, Queue<Item>>();

使用下面的代码，我需要等到第一批项目处理完毕后再开始新的批次。

result = Parallel.ForEach(ItemsTypes, Items => processor.ProcessItems(Items.Value));

【问题讨论】：

如果在迭代时添加/删除项目，则无法使用 foreach。如果你想处理一个新项目 - 在旧项目“已经”处理之后，这意味着你不想并行运行。您还谈到了批处理，这也是单线程世界中的一个术语。它会等待这个，而不是那个......你的话有点混乱。
每次添加项目时生成一个新线程似乎效率很低 - 这种“处理”的本质是什么？

标签： c# .net multithreading

【解决方案1】：

您在这里使用 Parallel.ForEach 所做的是数据并行。这个 ForEach 循环开始的那一刻，它会获取 Dictionary 当前状态的（某种）快照并运行它。我相信，一旦 forEach 开始，您是否向任何地方添加更多项目都没关系。

您实际需要做的是 Pub/Sub 模型，其中添加新项目/条目的代码会发布一个事件，该事件会导致创建新线程并处理项目。或者，您可以为每个项目添加一个 Task 对象。

【讨论】：

【解决方案2】：

代替 Parallel.Foreach()，在 foreach 中处理 Dictionary 元素，并为每个元素分离新线程并跟踪状态以防止队列项的并行处理。在 ProcessItems 中将 Status 设置为 false。

ConcurrentDictionary<Guid, bool> Status = new ConcurrentDictionary<Guid, bool>();

while (ItemsTypes.Values.Count() > 0)
{
    foreach (var item in ItemsTypes)
    {
      if (item.Value.Count > 0)
      {
        if (Status[issuer.Key] == false)
        {
          Status[issuer.Key] = true;
          Task.Run(() => ProcessItems(item.Value.Dequeue()));
         }
       }
     }
  }

【讨论】：