c# - 使用 Rx 和 SelectMany 限制并发请求

Question

我有一个要使用 HttpClient 同时下载的页面的 URL 列表。 URL 列表可能很大(100 个或更多!)

我目前有这段代码:

var urls = new List<string>
            {
                @"http:\\www.amazon.com",
                @"http:\\www.bing.com",
                @"http:\\www.facebook.com",
                @"http:\\www.twitter.com",
                @"http:\\www.google.com"
            };

var client = new HttpClient();

var contents = urls
    .ToObservable()
    .SelectMany(uri => client.GetStringAsync(new Uri(uri, UriKind.Absolute)));

contents.Subscribe(Console.WriteLine);

问题:由于SelectMany的使用，几乎同时创建了一大堆任务。似乎如果 URL 列表足够大，很多任务都会超时(我收到“任务已取消”异常)。

所以，我认为应该有一种方法，也许使用某种调度程序，来限制并发任务的数量，在给定时间不允许超过 5 或 6 个。

这样我就可以获得并发下载，而不会启动太多可能会停滞的任务，就像他们现在所做的那样。

如何做到这一点，才不会因为大量超时任务而饱和？

Answer 1

记住 SelectMany() 实际上是 Select().Merge()。虽然 SelectMany 没有 maxConcurrent 参数，Merge()做。所以你可以使用它。

根据您的示例，您可以这样做:

var urls = new List<string>
    {
        @"http:\\www.amazon.com",
        @"http:\\www.bing.com",
        @"http:\\www.facebook.com",
        @"http:\\www.twitter.com",
        @"http:\\www.google.com"
    };

var client = new HttpClient();

var contents = urls
    .ToObservable()
    .Select(uri => Observable.FromAsync(() => client.GetStringAsync(uri)))
    .Merge(2); // 2 maximum concurrent requests!

contents.Subscribe(Console.WriteLine);