C＃producer/consumer/observer？

Question

特定の種類のオブジェクトがあることを除いて、プロデューサ/コンシューマキューがあります。だから、どんな消費者も追加されたオブジェクトを消費することはできません。あまりにも多いので、それぞれのタイプに対して特定のキューを作成したくありません。 （これはプロデューサ/コンシューマの定義を拡張していますが、正しい用語がわかりません）

パラメータを持つパルスを許可するEventWaitHandleのようなものはありますか？例えばmyHandle.Set(AddedType = "foo")。今私はMonitor.Waitを使用していますが、各消費者がパルスが実際に意図されているかどうかをチェックしますが、それは無意味です。

私が今持っているもののpseduocodeバージョン：

class MyWorker { 
    public string MyType {get; set;} 
    public static Dictionary<string, MyInfo> data; 

    public static void DoWork(){ 
     while(true){ 
      if(Monitor.Wait(data, timeout)){ 
        if (data.ContainsKey(MyType)){ 
         // OK, do work 
        } 
      } 
     } 
    } 
} 

あなたが見ることができるように他のものが辞書に追加されたとき、私はパルスを得る可能性があります。 MyTypeがdictに追加されたときだけ気にします。それを行う方法はありますか？これは大したことではありませんが、たとえば、ロックを取得するたびにタイムアウト時間内に成功する可能性があるため、手動でタイムアウトを処理する必要があります。MyTypeは、timeout内のdictには追加されません。

Answer 1

これは興味深い質問です。ソリューションの鍵はpriority queueのブロックバリアントのようです。 JavaにはPriorityBlockingQueueがありますが、残念ながら.NET BCLに相当するものは存在しません。しかし、いったんそれを持っていれば、実装は簡単です。

class MyWorker 
{ 
    public string MyType {get; set;} 
    public static PriorityBlockingQueue<string, MyInfo> data; 

    public static void DoWork() 
    { 
     while(true) 
     { 
      MyInfo value; 
      if (data.TryTake(MyType, timeout, out value)) 
      { 
       // OK, do work 
      } 
     } 
    } 
} 

PriorityBlockingQueueの実装はそれほど難しくありません。 AddとTakeスタイルのメソッドを使用してBlockingCollectionと同じパターンに従って、私は次のコードを思いついた。

public class PriorityBlockingQueue<TKey, TValue> 
{ 
    private SortedDictionary<TKey, TValue> m_Dictionary = new SortedDictionary<TKey,TValue>(); 

    public void Add(TKey key, TValue value) 
    { 
     lock (m_Dictionary) 
     { 
      m_Dictionary.Add(key, value); 
      Monitor.Pulse(m_Dictionary); 
     } 
    } 

    public TValue Take(TKey key) 
    { 
     TValue value; 
     TryTake(key, TimeSpan.FromTicks(long.MaxValue), out value); 
     return value; 
    } 

    public bool TryTake(TKey key, TimeSpan timeout, out TValue value) 
    { 
     value = default(TValue); 
     DateTime initial = DateTime.UtcNow; 
     lock (m_Dictionary) 
     { 
      while (!m_Dictionary.TryGetValue(key, out value)) 
      { 
       if (m_Dictionary.Count > 0) Monitor.Pulse(m_Dictionary); // Important! 
       TimeSpan span = timeout - (DateTime.UtcNow - initial); 
       if (!Monitor.Wait(m_Dictionary, span)) 
       { 
        return false; 
       } 
      } 
      m_Dictionary.Remove(key); 
      return true; 
     } 
    } 
} 

これは迅速な実装であり、いくつかの問題があります。まず、私は全くテストしていません。次に、基本的なデータ構造として赤黒のツリー（SortedDictionary経由）を使用します。つまり、TryTakeメソッドはO（log（n））の複雑さを持ちます。優先度キューには、通常、O（1）の削除の複雑さがあります。優先順位キューの一般的なデータ構造はheapですが、私はskip listsが実際にはいくつかの理由で実際に優れていることがわかりました。これらのどちらも.NET BCLには存在しません。その理由は、このシナリオではパフォーマンスが劣るにもかかわらずSortedDictionaryを代わりに使用したためです。

ここでは、実際には無意味なWait/Pulseの動作を解決していないことを指摘しておきます。これは単にPriorityBlockingQueueクラスにカプセル化されています。しかし、これは少なくともあなたのコードの中核部分をきれいにするでしょう。

キーごとに複数のオブジェクトを処理するコードのようには表示されませんでしたが、辞書に追加するときには普通の古いMyInfoの代わりにQueue<MyInfo>を使用して簡単に追加できます。

Answer 2

プロデューサ/コンシューマキューとObserverパターン（汎用コンシューマスレッドまたはキューからの読み取り）を組み合わせて、必要なコードにイベントを渡したいようです。この場合、実際にObserverに通知するのではなく、消費者スレッドが特定の作業項目に関心を持つ人物を特定したときに呼び出すだけです。

.NETのオブザーバーパターンは、通常、C＃イベントを使用して実装されます。オブジェクトのイベントハンドラを呼び出すだけで、1つまたは複数のオブザーバが呼び出されます。ターゲットコードは、まず、イベントの到着時に通知するためにイベントに自身を追加することによって、監視されたオブジェクトに自身を登録する必要があります。