在并行处理列表时避免冲突

Question

我不确定标题是最好的，但我认为下面的代码会有所帮助。为了简单起见，假设我有一个类的列表，它有两个值category和itemid。我需要处理列表中的每个项目，但我不能同时处理具有相同类别的两条记录(它最终会更新数据库中的相同记录)。现有代码按行号对项目进行分组(按类别划分)。这样我就可以并行运行每个组的项目，因为永远不会有重复的类别。

我觉得有更好的方法来做这件事，但我不确定是什么。现在处理的顺序可能看起来像这样:

行号1:a, d, c, b, e

第2行:a, b, c

第3行:b, c

第4行:c

它可以工作，但是如果类别"需要一段时间。从理论上讲，它可以处理额外的记录，但它只能处理单个记录。

任何想法吗?或者我如何更好地搜索(因为到目前为止我一直一无所获)?

在下面的代码中，我注释掉了".Dump()"调用，但是如果您在LINQPad中运行代码，这些可能会很有用。

void Main()
{
var list = new List<Item> {
new Item("a", "1"),
new Item("a", "2"),
new Item("b", "3"),
new Item("b", "4"),
new Item("b", "5"),
new Item("c", "6"),
new Item("c", "7"),
new Item("c", "8"),
new Item("c", "9"),
new Item("d", "10"),
new Item("e", "11"), 
};
var itemsToProcess = list
.OrderBy(l => l.Category)
.GroupBy(i => i.Category)
.Select( group => new { Group = group, Count = group.Count() } )
.SelectMany( groupWithCount =>
groupWithCount.Group.Select(b => b)
.Zip(
Enumerable.Range(1, groupWithCount.Count),
(j, i) => new { j.Category, j.ItemId, RowNumber = i }
)
);

// itemsToProcess.Dump();
var itemsGroupedByRow = itemsToProcess.GroupBy(i => i.RowNumber);
foreach (var items in itemsGroupedByRow)
{
// items.Key.Dump("Row Number");
Parallel.ForEach(items,
(item) =>
{
// do stuff
// $"{item.Category} - {item.ItemId}".Dump();
}
);
}
}
public class Item
{
public Item(string category, string itemId)
{
Category = category;
ItemId = itemId;
}
public string Category { get; set; }
public string ItemId { get; set; }
}

Answer 1

您应该按类别单独考虑这个问题，而不是试图自己组装处理单元(您的"行")。基本上，并行处理每个类别，但每次处理每个类别的每个项目，按项目id排序。

像这样(Sleep是为了说明e的长时间工作不影响其余的处理):

public static void Run()
{
var list = new List<Item> {
new ("a", "1"),
new ("a", "2"),
new ("b", "3"),
new ("b", "4"),
new ("b", "5"),
new ("c", "6"),
new ("c", "7"),
new ("c", "8"),
new ("c", "9"),
new ("d", "10"),
new ("e", "11")
};
var categories = list.Select(i => i.Category).Distinct();
Parallel.ForEach(categories,
(category) => ProcessCategory(list
.OrderBy(i => i.ItemId)
.Where(i => i.Category.Equals(category))
.ToList()));
Console.WriteLine("DONE!");
}
private static void ProcessCategory(List<Item> items)
{
foreach (var item in items)
{
if (item.Category == "e")
Thread.Sleep(1000);
Console.WriteLine($"{item.Category} - {item.ItemId}");
}
}