多线程在我的 c# 程序中执行比顺序执行需要更多时间

Posted 2023-04-12

【中文标题】多线程在我的 c# 程序中执行比顺序执行需要更多时间

【英文标题】：Multithreading takes more time to execute in my c# program than sequential execution

【发布时间】：2021-11-01 11:15:12

【问题描述】：

我的 C# 程序做了两个工作：

将文件中的记录读入队列 - 每条记录一行 从队列中读取数据并对数据进行处理。

该文件有大约 10 亿条记录。每条记录最多 40 个字符。下面的示例代码描述了我在多线程方式中的表现。

//File 1: MessageQueue.cs (Reads data from file into queue)
public class MessageQueue

    public Queue<string> logQueue = new Queue<string>();
    public bool fileread = false;
    string textFile = <file name with location>
    public void populateQueue()

            using (StreamReader file = new StreamReader(textFile))
            
                string ln;
                while (true)
                
                    while (logQueue.Count < 1000000)
                    
                        ln = file.ReadLine();
                        if(ln == null)
                        
                            lock (logQueue)
                            
                                fileread = true;
                            
                            file.Close();
                            break;
                        
                        lock (logQueue)
                        
                            logQueue.Enqueue(ln);
                        
                    
                    
                    if (fileread)
                        break;
                    Thread.Sleep(1);
                
            
        


//File 2: Processes data

   

     public class DoSomething
        
            MessageQueue msgQueue;
    
        public DoSomething(MessageQueue msgQueue)
        
        this.msgQueue = msgQueue;
        
        
    
            public void ProcessData()
            
                Queue<String> logsQueue = msgQueue.logQueue;
    
                while (true)
                
                    string logMessage = "";
                    lock (logsQueue)
                    
                        if (logsQueue.Count != 0)
                            logMessage = logsQueue.Dequeue();
    
                        else if (!msgQueue.fileread)
                            
                                continue; //Queue is empty but there are still records in file
                            
                            else
                            
                                break; //Whole file is read and queue is also empty
                            
                    
    //Do more processing with logMessage
            

File 3: Main

static void Main(string[] args)
        

            DateTime startTime = DateTime.Now;
            MessageQueue msgQueue = new MessageQueue();
            DoSomething doSomething= new DoSomething(msgQueue);

            Thread thread1 = new Thread(new ThreadStart(msgQueue.populateQueue));
            Thread thread2 = new Thread(new ThreadStart(doSomething.ProcessData));

            thread1.Start();
            thread2.Start();
            DateTime endTime = DateTime.Now;
            Console.WriteLine("Time spent in whole loop is " + endTime.Subtract(startTime).TotalMilliseconds);

        

我的想法是确保队列在一个线程中不断填充，而另一个线程正在处理来自队列的数据。另一种方法是顺序读取队列中的 100 万条记录，然后处理该数据，一旦队列为空，再读取 100 万条记录/进程并以这种方式继续，直到不读取整个文件。事实证明，顺序方式比多线程方式更快。我在多线程代码中做错了吗？

【问题讨论】：

我们无法调试故事，有很多方法可以让多线程变慢并做错事。也许你可以创建一个小的minimal reproducible example

消费者线程如何知道Queue<T> 何时包含足够的数据？它是否将队列集中在一个循环中？您是否在循环中添加了任何Thread.Sleep，或者它正在不停地旋转？

您是否也只是想解释一下为什么您的并行实现比顺序实现稍慢，或者您想问什么是快速高效地处理这种工作负载的最佳技术？跨度>

【参考方案1】：

您的代码存在一些问题。首先，您不需要在读取或设置本地标志时锁定集合。其次，我认为你把事情复杂化了。只需一次加载一百万条记录，然后并行处理它们，再加载一百万条记录，等等...... System.Collections.Concurrent 命名空间提供（顾名思义）适合并行执行的集合。它们中的每一个都实现了内部分区，以避免必须手动管理锁。但是你甚至需要一个队列吗？ System.Threading.Tasks.Parallel 提供了一个 .ForEach 实现来并行遍历集合。无需锁或特殊收藏：

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading.Tasks;

...

        var batchSize = 1000000;
        using (var file = new StreamReader(@"C:\somefile.log"))
        
            var done = false;

            while (!done)
            
                var cnt = batchSize;
                var lines = new List<string>();

                while (cnt > 0)
                
                    var line = file.ReadLine();
                    if (string.IsNullOrEmpty(line))
                    
                        done = true;
                        break;
                    
                    else
                    
                        lines.Add(line);
                        cnt -= 1;
                                            
                

                Parallel.ForEach(lines, 
                    line =>
                    
                        ProcessLine(line);
                    );

                lines.Clear();
                lines = null;
            

            file.Close();
        

当然，您可以使用线程或任务异步启动整个事情。

但请注意，您无法保证使用此类并行处理处理项目的顺序。如果处理记录的顺序很重要，那么您需要将十亿条记录划分为实际数量的记录（将记录数除以 Environment.ProcessorCount - 1），然后您可以在单独的线程中处理，然后按顺序重组。

【参考方案2】：

将 100 万条记录读入队列听起来会占用大量内存。我可以想象你也在使用大量的文本操作，这也是内存密集型的，并且会使垃圾收集器和 CPU 非常繁忙。

我建议使用专用线程尽可能快地读取文件，但可能会将其限制为一次实际的 1000 条记录。然后一个单独的线程池或 Paralel.For 循环配置 MaxDegreeOfParallelism 来处理用于处理行条目的总线程。

由于您的线程已经处理了数据，它们应该向文件读取器发出信号，从要处理的文件中获取更多记录。

底线是十亿行包含大量数据和记录。您需要将其分解为小块以有效地处理它。

快速而有效的解决方案。 如果您不需要按顺序处理这些行，那么您可以将一个完整的单独应用程序作为第 1 步，将海量文件拆分为多个 100 万行的文件（使其可配置），每个文件的大小。然后将其放入另一个文件夹中，您的原始应用程序将在该文件夹中拾取并处理它。您甚至可以启动原始应用的多个实例来同时处理文件。

【讨论】：

Quick and Dirty 特性可能不合适，因为如果您知道自己在做什么，只需不到 10 行代码就可以在 10 分钟内实现高效的多线程解决方案.很难通过将数据拆分为多个文件并产生多个进程来击败它。 低技术和肮脏 恕我直言。 ?

@TheodorZoulias 确实，多线程会更高效。我添加了快速和肮脏的解决方案，作为对编码新手的开发人员的简单解决方案，而不是仅仅将它们扔到深处。

@DrunkenCodeMonkey 读取文件不是问题。它将资源分配给他想要使用每行数据执行的代码。 10 亿条记录是很多数据。例如，您是否曾在包含这么多记录的数据库上工作过？处理每一行的代码例程会花费很长时间，如果处理不当会消耗大量时间和系统资源。

前几天我忘记删除这条评论了，我原本误以为是100万条记录...我在答案中编辑了代码，一次读取和处理100万条记录，但没有'不要删除此评论。