并行处理 - 管理父/子依赖项答案

【问题标题】：Parallel Processing - Manage Parent/Child Dependencies并行处理 - 管理父/子依赖项
【发布时间】：2014-07-05 00:07:12
【问题描述】：

我正在处理一个文本文件，其中包含Parallel.ForEach 中的员工及其家属。无法保证文件中的顺序。在每次迭代中，我都会创建一个员工或依赖对象。员工对象被添加到ConcurrentDictionary。家属是员工的财产，需要与员工相关联。问题是如果员工不在字典中，当我尝试添加受抚养人时，永远不会添加受抚养人。我可以缓存这些“孤立的”依赖项并在 ForEach 完成时添加它们，但我认为有更好的方法。

有什么方法可以在我的Parallel.ForEach 中等待/旋转/加入，直到添加员工，然后添加受抚养人。我不赞成这个解决方案，所以欢迎使用替代方案。

这是我的代码（为简洁而编辑）：

var cx = System.IO.File.ReadLines(path);
var _employeeDictionary = new ConcurrentDictionary<string, Employee>();

Parallel.ForEach(cx, _options, line =>
{
    Employee employee = null;
    switch (line.Substring(0, 2))
    {
        case EmployeeLine:
            // Employee is created and added to dictionary....
            _employeeDictionary.GetOrAdd(winID, employee);
            break;

        case DependentLine:
            // Dependent is created
            // WILL NOT BE ADDED IF THE EMPLOYEE HASN'T BEED ADDED YET
            if (_employeeDictionary.TryGetValue(dependent.EmpID, out employee))
            {
                lock (locker)
                {
                    employee.AddDependent(ci, dependent);
                }
            }
            break;
    }
});

【问题讨论】：

标签： c# parallel-processing task-parallel-library

【解决方案1】：

有什么方法可以等待/旋转/加入我的 Parallel.ForEach

是的，但它们都会降低您的性能并引入死锁的可能性。所以这不是要走的路。

当然最后一个if (Trygetvalue...)) 需要一个else 分支，否则你会丢失数据。将 Dependents 存储在第一个 parallel.ForEach() 之后的列表中以供处理。

然后您不妨将所有 Dependents 存储在那里，而不是在第一次运行时查找它们。更简单，甚至可能更快。

【讨论】：

【解决方案2】：

让并发对你有用而不是对你不利的关键是让你的代码在功能上更有效。与其以必要术语（“对每一行都这样做”）进行思考，不如从如何转换数据的角度来思考。并行进行数据转换：如果您不修改状态，则本质上是线程安全的。然后只对真正需要命令的部分使用命令式编程。

var cx = System.IO.File.ReadAllLines(path);

var lineInfo = cx.AsParallel()
    .Select(line => new {
        lineCode = line.Substring(0, 2),
        line
    })
    .ToList()
    .AsParallel();

var employeeDictionary = lineInfo
    .Where(e => e.lineCode == EmployeeLine)
    .Select(e => ParseEmployee(e.line))
    .ToDictionary(e => e.winId);

var dependentLookup = lineInfo
    .Where(e => e.lineCode == DependentLine)
    .Select(e => ParseDependent(e.line))
    .ToLookup(d => d.EmpId);

Parallel.ForEach(employeeDictionary.Values, _options, employee => 
{
    foreach(var dependent in dependentLookup[employee.winId])
    {
        // It's even better if you can have an "AddDependents" method
        // to avoid the foreach, and leverage the efficiencies of "AddRange"-type
        // methods.
        employee.AddDependent(dependent);
    }
});

还值得注意的是，在此代码中进行并行处理实际上可能并不值得。我建议在有或没有并行性的情况下对其进行基准测试，如果您没有看到明显的改进，请不要打扰。

【讨论】：

但这会多次迭代数据。
@AgentShark：只要您不多次处理数据，多次迭代数据并不是什么大问题。感谢您的评论，我确实将ReadLines 更改为ReadAllLines，以便在这种情况下我们迭代一个数组。但除此之外，应该不会对性能造成明显影响。
迭代 ReadLines 是（曾经）一个问题，你不想重复任何 I/O。切换到 ReadAllLines 也是一个（小的）去优化。
你的意思是foreach(var dependent in dependentLookup[employee.winId])？
@HenkHolterman：您对避免使用 ReadAllLines 有什么建议吗？对我来说，这似乎是所有可能的邪恶中较小的一个。

【解决方案3】：

你拼出了你自己的答案。他们是孤儿，因为员工还不存在。您必须等到最后才能确保员工在那里。锁定会降低性能。完成初始文件读取后，只需将孤儿添加到并发列表和该列表上的并行 foreach 即可。

【讨论】：

【解决方案4】：

假设您的行数超过了您正在使用的机器上的处理器数，那么执行父集然后子集不会显着降低性能。这甚至可以避免检查这些值的开销，尤其是因为您必须为此类选项锁定字典。

另一种选择是实现一个 UnknownEmployee，它派生自 Employee 并表示其家属已知但员工不知道的条目。每当父项创建依赖项时未知，创建一个 UnknownEmployee 条目并将其设置为依赖项。每当创建 Employee 时，检查它们的键是否已经存在，如果存在，请确保在添加到字典之前复制 Dependents。

【讨论】：