Begrenzen gleichzeitiger Anfragen mit Rx und SelectMany

Question

Ich habe eine Liste von URLs von Seiten, die ich gleichzeitig herunterladen möchte mit HttpClient. Die Liste der URLs groß sein kann (100 oder mehr!)

Ich habe derzeit haben diesen Code:

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); 

Das Problem: aufgrund der Verwendung von SelectMany, ein großes Bündel von Aufgaben werden erstellt fast zur gleichen Zeit. Es scheint, dass, wenn die Liste der URLs groß genug ist, viele Aufgaben Timeouts geben (ich bekomme "Eine Aufgabe wurde abgebrochen" Ausnahmen).

Also dachte ich, es sollte einen Weg geben, vielleicht mit einer Art von Scheduler, um die Anzahl der gleichzeitigen Aufgaben zu begrenzen, nicht mehr als 5 oder 6 zu einem bestimmten Zeitpunkt zuzulassen.

Auf diese Weise konnte ich gleichzeitige Downloads erhalten, ohne zu viele Aufgaben zu starten, die möglicherweise zum Stillstand kommen, wie sie es gerade jetzt tun.

Wie geht das, damit ich nicht mit vielen zeitgesteuerten Aufgaben sättigen?

Vielen Dank.

Answer 1

Denken Sie daran, SelectMany() tatsächlich Select().Merge() ist. Während keinen maxConcurrent Parameter hat, funktioniert Merge(). So können Sie das verwenden.

Von Ihrem Beispiel können Sie dies tun:

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); 

Answer 2

Hier ist ein Beispiel dafür, wie Sie es mit dem DataFlow API tun können:

private static Task DoIt() 
{ 
    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(); 

    //Create a block that takes a URL as input 
    //and produces the download result as output 
    TransformBlock<string,string> downloadBlock = 
     new TransformBlock<string, string>(
      uri => client.GetStringAsync(new Uri(uri, UriKind.Absolute)), 
      new ExecutionDataflowBlockOptions 
      { 
       //At most 2 download operation execute at the same time 
       MaxDegreeOfParallelism = 2 
      }); 

    //Create a block that prints out the result 
    ActionBlock<string> doneBlock = 
     new ActionBlock<string>(x => Console.WriteLine(x)); 

    //Link the output of the first block to the input of the second one 
    downloadBlock.LinkTo(
     doneBlock, 
     new DataflowLinkOptions { PropagateCompletion = true}); 

    //input the urls into the first block 
    foreach (var url in urls) 
    { 
     downloadBlock.Post(url); 
    } 

    downloadBlock.Complete(); //Mark completion of input 

    //Allows consumer to wait for the whole operation to complete 
    return doneBlock.Completion; 
} 

static void Main(string[] args) 
{ 
    DoIt().Wait(); 
    Console.WriteLine("Done"); 
    Console.ReadLine(); 
} 

Answer 3

Können Sie, ob das hilft zu sehen?

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

var contents = 
    urls 
     .ToObservable() 
     .SelectMany(uri => 
      Observable 
       .Using(
        () => new System.Net.Http.HttpClient(), 
        client => 
         client 
          .GetStringAsync(new Uri(uri, UriKind.Absolute)) 
          .ToObservable()));