ConcurrentQueue with multithreading

Question

私はマルチスレッドの概念に慣れていません。特定の数の文字列をキューに追加し、それらを複数のスレッドで処理する必要があります。スレッドセーフであるConcurrentQueueを使用する。

これは私が試みたものです。ただし、並行キューに追加されたアイテムはすべて処理されません。最初の4項目のみが処理されます。

class Program 
{ 
    ConcurrentQueue<string> iQ = new ConcurrentQueue<string>(); 
    static void Main(string[] args) 
    { 
     new Program().run(); 
    } 

    void run() 
    { 
     int threadCount = 4; 
     Task[] workers = new Task[threadCount]; 

     for (int i = 0; i < threadCount; ++i) 
     { 
      int workerId = i; 
      Task task = new Task(() => worker(workerId)); 
      workers[i] = task; 
      task.Start(); 
     } 

     for (int i = 0; i < 100; i++) 
     { 
      iQ.Enqueue("Item" + i); 
     } 

     Task.WaitAll(workers); 
     Console.WriteLine("Done."); 

     Console.ReadLine(); 
    } 

    void worker(int workerId) 
    { 
     Console.WriteLine("Worker {0} is starting.", workerId); 
     string op; 
     if(iQ.TryDequeue(out op)) 
     { 
      Console.WriteLine("Worker {0} is processing item {1}", workerId, op); 
     } 

     Console.WriteLine("Worker {0} is stopping.", workerId); 
    } 


} 

Answer 1

実装にはいくつか問題があります。最初と明らか一つはworker方法が唯一、ゼロまたは1つのアイテムをデキューしてから停止していることである：

if(iQ.TryDequeue(out op)) 
    { 
     Console.WriteLine("Worker {0} is processing item {1}", workerId, op); 
    } 

それは次のようになります。しかし、あなたのプログラムを動作させるには十分ではありません

while(iQ.TryDequeue(out op)) 
    { 
     Console.WriteLine("Worker {0} is processing item {1}", workerId, op); 
    } 

正しく。あなたのワーカーがメインスレッドがエンキューするよりも速くデキューしている場合、メインタスクがまだエンキュー中に停止します。あなたが止めることができる労働者に知らせる必要があります。あなたが別の番号にTHREADCOUNTを変更した場合

void worker(int workerId) 
{ 
    Console.WriteLine("Worker {0} is starting.", workerId); 
    do { 
     string op; 
     while(iQ.TryDequeue(out op)) 
     { 
      Console.WriteLine("Worker {0} is processing item {1}", workerId, op); 
     } 
     SpinWait.SpinUntil(() => Volatile.Read(ref doneEnqueueing) || (iQ.Count > 0)); 
    } 
    while (!Volatile.Read(ref doneEnqueueing) || (iQ.Count > 0)) 
    Console.WriteLine("Worker {0} is stopping.", workerId); 
} 

Answer 2

あなたの労働者がqueueが空になるまで、ちょうどそれらが動作させ、queueのうちの一つのアイテムを取ると、その後の作業を終えます。

あなたはすべてのアイテムに近い2人の労働者によって処理されることを、あなたが表示されます、それを実行するとwhile

void worker(int workerId) 
{ 
    Console.WriteLine("Worker {0} is starting.", workerId); 
    string op; 
    while (iQ.TryDequeue(out op)) 
    { 
     Console.WriteLine("Worker {0} is processing item {1}", workerId, op); 
    } 

    Console.WriteLine("Worker {0} is stopping.", workerId); 
} 

でワーカー機能でifを交換してください。理由：あなたのCPUには2つのコアがあり、どちらも動作しており、新しいタスクを作成するための「フリーティームスロット」はありません。あなたがアイテムを処理するために、すべての4タスクを持っているしたい場合は、、のようなものをanotehrタスクを作成するには、お使いのプロセッサ時間を与えるために遅延を追加することができます。あなたが望むことを、あなたの出力が得られます

while (iQ.TryDequeue(out op)) 
{ 
    Console.WriteLine("Worker {0} is processing item {1}", workerId, op); 
    Task.Delay(TimeSpan.FromMilliseconds(1)).Wait(); 
} 

：

... 
Worker 0 is processing item Item8 
Worker 1 is processing item Item9 
Worker 2 is processing item Item10 
Worker 3 is processing item Item11 
Worker 3 is processing item Item13 
Worker 1 is processing item Item12 
...