ConcurrentQueue und Parallel.ForEach

Question

Ich habe eine ConcurrentQueue mit einer Liste von URLs, die ich brauche, um die Quelle zu erhalten. Wenn Sie das Parallel.ForEach mit dem ConcurrentQueue-Objekt als den Eingabeparameter verwenden, wird die Pop-Methode nichts (sollte eine Zeichenfolge zurückgeben) nicht funktionieren.

Ich verwende Parallel mit der MaxDegreeOfParallelism auf vier festgelegt. Ich muss wirklich die Anzahl der gleichzeitigen Threads blockieren. Wird eine Warteschlange mit Parallelität redundant verwendet?

Vielen Dank im Voraus.

// On the main class 
var items = await engine.FetchPageWithNumberItems(result); 
// Enqueue List of items 
itemQueue.EnqueueList(items); 
var crawl = Task.Run(() => { engine.CrawlItems(itemQueue); }); 

// On the Engine class 
public void CrawlItems(ItemQueue itemQueue) 
{ 
Parallel.ForEach(
      itemQueue, 
      new ParallelOptions {MaxDegreeOfParallelism = 4}, 
      item => 
      { 

       var worker = new Worker(); 
       // Pop doesn't return anything 
       worker.Url = itemQueue.Pop(); 
       /* Some work */ 
      }); 
} 

// Item Queue 
class ItemQueue : ConcurrentQueue<string> 
    { 
     private ConcurrentQueue<string> queue = new ConcurrentQueue<string>(); 

     public string Pop() 
     { 
      string value = String.Empty; 
      if(this.queue.Count == 0) 
       throw new Exception(); 
      this.queue.TryDequeue(out value); 
      return value; 
     } 

     public void Push(string item) 
     { 
      this.queue.Enqueue(item); 
     } 

     public void EnqueueList(List<string> list) 
     { 
      list.ForEach(this.queue.Enqueue); 
     } 
    } 

Answer 1

Das Problem ist mit CrawlItems Methode, da Sie nicht Pop in der Aktion der ForEach-Methode zur Verfügung gestellt nennen sollen. Der Grund dafür ist, dass die Aktion bei jedem Popup-Objekt aufgerufen wird, daher wurde das Element bereits geblockt. Dies ist der Grund, dass die Aktion ein "Element" -Argument hat.

Ich nehme an, dass Sie null erhalten, da alle Elemente bereits von der ForEach-Methode von den anderen Threads abgerufen wurden.

Daher sollten Sie den Code wie folgt aussehen:

public void CrawlItems(ItemQueue itemQueue) 
{ 
    Parallel.ForEach(
     itemQueue, 
     new ParallelOptions {MaxDegreeOfParallelism = 4}, 
     item => 
     { 
      worker.Url = item; 
      /* Some work */ 
     }); 
} 

Answer 2

Sie nicht ConcurrentQueue<T> brauchen, wenn alles, was Sie tun werden, ist zum ersten Elemente aus einem einzigen Thread hinzufügen und wiederholen sie dann in Parallel.ForEach(). Ein normaler List<T> würde dafür ausreichen.

Auch die Implementierung von ItemQueue ist sehr verdächtig:

• Aus ConcurrentQueue<string> erbt und enthält auch eine andere ConcurrentQueue<string>. Das macht wenig Sinn, ist verwirrend und ineffizient.

• Die Methoden auf ConcurrentQueue<T> wurden sehr sorgfältig entwickelt, um fadensicher zu sein. Ihre Pop() ist nicht threadsicher. Was passieren könnte ist, dass Sie Count überprüfen, es ist 1, dann rufen Sie TryDequeue() und keinen Wert erhalten (d. H. value wird null sein), weil ein anderer Thread das Element aus der Warteschlange in der Zeit zwischen den beiden Aufrufen entfernt.