关于处理巨型文本文件和处理 URL 的建议答案

【问题标题】：Advice on processing giant text file and processing URL's关于处理巨型文本文件和处理 URL 的建议
【发布时间】：2014-10-01 12:38:15
【问题描述】：

我目前正在尝试遍历一个大小约为 1.5gb 的文本文件，然后使用从中获取的 URL 从站点中提取 html。

为了速度，我试图在一个新线程上处理所有 HTTP 请求，但由于 C# 不是我最强大的语言，而是我正在做的事情的要求，所以我对良好的线程实践有点困惑。

这就是我处理列表的方式

private static void Main()
    {
        const Int32 BufferSize = 128;
        using (var fileStream = File.OpenRead("dump.txt"))
        using (var streamReader = new StreamReader(fileStream, Encoding.UTF8, true, BufferSize))
        {
            String line;
            var progress = 0;

            while ((line = streamReader.ReadLine()) != null)
            {
                var stuff = line.Split('|');
                getHTML(stuff[3]);

                progress += 1;
                Console.WriteLine(progress);
            }
        }
    }

我正在拉下 HTML

 private static void getHTML(String url)
    {
        new Thread(() =>
        {
            var client = new DecompressGzipResponse();
            var html = client.DownloadString(url);

        }).Start();
    }

虽然最初这样做的速度很快，但在大约 20000 之后它们会变慢，最终在 32000 之后应用程序将挂起并崩溃。我的印象是函数完成时 C# 线程终止了？

任何人都可以就如何更好地做到这一点提供任何示例/建议吗？

【问题讨论】：

20k 线程在任何语言中都是一个坏主意。您不能仅仅创建更多线程并获得性能的线性增益，这不是 CPU 的工作方式。此外，您的线程不会产生任何东西。他们只是下载一个字符串，该字符串会立即被丢弃，因为它是函数本地的。您需要以合理的大小批量处理这些请求。
您可能希望考虑限制活动线程的数量。您可以读取文件并创建新线程，速度远远快于它们完成的速度。
@EdS。出于测试目的，我无论如何都没有返回字符串，只是通过创建我每秒能够处理几千个的线程来查看下载 html 的请求可以多快，而不用线程化它不合理的慢；然而。正如我在最初的帖子中所说，我认为他们会在功能完成后自我毁灭。你有关于如何解决这个问题的代码示例吗？
@user3037561：线程在退出后被销毁，这不是你的问题。您的问题是同时启动如此大量的它们。不要期望它们以同步方式运行和完成。你在折腾。线程是昂贵的资源。
@EdS。据我所知，但不幸的是，我不知道如何以符合 c# 标准或至少产生高质量结果的方式为该任务编写稳定的代码。

标签： c# multithreading thread-safety large-files

【解决方案1】：

一种非常可靠的方法是使用生产者-消费者模式。您创建一个线程安全的 URL 队列（例如，BlockingCollection<Uri>）。您的主线程是生产者，它将项目添加到队列中。然后，您有多个使用者线程，每个线程从队列中读取 Urls 并执行 HTTP 请求。见BlockingCollection。

设置起来并不难：

BlockingCollection<Uri> UrlQueue = new BlockingCollection<Uri>();

// Main thread starts the consumer threads
Task t1 = Task.Factory.StartNew(() => ProcessUrls, TaskCreationOptions.LongRunning);
Task t2 = Task.Factory.StartNew(() => ProcessUrls, TaskCreationOptions.LongRunning);
// create more tasks if you think necessary.

// Now read your file
foreach (var line in File.ReadLines(inputFileName))
{
    var theUri = ExtractUriFromLine(line);
    UrlQueue.Add(theUri);
}

// when done adding lines to the queue, mark the queue as complete
UrlQueue.CompleteAdding();

// now wait for the tasks to complete.
t1.Wait();
t2.Wait();
// You could also use Task.WaitAll if you have an array of tasks

各个线程使用此方法处理 url：

void ProcessUrls()
{
    foreach (var uri in UrlQueue.GetConsumingEnumerable())
    {
        // code here to do a web request on that url
    }
}

这是一种简单可靠的做事方式，但速度不是特别快。您可以通过使用发出异步请求的 WebCient 对象的第二个队列来做得更好例如，假设您想要有 15 个异步请求。以 BlockingCollection 开始，但您只有一个持久的消费者线程。

const int MaxRequests = 15;
BlockingCollection<WebClient> Clients = new BlockingCollection<WebClient>();

// start a single consumer thread
var ProcessingThread = Task.Factory.StartNew(() => ProcessUrls, TaskCreationOptions.LongRunning);

// Create the WebClient objects and add them to the queue
for (var i = 0; i < MaxRequests; ++i)
{
    var client = new WebClient();
    // Add an event handler for the DownloadDataCompleted event
    client.DownloadDataCompleted += DownloadDataCompletedHandler;
    // And add this client to the queue
    Clients.Add(client);
}

// add the code from above that reads the file and populates the queue

你的处理功能有些不同：

void ProcessUrls()
{
    foreach (var uri in UrlQueue.GetConsumingEnumerable())
    {
        // Wait for an available client
        var client = Clients.Take();
        // and make an asynchronous request
        client.DownloadDataAsync(uri, client);
    }
    // When the queue is empty, you need to wait for all of the
    // clients to complete their requests.
    // You know they're all done when you dequeue all of them.
    for (int i = 0; i < MaxRequests; ++i)
    {
        var client = Clients.Take();
        client.Dispose();
    }
}

您的DownloadDataCompleted 事件处理程序对下载的数据进行处理，然后将WebClient 实例添加回客户端队列。

void DownloadDataCompleteHandler(Object sender, DownloadDataCompletedEventArgs e)
{
    // The data downloaded is in e.Result
    // be sure to check the e.Error and e.Cancelled values to determine if an error occurred

    // do something with the data

    // And then add the client back to the queue
    WebClient client = (WebClient)e.UserState;
    Clients.Add(client);
}

这应该可以让您继续处理 15 个并发请求，这几乎是您可以做的所有事情，而不会变得更复杂。您的系统可能可以处理更多的并发请求，但是WebClient 启动异步请求的方式需要预先进行一些同步工作，而这种开销使您可以处理的最大数量约为 15。

您可能可以让多个线程启动异步请求。在这种情况下，您可能拥有与处理器内核一样多的线程。所以在四核机器上，你可以有主线程和三个消费者线程。使用三个消费者线程，此技术可以为您提供 45 个并发请求。我不确定确定它是否可以很好地扩展，但它可能值得一试。

有数百个并发请求的方法，但实现起来相当复杂。

【讨论】：

【解决方案2】：

你需要线程管理。

我的建议是使用Tasks 而不是创建自己的线程。

通过使用任务并行库，您可以让运行时处理线程管理。默认情况下，它将在 ThreadPool 中的线程上分配您的任务，并允许一定程度的并发，这取决于您拥有的 CPU 内核数量。它还会在现有线程可用时重用它们，而不是浪费时间创建新线程。

如果您想更高级，可以创建自己的任务调度程序来自己管理调度方面。

另见What is difference between Task and Thread?

【讨论】：