BlockingCollectionをキューとして使用しているコンシューマの非同期メソッドを待ちます。

Question

複数のWCFサービスからデータを受信し、その結果をいくつか処理した後、SignalRを使用して単一の接続でIISサーバーに転送します。

WCFサービスがプロデューサであり、SignalRを使用してデータを送信するクラスがコンシューマであるプロデューサコンシューマパターンを使用して実装しようとしました。キューの場合、私はBlockingCollectionを使用しました。

ただし、await/asyncを使用してコンシューマのデータを送信する場合、whileループは他のすべてのスレッドがデータをキューに追加するまで停止します。

私は実際にデータを送信するコードをTask.Delay(1000).Wait();またはawait Task.Delay(1000);に置き換えてテストしましたが、どちらも問題にはなりません。 単純なThread.Sleep(1000);はうまく動作しているようですが、非同期コードが問題だと思うようになります。

私の質問です：非同期コードがwhileループで完了しないようにする何かがありますか？私は何が欠けていますか？

new Thread(Worker).Start(); 

そしてコンシューマコード：

私はこのような消費者のスレッドを開始しています

浪費家が正しく指摘したように、（名前が示すように） BlockingCollectionだけのために意図されて

private void Worker() 
{ 
    while (!_queue.IsCompleted) 
    { 
     IMobileMessage msg = null; 
     try 
     { 
      msg = _queue.Take(); 
     } 
     catch (InvalidOperationException) 
     { 
     } 

     if (msg != null) 
     { 
      try 
      { 
       Trace.TraceInformation("Sending: {0}", msg.Name); 
       Thread.Sleep(1000); // <-- works 
       //Task.Delay(1000).Wait(); // <-- doesn't work 
       msg.SentTime = DateTime.UtcNow; 
       Trace.TraceInformation("X sent at {1}: {0}", msg.Name, msg.SentTime); 
      } 
      catch (Exception e) 
      { 
       TraceException(e); 
      } 
     } 
    } 
} 

Answer 1

ブロックコードで使用し、非同期コードではうまく動作しません。

BufferBlock<T>などの非同期互換のプロデューサ/コンシューマキューがあります。この場合、私はActionBlock<T>も良いだろうと思うだろう：

private ActionBlock<IMobileMsg> _block = new ActionBlock<IMobileMsg>(async msg => 
{ 
    try 
    { 
    Trace.TraceInformation("Sending: {0}", msg.Name); 
    await Task.Delay(1000); 
    msg.SentTime = DateTime.UtcNow; 
    Trace.TraceInformation("X sent at {1}: {0}", msg.Name, msg.SentTime); 
    } 
    catch (Exception e) 
    { 
    TraceException(e); 
    } 
}); 

これはあなたの全体の消費スレッドとメインループを置き換えます。