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

Question

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

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

该文件有大约 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 万条记录/进程并以这种方式继续，直到不读取整个文件。事实证明，顺序方式比多线程方式更快。我在多线程代码中做错了吗？

Answer 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），然后您可以在单独的线程中处理，然后按顺序重组。

Answer 2

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

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

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

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

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