并行foreach循环中缺少日志语句

问题描述

我正在使用并行的foreach / for循环，在特殊情况下，我需要使用嵌套的并行的foreach / for循环。当我尝试打印集合中的值时，有时不会打印不一致的控制台语句。请参见下面的代码。

Parallel.For(0,RunModuleConfigVariables.Count,new ParallelOptions { MaxDegreeOfParallelism = 3 },index => {
                string log = null;
                int count = 0;
                log += "Module Name " + RunModuleConfigVariables.Keys.ElementAt(index) + " thread: " + Thread.CurrentThread.ManagedThreadId + "\n";
                Parallel.ForEach(RunModuleConfigVariables[RunModuleConfigVariables.Keys.ElementAt(index)],new ParallelOptions { MaxDegreeOfParallelism = 10 },eachendpoint => {

                    log += "\t" + count + " Endpoint Name " + eachendpoint + "\n";
                    count++;
                });
                Console.WriteLine(log);
            });

集合：

集合类型为 ConcurrentDictionary （）

RunModuleConfigVariables:
        {
      "Module_1": [
        "Module_1_Endpoint_1","Module_1_Endpoint_2","Module_1_Endpoint_3","Module_1_Endpoint_4","Module_1_Endpoint_5","Module_1_Endpoint_6","Module_1_Endpoint_7","Module_1_Endpoint_8","Module_1_Endpoint_9","Module_1_Endpoint_10","Module_1_Endpoint_11","Module_1_Endpoint_12","Module_1_Endpoint_13","Module_1_Endpoint_14","Module_1_Endpoint_15","Module_1_Endpoint_16","Module_1_Endpoint_17","Module_1_Endpoint_18","Module_1_Endpoint_19"
      ],"Module_2": [
        "Module_2_Endpoint_1","Module_2_Endpoint_2","Module_2_Endpoint_3"
      ],"Module_3": [
        "Module_3_Endpoint_1"
      ]
    }

实际输出：

Module Name Module_1 thread: 4
        0 Endpoint Name Module_1_Endpoint_2
        1 Endpoint Name Module_1_Endpoint_1
        2 Endpoint Name Module_1_Endpoint_4
        3 Endpoint Name Module_1_Endpoint_5
        4 Endpoint Name Module_1_Endpoint_6
        5 Endpoint Name Module_1_Endpoint_7
        6 Endpoint Name Module_1_Endpoint_8
        18 Endpoint Name Module_1_Endpoint_9

Module Name Module_3 thread: 5
        0 Endpoint Name Module_3_Endpoint_1

Module Name Module_2 thread: 1
        0 Endpoint Name Module_2_Endpoint_2
        1 Endpoint Name Module_2_Endpoint_3
        2 Endpoint Name Module_2_Endpoint_1

预期的输出：（不需要排在同一顺序）

Module Name Module_1 thread: 5
        0 Endpoint Name Module_1_Endpoint_2
        1 Endpoint Name Module_1_Endpoint_3
        2 Endpoint Name Module_1_Endpoint_4
        3 Endpoint Name Module_1_Endpoint_5
        4 Endpoint Name Module_1_Endpoint_6
        5 Endpoint Name Module_1_Endpoint_7
        6 Endpoint Name Module_1_Endpoint_8
        7 Endpoint Name Module_1_Endpoint_9
        8 Endpoint Name Module_1_Endpoint_10
        9 Endpoint Name Module_1_Endpoint_11
        10 Endpoint Name Module_1_Endpoint_12
        11 Endpoint Name Module_1_Endpoint_13
        12 Endpoint Name Module_1_Endpoint_14
        13 Endpoint Name Module_1_Endpoint_15
        14 Endpoint Name Module_1_Endpoint_16
        15 Endpoint Name Module_1_Endpoint_17
        16 Endpoint Name Module_1_Endpoint_18
        17 Endpoint Name Module_1_Endpoint_19
        18 Endpoint Name Module_1_Endpoint_1

Module Name Module_2 thread: 4
        0 Endpoint Name Module_2_Endpoint_2
        1 Endpoint Name Module_2_Endpoint_3
        2 Endpoint Name Module_2_Endpoint_1

Module Name Module_3 thread: 1
        0 Endpoint Name Module_3_Endpoint_1

注意：输出不一致。有时能够看到所有子孩子，有时却看不到。我如何理解这一点，以及如何解决这个问题？

解决方法

我怎么理解？

并行处理意味着多个线程在同一时间处理事务。这会导致您需要注意各种奇怪的事情。

考虑这一行：

count++;

这条C＃指令实际上代表多种操作：

1. 将count变量中的值从内存加载到处理器中。
2. 将1添加到加载到处理器中的值中。
3. 将新值存储到count变量的内存位置。

现在想象两个线程同时执行这三个指令。他们都有可能在完成第3步之前先完成第1步。这意味着，如果count从零开始，那么两个线程现在都将count设置为1，这是“ t您的意图。

此行在读取log到写入点之间还有许多步骤：

log += "\t" + count + " Endpoint Name " + eachendpoint + "\n";

因此，您会发现一个线程覆盖（而不是添加）另一线程已经写入的值的频率要高得多。这就是您注意到的行为。

...让我知道，该怎么做才能克服这个问题。

首先，尽可能避免并行处理。

如果通过简单的foreach循环运行得足够快，请不要尝试对其进行优化。

如果通过简单的foreach循环运行得不够快，请找出原因。大多数情况下，这是由于I / O操作（磁盘或网络访问）所致。在这种情况下，请使用异步任务的并发执行，而不是多线程。参见https://stackoverflow.com/a/14130314/120955和What is the difference between asynchronous programming and multithreading?。

如果您执行的操作需要CPU能力，并且确实需要它们并行运行以从中挤出额外的性能，请尝试避免更改每个状态的状态（例如，设置共享变量的值，像count++）。一种很好的策略是命令/查询分离，您可以对不可变的数据结构进行并行处理以产生“答案”，然后使用这些答案进行必须在同一线程上全部进行的更改。这就是您的代码中的样子：

var logs = RunModuleConfigVariables
    .AsParallel()
    .WithDegreeOfParallelism(3)
    .Select(e =>
        "Module Name " + e.Key + " thread: " + Thread.CurrentThread.ManagedThreadId + "\n"
            + string.Join("\n",e.Value
                    .AsParallel()
                    .WithDegreeOfParallelism(10)
                    .Select((eachendpoint,index) => "\t" + index + " Endpoint Name " + eachendpoint)

    ));

Console.WriteLine(string.Join("\n",logs));

最后，如果您绝对必须并行更改状态，则需要花一些时间来了解锁，互斥锁，并发集合，atomic operations和其他类似工具，并确保仅使用线程并行上下文中的-safe方法，以确保您做的“正确”。

这可能会导致如下情况：

Parallel.ForEach(RunModuleConfigVariables,new ParallelOptions { MaxDegreeOfParallelism = 3 },pair =>
{
    Console.WriteLine("Module Name " + pair.Key + " thread: " + Thread.CurrentThread.ManagedThreadId);
    var count = 0;
    Parallel.ForEach(pair.Value,new ParallelOptions { MaxDegreeOfParallelism = 10 },eachendpoint =>
    {
        var thisCount = Interlocked.Increment(ref count);
        Console.WriteLine("\t" + thisCount + " Endpoint Name " + eachendpoint + "\n");
    });
});

,

问题是您的变量log被多个线程分配。在尝试写入之前，您需要先lock。

Parallel.For(0,RunModuleConfigVariables.Count,index => {
                string log = null;
                int count = 0;
                log += "Module Name " + RunModuleConfigVariables.Keys.ElementAt(index) + " thread: " + Thread.CurrentThread.ManagedThreadId + "\n";
                object locker = new object();
                Parallel.ForEach(RunModuleConfigVariables[RunModuleConfigVariables.Keys.ElementAt(index)],eachendpoint => {
                    lock(locker)
                        log += "\t" + (count++) + " Endpoint Name " + eachendpoint + "\n";
                });
                Console.WriteLine(log);
            });