C# 中的并发下载/处理答案

【问题标题】：Concurrent download/processing in C#C# 中的并发下载/处理
【发布时间】：2012-07-23 10:48:10
【问题描述】：

我正在寻找最快和最可靠的方法来同时使用 C# 下载 1000 个远程网页（使用 HttpWebRequest），将它们写入单个本地文件并在所有文件下载后运行一些处理代码，同时充分利用并行性和非阻塞并发可用。

服务器是一个四核 (vCPU) VPS，运行 Windows 2008 和 .NET 4.0（不能使用较新的 async/await 东西）。

你有什么建议？

更新：目前建议的选项有：反应式扩展 (Rx)、异步 CTP、TPL。

看起来 Async CTP 是理想的方法，其次是 Rx 和 TPL。伙计们怎么说？

【问题讨论】：

所有这些页面是来自一个网站还是来自 1000 个不同的网站？
@Ramhound 我不认为 WebClient 本身支持多线程。
@svick 所有页面来自不同的站点。
@Nick WebClient 有异步方法，不会占用额外的 ThreadPool 线程。
@Christoph / Ramhound 我想你们指的是 WebClient 异步方法，如下面的 James Cuthbert 示例所示stackoverflow.com/a/11611613/1157545

标签： c# multithreading concurrency parallel-processing task-parallel-library

【解决方案1】：

无论您最终使用哪种异步方法，请不要忘记您需要增加允许的最大连接数，因为默认值为 每个域 2 个。因此，如果您针对单个域进行大量调用，您的速率将受到限制。

您可以使用基本配置在独立（非 ASP.NET）应用程序中解决此问题：

<system.net>
   <connectionManagement>
       <add address="*" maxconnections="200" />
   </connectionManagement>
</system.net>

但是，如果您在 ASP.NET 中，这将无法按预期工作，因为默认的 <processModel autoConfig="true" ...> 属性将导致它自动配置为每个内核 12 个，虽然总数量超过 2 个，但仍可能不适合您的需求.因此，您将不得不在 Application_Start 之类的东西中使用基于代码的方法：

ServicePointManager.DefaultConnectionLimit = 200;

注意：这种基于代码的方法同样适用于非 ASP.NET 应用程序，因此如果您想避免使用 .config，可以将其用作“通用”解决方案。

【讨论】：

【解决方案2】：

我会使用 Rx 来完成这项任务。

string[] webpages = { "http://www.google.com", "http://www.spiegel.de"};

webpages
    .Select(w => FetchWebPage(w))
    .ForkJoin()
    .Subscribe(x => /*This runs when all webpages have been fetched*/  Console.WriteLine(x));

或者，如果您想控制并发性以按照 svick 的建议同时处理最多 4 个请求，您可以将其更改为：

Observable.ForkJoin(
    webpages
        .Select(w => FetchWebPage(w))
        .Merge(4))
        .Subscribe(x => /*This runs when all webpages have been fetched*/  Console.WriteLine(x));

您还需要一个辅助方法来从常规异步方式转换为 Rx 方式

public static IObservable<string> FetchWebPage(string address)
{
    var client = new WebClient();

    return Observable.Create<string>(observer =>
    {
        DownloadStringCompletedEventHandler handler = (sender, args) =>
        {
            if (args.Cancelled)
                observer.OnCompleted();
            else if(args.Error != null)
                observer.OnError(args.Error);
            else
            {
                observer.OnNext(args.Result);
                observer.OnCompleted();
            }
        };

        client.DownloadStringCompleted += handler;

        try
        {
            client.DownloadStringAsync(new Uri(address));
        }
        catch (Exception ex)
        {
            observer.OnError(ex);
        }

        return () => client.DownloadStringCompleted -= handler;
    });
}

【讨论】：

你能修改这个来限制并行度吗？因为我不认为同时开始 1000 次下载是个好主意。
如果我没记错的话，Rx 内部使用了 TPL。看起来 TPL 提供了比 Rx 更直接的异步操作控制。
@Nick：Rx 使用自己的内在函数（调度程序等）。我不相信它是建立在 TPL 之上的。
Observable.ForkJoin() 似乎只存在于 Windows Phone 的 Rx 中，而不是普通的 Rx。我对吗？那么你会如何使用普通的 Rx 来编写呢？
不，它是最新的测试版。但是，如果缺少 ForkJoin 替代方案，可以轻松编写它。我现在在移动设备上，但这里是如何用 RxJS 编写它。它与 C# 中的 github.com/cburgdorf/rxjs-contribute#L82 几乎相同

【解决方案3】：

VS2010 SP1 可以使用异步 CTP 在 .NET 4.0 上执行 async/await。 VS2012 RC 可以使用 Async Targeting Pack 在 .NET 4.0 上做async/await。

但如果你真的不想使用async/await，你仍然可以使用任务和延续（任务并行库是 .NET 4.0 的一部分）。

【讨论】：

【解决方案4】：

我也有类似的需求，但对我来说，网址数超过 7,000（过去大约需要 25 到 28 分钟才能完成）。对于我的解决方案，我使用了 TPL。由于每个 URL 没有依赖关系，因此很容易将每个 URL 封装在一个对象中，将其放入一个集合中，然后将该集合传递给 Parallel.ForEach() 调用。

每次下载完成后，我们都会查看页面内容，并根据我们找到的内容，将其发送出去进行额外处理。

正如我所说，这过去需要半小时才能完成，但现在运行时间约为 4.5 分钟（我有双四核 Xeon 处理器 @ 3GHz、Windows 7 Ultimate 64 位版和 24 GB 的 RAM .... 大量的电力正在被利用，而不是大部分被浪费掉）。

Microsoft 的 TPL 给我留下了深刻的印象，以至于我回到了我的大部分遗留项目/代码并重构了设计以尽可能利用 TPL，并且我总是对我的任何新代码进行“TPL 处理”写（如果你在循环迭代之间有任何类型的依赖，这并不总是可能的）。

【讨论】：

【解决方案5】：

我最近使用 C# 5 的新异步功能和 WebClent 而不是 HttpWebRequest 做了类似的事情。您可以使用 WebClient 获得一些不错的异步方法，例如 DownloadDataTaskAsync。

WebClient client = new WebClient();
byte[] data = await client.DownloadDataTaskAsync(url)

【讨论】：

这可能很高兴知道，但它不能回答问题。特别是因为OP说他不能使用async-await。