如何防止 Parallel.ForEach 循环在运行时更改任务数？

Question

我正在使用Parallel.ForEach 循环来完成一些工作，并使用localInit 对其进行初始化，如下所示：

localInit: () => new
{
    foo = new Foo(),
    bars = CreateBars(),
}

根据 MSDN：

localInit，或初始化线程局部变量的函数。 该函数在每个分区调用一次 Parallel.ForEach 操作执行。我们的示例初始化 线程局部变量为零。

所以我尝试这样使用它，但我观察到循环不断地杀死并创建新任务，这导致频繁调用localInit。我的选择会适得其反，不能按预期工作。

我认为当Parallel.ForEach 会创建例如四个分区时，它会让它们保持活动状态，直到它迭代所有项目但它没有。它正在调用localFinally 和localInit 数百次，以获得一个包含几千个项目的集合。怎么样？

可以以某种方式阻止这种行为吗？我真的希望节省一些资源，但它并没有真正让我。

---

下面是循环的样子：

var parallelLoopResult = Parallel.ForEach
(
    source: items,
    parallelOptions: parallelOptions,
    localInit: () => new
    {
        foo = new Foo(),
        bars = CreateBars(),
    },
    body: (item, loopState, i, local) =>
    {
        parallelOptions.CancellationToken.ThrowIfCancellationRequested();

        var results = local.bars.Select(x => ...).ToList().

        ....

        return local;
    },
    localFinally: local =>
    {
        local.foo.Dispose();
        lock (aggregateLock)
        {
            ... process transformed bars
        }
    }
);

并行选项：

var parallelOptions = new ParallelOptions
{
    CancellationToken = cancellationTokenSource.Token,
#if DEBUG
    MaxDegreeOfParallelism = 1
    //MaxDegreeOfParallelism = Environment.ProcessorCount
#else
    MaxDegreeOfParallelism = Environment.ProcessorCount
#endif
};

Answer 1

如果我对the code 的理解正确，Parallel.ForEach() 会每隔几百毫秒重新启动每个Task。这意味着，如果每次迭代都很大（通常应该如此），您将获得大量Tasks，因此会收到大量对localInit 和localFinally 的调用。这样做的原因是对于同一进程中也使用相同 ThreadPool 的其他代码的公平性。

我认为没有办法改变Parallel.ForEach() 的这种行为。我认为解决这个问题的一个好方法是编写自己的简单版本的Parallel.ForEach()。考虑到您可以利用Partitioner<T> 并根据您需要Parallel.ForEach() 的哪些功能，它可能相对简单。例如，类似：

public static void MyParallelForEach<TSource, TLocal>(
    IEnumerable<TSource> source, int degreeOfParallelism,
    Func<TLocal> localInit, Func<TSource, TLocal, TLocal> body, Action<TLocal> localFinally)
{
    var partitionerSource = Partitioner.Create(source).GetDynamicPartitions();

    Action taskAction = () =>
    {
        var localState = localInit();

        foreach (var item in partitionerSource)
        {
            localState = body(item, localState);
        }

        localFinally(localState);
    };

    var tasks = new Task[degreeOfParallelism - 1];

    for (int i = 0; i < degreeOfParallelism - 1; i++)
    {
        tasks[i] = Task.Run(taskAction);
    }

    taskAction();

    Task.WaitAll(tasks);
}

Answer 2

每个 thread 执行栏只创建一次。但是你知道完成了多少并行执行吗？并行执行引擎可以自行决定是否启动任意数量的并行执行。

如果要限制并行执行，请使用MaxDegreeOfParallelism 属性。这将对一次创建多少条柱形设置一个上限。它仍然无法控制创建的条形总数，而且条形总数可能低于您现在的预期。

如果您想要明确控制，请手动创建任务。

Answer 3

This overload 不是唯一的，所以你可以试试这个：

var bars = CreateBars();
Parallel.Foreach(bars, b => { /* your action here */};

但是，如果您真的想为每个线程创建bars 的副本，您可以使用 LINQ 中的一些复制方法（假设您的条形图是一个 IEnumerable<T> 变量）：

var bars = CreateBars();
localInit: () => new
{
    foo = new Foo(),
    bars = new List<IBar>(bars),
}