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

Question

我正在使用

Parallel.ForEach

循环来做一些工作，并使用

localInit

对其进行初始化，如下所示：

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

根据文档：

localInit

，或者初始化线程局部变量的函数。 对于每个分区调用此函数一次

Parallel.ForEach<TSource>

操作执行。我们的示例初始化 线程局部变量为零。

所以我尝试像这样使用它，但我观察到循环不断地杀死并创建新任务，这导致频繁调用

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

如果我正确理解代码，

Parallel.ForEach()

每隔几百毫秒就会重新启动每个

Task

。这意味着，如果每次迭代都很大（通常应该如此），您将获得大量

Task

，从而获得大量对

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

每个线程执行栏仅创建一次。但你知道有多少并行执行吗？并行执行引擎可以自行决定启动任意数量的并行执行。

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

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

Answer 3

这个重载不是唯一的，所以你可以尝试这个：

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),
}