在 C# Parallel.ForEach 中的线程上保持变量答案

【问题标题】：Persist variable on thread in C# Parallel.ForEach在 C# Parallel.ForEach 中的线程上保持变量
【发布时间】：2021-02-12 19:56:52
【问题描述】：

我希望并行处理依赖于对象 (State)、不是线程安全且其构造耗时的任务。

出于这个原因，我正在调查partition-local variables，但要么我做错了，要么正在寻找其他东西。这或多或少代表了我当前的实现：

Parallel.ForEach<string, State>(folders, config, () => new State(), (source, loopState, index, threadState) => {
    var content = File.ReadAllText(source);        // read file
    var result = threadState.doSomething(content); // do something
    File.WriteAllText(outputFile, result);         // write output
    return threadState;
}, (threadState) => { });

但是，我在 State 初始化程序中添加了一个 Console.WriteLine，并且我看到对于循环的每次迭代，都会调用 State 构造函数，从而对性能造成很大影响。我希望将一个线程中State 的实例传递给同一线程上的后续迭代。

我怎样才能做到这一点？

【问题讨论】：

either I am doing it wrong - 你不是，只是not documented very well。
@GSerg 似乎同时docs 已更新，相关信息现在正确："每个任务调用一次localInit 委托 参与循环的执行”（强调）

标签： c# multithreading parallel-processing

【解决方案1】：

您有几个选择。最简单的方法是创建单个State 对象，并使用lock 同步对其的访问：

var state = new State();

Parallel.ForEach(folders, config, source =>
{
    var content = File.ReadAllText(source);
    string result;
    lock (state) { result = state.DoSomething(content); }
    File.WriteAllText(outputFile, result);
});

我认为这是不可行的，因为 DoSomething 方法很耗时，并且同步它会破坏并行性。

另一种选择是使用ThreadLocal<State>。此类提供数据的线程本地存储，因此创建的State 对象的数量将等于Parallel.ForEach 使用的线程数量。

var threadLocalState = new ThreadLocal<State>(() => new State());

Parallel.ForEach(folders, config, source =>
{
    var content = File.ReadAllText(source);
    var result = threadLocalState.Value.DoSomething(content);
    File.WriteAllText(outputFile, result);
});

这可能会比Parallel.ForEach<TSource, TLocal> 重载创建更少的State 对象，但仍不等于配置的MaxDegreeOfParallelism。 Parallel.ForEach 使用来自ThreadPool 的线程，如果folders 的列表足够长，它很有可能在计算过程中使用所有线程。而且您几乎无法控制ThreadPool 的大小。所以这也不是一个特别诱人的解决方案。

我能想到的第三个也是最后一个选项是创建一个由State 对象组成的池，并在每个循环中创建一个Rent/Return：

var statePool = new ObjectPool<State>(() => new State());

Parallel.ForEach(folders, config, source =>
{
    var state = statePool.Rent();
    var content = File.ReadAllText(source);
    var result = state.DoSomething(content);
    File.WriteAllText(outputFile, result);
    statePool.Return(state);
});

这样实例化的State对象的数量将等于最大并行度。

唯一的问题是.NET 平台中没有ObjectPool 类（只有一个ArrayPool 类），所以您必须找到一个。这是一个基于ConcurrentBag的简单实现：

public class ObjectPool<T> : IEnumerable<T> where T : new()
{
    private readonly ConcurrentBag<T> _bag = new ConcurrentBag<T>();
    private readonly Func<T> _factory;

    public ObjectPool(Func<T> factory = null) => _factory = factory;

    public T Rent()
    {
        if (_bag.TryTake(out var obj)) return obj;
        return _factory != null ? _factory() : new T();
    }

    public void Return(T obj) => _bag.Add(obj);

    public IEnumerator<T> GetEnumerator() => _bag.GetEnumerator();
    IEnumerator IEnumerable.GetEnumerator() => this.GetEnumerator();
}

【讨论】：