архитектура для одного производителя несколько потребителей

Question

Я занимаюсь проектированием графического интерфейса для отображения рыночных данных. В настоящее время у меня есть несколько окон, которые потребляют одни и те же данные, но отображают разные аспекты. Я пытаюсь найти лучший способ архитектовать эту систему.

У меня есть один производитель рыночных данных и несколько потребителей, которые регистрируют обратный вызов с производителем. Когда производитель готов с данными, он выполняет итерацию через список потребителей и распределяет данные каждому потребителю посредством обратного вызова. Я не уверен, что это лучший способ распространения данных. Каждому потребителю приходится ждать, пока предыдущий потребитель завершит обработку обратного вызова, прежде чем он сможет получить свои данные.

Есть ли способ, которым все потребители могут получать данные в одно и то же время или с минимальными задержками? Я использую C# 4.0 и хотел бы знать, есть ли какие-либо языковые возможности, которые позволяют это.

Answer 1

Вы можете найти довольно простое решение, используя отдельный BlockingCollection для каждого потребителя.

В потоке производителя требуется только List<> из BlockingCollection очередей, которыми он управляет. Когда поток производителя создает элемент, ему просто нужно добавить его в в очередь в списке, используя BlockingCollection.Add(). Примечание. То есть тот же элемент ставится в очередь несколько раз - один раз для каждой очереди потребителей.

Тип элемента должен быть неизменным ссылочным типом - тогда безопасно делиться между потребителями, а накладные расходы - это всего лишь ссылка в каждой очереди потребителей.

Это не самый эффективный маршрут, поскольку для каждого потребителя требуется очередь, но я не думаю, что это слишком плохо, если очереди не большие. Это очень легко понять и реализовать.

A BlockingCollection позволяет легко рассказать потребительские потоки, когда данных больше нет, так что они могут выйти чисто.

Нить производителя вызывает BlockingCollection.CompleteAdding(), чтобы сигнализировать потребляющему потоку, что они должны выйти, когда данных больше нет.

Между тем, все потребляющие нити должны сделать это (при условии, queue их очередь BlockingCollection)

foreach (var item in queue.GetConsumingEnumerable()) 
{ 
    ... process item 
} 

Вот это. Потребители автоматически блокируются в цикле , когда данных больше нет, и они автоматически выйдут из foreach, когда данных больше нет и производитель позвонил CompleteAdding().

Answer 2

Для этих сценариев вы должны использовать Очередь, такую ​​как RabbitMQ или MSMQ. Они оптимальны для этих типов операций издателя/подписчика.

Если это не вариант, вы можете активировать обратные вызовы с помощью потоков или установить их как огонь WCF и забыть вызовы (OneWay).

Answer 3

Если это важно, чтобы каждый из зарегистрированных обратных вызовов вызываются из их собственной резьбы (скорее всего, поток пул потоков), что достаточно легко сделать:

public class Foo 
{ 
    private HashSet<Action> callbacks = new HashSet<Action>(); 
    public event Action MyEvent 
    { 
     add 
     { 
      callbacks.Add(value); 
     } 
     remove 
     { 
      callbacks.Remove(value); 
     } 
    } 

    private void FireMyEvent() 
    { 
     foreach (var action in callbacks) 
     { 
      ThreadPool.QueueUserWorkItem(o => { action(); }); 
     } 
    } 
} 

Answer 4

Почему бы не использовать Reactive Extensions Framework?Это должна быть идея того, что вы описываете, и это дает большую гибкость. Как управлять multiple subscriptions