为啥我的多线程代码没有更快？

Posted 2023-04-12

【中文标题】为啥我的多线程代码没有更快？

【英文标题】：Why is my multi-threaded code not faster?为什么我的多线程代码没有更快？

【发布时间】：2021-10-11 16:29:00

【问题描述】：

我在控制台应用程序中运行它：

public void ForEachParallel(Action<TElement> action)

    var elements = new Queue<TElement>(_set);

    var tasks = Enumerable.Range(0, _threadCount)
                          .Where(index => elements.Any())
                          .Select(index => elements.Dequeue())
                          .Select(element => Task.Run(() => action(element)))
                          .ToList();

    while (tasks.Any())
    
        var index = Task.WaitAny(tasks.ToArray());

        tasks.RemoveAt(index);

        if (elements.Any())
        
            var element = elements.Dequeue();
            tasks.Add(Task.Run(() => action(element)));
        

    

我有一个等效的ForEach 方法，它以串行方式完成所有这些工作。我正在使用 10 个线程，但 ForEachParallel 所花费的时间与 ForEach 一样多。我有一个 6 核的 i7。要么这有很多开销，要么它以某种方式使用单个线程运行这些任务。

每个动作都是独立的读取、处理和写入。

【问题讨论】：

这在很大程度上取决于您尝试运行的“操作”。并行化这些过程会产生固定的开销。另外取决于你在里面做什么，如果你正在访问内存，那实际上可能根本不是并行的。

您是否尝试过在Parallel.ForEach 中运行以查看是否得到相同的结果？如果这样更快，您至少可以更有信心这是您的方法有问题。

如果你有 6 个内核，那么试试 6 个线程。您不能过度并行内核。

@Jordan 显然如果不知道action 做了什么以及从这样的代码片段中，我们无法知道您的程序为什么很慢。你必须产生一个最小的例子并给出统计数据。此外，您的性能问题很可能不在上述代码中，例如争用共享资源。

@Jordan - 我刚刚使用您的代码进行了测试，它确实比串联运行更快。我将它与使用 Observables 的相同实现进行了比较，它同样快。显然问题出在您的action 代码中。

【参考方案1】：

这是我的测试代码：

void Main()

    Action<int> action = n =>
    
        Console.Write($" +n ");
        Thread.Sleep(TimeSpan.FromSeconds(n + 1));
        Console.Write($" n- ");
    ;
    
    ForEachParallel(Enumerable.Range(0, 6), 4, action);


public void ForEachParallel<TElement>(IEnumerable<TElement> source, int threadCount, Action<TElement> action)

    var elements = new Queue<TElement>(source);

    var tasks =
        source
            .Take(threadCount)
            .Where(index => elements.Any())
            .Select(index => elements.Dequeue())
            .Select(element => Task.Run(() => action(element)))
            .ToList();

    while (tasks.Any())
    
        var index = Task.WaitAny(tasks.ToArray());

        tasks.RemoveAt(index);

        if (elements.Any())
        
            var element = elements.Dequeue();
            tasks.Add(Task.Run(() => action(element)));
        
    

它实际上与您的 ForEachParallel 相同，但我使它更通用。

当我更改 threadCount 时，我得到不同的执行长度。这显然按预期运行。