C# producent/konsument/obserwator?

Question

Mam kolejkę producenta/konsumenta, z wyjątkiem tego, że istnieją określone typy obiektów. Więc nie tylko każdy konsument może spożywać dodany obiekt. Nie chcę tworzyć specyficznej kolejki dla każdego typu, ponieważ jest ich zbyt wiele. (W pewnym sensie rozciąga się definicja producenta/konsumenta, ale nie jestem pewien, jaki jest właściwy termin.)

Czy istnieje coś takiego jak EventWaitHandle, która pozwala na impulsy z parametrem? na przykład myHandle.Set(AddedType = "foo"). Teraz używam Monitor.Wait, a następnie każdy konsument sprawdza, czy puls rzeczywiście był dla nich przeznaczony, ale wydaje się to bezsensowne.

Wersja pseduocode co mam teraz:

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

Jak widać, mogę dostać impulsy gdy inny materiał zostanie dodany do dict. Zastanawiam się tylko, kiedy MyType zostanie dodany do dyktatury. Czy jest jakiś sposób na zrobienie tego? Nie jest to wielka sprawa, ale na przykład muszę ręcznie obsłużyć limity czasu, ponieważ każde wejście blokady może zakończyć się w określonym czasie, ale MyType nigdy nie jest dodawane do dyktatury w ramach timeout.

Answer 1

To interesujące pytanie. Wygląda na to, że kluczem do rozwiązania jest blokujący wariant priority queue. Java ma numer PriorityBlockingQueue, ale niestety odpowiednik dla .NET BCL nie istnieje. Jednak gdy już to zrobisz, implementacja jest łatwa.

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

Implementacja PriorityBlockingQueue nie jest strasznie trudna. Podążając za tym samym wzorcem, co BlockingCollection, używając metod w stylu Add i Take, wymyśliłem następujący kod.

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; 
     } 
    } 
} 

To była szybka implementacja i ma kilka problemów. Po pierwsze, w ogóle go nie testowałem. Po drugie, wykorzystuje czerwono-czarne drzewo (przez SortedDictionary) jako podstawową strukturę danych. Oznacza to, że metoda TryTake będzie miała złożoność O (log (n)). Kolejki priorytetowe mają zazwyczaj złożoność usuwania O (1). Zazwyczaj wybraną strukturą danych dla kolejek priorytetowych jest heap, ale uważam, że skip lists są faktycznie lepsze w praktyce z kilku powodów. Żadne z nich nie istnieje w .NET BCL, dlatego użyłem zamiast tego SortedDictionary, pomimo jego gorszej wydajności w tym scenariuszu.

Chciałbym tutaj zaznaczyć, że to nie rozwiązuje właściwie bezsensownego zachowania Wait/Pulse. Jest on po prostu zamknięty w klasie PriorityBlockingQueue. Ale przynajmniej to z pewnością oczyści rdzenną część twojego kodu.

Nie wyglądało na to, że kod obsługiwał wiele obiektów na klucz, ale byłoby to łatwe do dodania przy użyciu Queue<MyInfo> zamiast zwykłego starego MyInfo podczas dodawania do słownika.

Answer 2

Wygląda na to, że chcesz połączyć kolejkę producenta/konsumenta z wzorcem Obserwatora - ogólny wątek konsumenta lub wątki odczytuje z kolejki, a następnie przekazuje zdarzenie do wymaganego kodu. W tym przypadku nie zasygnalizowałbyś obserwatora, ale po prostu zadzwoń, gdy wątek konsumenta wskaże, kto jest zainteresowany danym przedmiotem pracy.

Wzorzec obserwatora w .Net jest zwykle implementowany przy użyciu zdarzeń C#. Po prostu trzeba wywołać procedurę obsługi zdarzenia dla obiektu i jeden lub więcej obserwatorów zostanie wywołanych za jego pośrednictwem. Kod docelowy musiałby najpierw zarejestrować się w obserwowanym obiekcie, dodając się do zdarzenia w celu powiadomienia o nadejściu pracy.