ConcurrentQueueとParallel.ForEach

Question

私はソースを取得する必要があるURLのリストを持つConcurrentQueueを持っています。 ConcurrentQueueオブジェクトを入力パラメーターとして使用してParallel.ForEachを使用する場合、Popメソッドは何も動作しません（文字列を返す必要があります）。

私はMaxDegreeOfParallelismを4に設定してParallelを使用しています。私は本当に同時スレッドの数をブロックする必要があります。並列処理のキューを冗長使用していますか？

ありがとうございます。

// 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

あなたがするForEachメソッドに提供アクションでポップを呼び出すべきではありませんので、問題は、CrawlItems方法です。その理由は、アクションがポップされたアイテムごとに呼び出されているため、アイテムが既にポップされているためです。これは、アクションに 'item'引数がある理由です。

ForEachメソッドによって、他のスレッドによって既にすべてのアイテムがポップされているため、nullが返ってきたとします。

したがって、あなたのコードは次のようになります。あなたがやろうとしているすべての最初のシングルスレッドからそれにアイテムを追加して、それを反復処理する場合

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

Answer 2

あなたはConcurrentQueue<T>は必要ありません。 Parallel.ForEach()。通常はList<T>で十分です。

はまた、ItemQueueの実装は非常に疑わしいです：

• それはConcurrentQueue<string>から継承し、また別のConcurrentQueue<string>が含まれています。それはあまり意味をなさない、混乱して非効率的です。

• ConcurrentQueue<T>のメソッドは、スレッドセーフであるように非常に注意深く設計されました。 Pop()はスレッドセーフではありません。何が起きるかは、Countを確認してから、TryDequeue()に電話し、2つの呼び出しの間に別のスレッドがキューからアイテムを削除したため、値を取得できません（つまりvalueはnull）。