c# - 为什么在 Parallel.ForEach 中每个线程会多次调用 localInit Func

Question

我正在编写一些代码来处理大量数据，我认为让 Parallel.ForEach 为它创建的每个线程创建一个文件会很有用，这样输出就不需要同步(至少对我来说是这样) ).

看起来像这样:

Parallel.ForEach(vals,
    new ParallelOptions { MaxDegreeOfParallelism = 8 },
    ()=>GetWriter(), // returns a new BinaryWriter backed by a file with a guid name
    (item, state, writer)=>
    {
        if(something)
        {
            state.Break();
            return writer;
        }
        List<Result> results = new List<Result>();

        foreach(var subItem in item.SubItems)
            results.Add(ProcessItem(subItem));

        if(results.Count > 0)
        {
            foreach(var result in results)
                result.Write(writer);
        }
        return writer;
    },
    (writer)=>writer.Dispose());

我预计会发生的情况是最多会创建 8 个文件，并且会在整个运行时持续存在。然后，当整个 ForEach 调用完成时，每个都将被处置。真正发生的是 localInit 似乎为每个项目调用一次，所以我最终得到了数百个文件。编写器也会在每个处理的项目结束时被处理掉。

这表明发生了同样的事情:

var vals = Enumerable.Range(0, 10000000).ToArray();
        long sum = 0;
        Parallel.ForEach(vals,
            new ParallelOptions { MaxDegreeOfParallelism = 8 },
            () => { Console.WriteLine("init " + Thread.CurrentThread.ManagedThreadId); return 0L; },
            (i, state, common) =>
            {
                Thread.Sleep(10);
                return common + i;
            },
                (common) => Interlocked.Add(ref sum, common));

我明白了:

init 10
init 14
init 11
init 13
init 12
init 14
init 11
init 12
init 13
init 11
... // hundreds of lines over < 30 seconds
init 14
init 11
init 18
init 17
init 10
init 11
init 14
init 11
init 14
init 11
init 18

注意:如果我省略了 Thread.Sleep 调用，它有时似乎“正常”运行。对于它决定在我的电脑上使用的 4 个线程，localInit 只被调用一次。然而，并非每次都如此。

这是函数的预期行为吗？导致它这样做的幕后发生了什么？最后，什么是获得我想要的功能的好方法，ThreadLocal？

顺便说一句，这是在 .NET 4.5 上。

Answer 1

Parallel.ForEach 并不像您想象的那样工作。重要的是要注意该方法是在 Task 类之上构建的，Task 和 Thread 之间的关系不是 1: 1。例如，您可以有 10 个任务在 2 个托管线程上运行。

尝试在您的方法主体中使用这一行而不是当前的:

Console.WriteLine("ThreadId {0} -- TaskId {1} ",
                  Thread.CurrentThread.ManagedThreadId, Task.CurrentId);

您应该看到 ThreadId 将在许多不同的任务中重复使用，由它们的唯一 ID 显示。如果您保留或增加对 Thread.Sleep 的调用，您会看到更多。

Parallel.ForEach 方法如何工作的(非常)基本思想是，它需要您的枚举创建一系列将运行枚举的进程部分的任务，这是完成的方式很大程度上取决于输入。还有一些特殊的逻辑，检查任务超过一定毫秒数而没有完成的情况。如果情况属实，则可能会产生一个新任务来帮助减轻工作量。

如果您查看了 Parallel.ForEach 中关于 localinit 函数的文档，您会注意到它说它返回每个_task_ 的本地数据的初始状态，而不是每个线程。

您可能会问为什么要生成 8 个以上的任务。该答案与上一个相似，可在 ParallelOptions.MaxDegreeOfParallelism 的文档中找到.

Changing MaxDegreeOfParallelism from the default only limits how many concurrent tasks will be used.

此限制仅针对并发任务的数量，而不是对在整个处理过程中创建的任务数量的硬性限制。正如我上面提到的，有时会产生一个单独的任务，这会导致您的 localinit 函数被多次调用并将数百个文件写入磁盘。

写入磁盘肯定是一个有一点延迟的操作，尤其是当您使用同步 I/O 时。当磁盘操作发生时，它会阻塞整个线程； Thread.Sleep 也会发生同样的情况。如果 Task 执行此操作，它将阻塞当前正在运行的线程，并且没有其他任务可以在其上运行。通常在这些情况下，调度程序会生成一个新的 Task 来帮助弥补不足。

And lastly, what's a good way to get my desired functionality, ThreadLocal?

归根结底，线程局部变量对 Parallel.ForEach 没有意义，因为您不是在处理线程；你正在处理任务。本地线程可以在任务之间共享，因为许多任务可以同时使用同一个线程。此外，任务的本地线程可能会在执行过程中发生变化，因为调度程序可能会抢占它的运行，然后在另一个线程上继续执行，该线程将具有不同的本地线程。

我不确定最好的方法，但是你可以依靠 localinit 函数来传递你想要的任何资源，只允许一个资源在一个线程中使用一次。您可以使用 localfinally 将其标记为不再使用，从而可供其他任务获取。这就是这些方法的设计目的；每个方法仅在每个生成的任务中调用一次(请参阅 Parallel.ForEach MSDN 文档的备注部分)。

您也可以自己拆分工作，创建自己的线程集并运行您的工作。然而，在我看来，这并不是什么好主意，因为 Parallel 类已经为您完成了这项繁重的工作。