समांतर घटना ग्राहक .net 4.5

Question

मैं एक समांतर घटना ग्राहक बनाने की कोशिश कर रहा हूं। यह मेरा पहला प्रयास है:

using System; 
using System.Collections.Generic; 
using System.Threading.Tasks; 
using EventStore.ClientAPI; 

namespace Sandbox 
{ 
    public class SomeEventSubscriber 
    { 
     private Position? _latestPosition; 
     private readonly Dictionary<Type, Action<object>> _eventHandlerMapping; 
     private IEventStoreConnection _connection; 

     public Dictionary<Type, Action<object>> EventHandlerMapping 
     { 
      get { return _eventHandlerMapping; } 
     } 

     public SomeEventSubscriber() 
     { 
      _eventHandlerMapping = CreateEventHandlerMapping(); 
      _latestPosition = Position.Start; 
     } 

     public void Start() 
     { 
      ConnectToEventstore(); 
     } 

     private void ConnectToEventstore() 
     { 
      _connection = EventStoreConnectionWrapper.Connect(); 
      _connection.Connected += 
      (sender, args) => _connection.SubscribeToAllFrom(_latestPosition, false, EventOccured, LiveProcessingStarted, HandleSubscriptionDropped); 
     } 

     private Dictionary<Type, Action<object>> CreateEventHandlerMapping() 
     { 
      return new Dictionary<Type, Action<object>> 
      { 
       {typeof (FakeEvent1), o => Handle(o as FakeEvent1)}, 
       {typeof (FakeEvent2), o => Handle(o as FakeEvent2)}, 
      }; 
     } 

     private async Task Handle(FakeEvent1 eventToHandle) 
     { 
      SomethingLongRunning(eventToHandle); 
     } 

     private async Task Handle(FakeEvent2 eventToHandle) 
     { 
      SomethingLongRunning(eventToHandle); 
     } 

     private async Task SomethingLongRunning(BaseFakeEvent eventToHandle) 
     { 
      Console.WriteLine("Start Handling: " + eventToHandle.GetType()); 
      var task = Task.Delay(10000); 
      await task; 
      Console.WriteLine("Finished Handling: " + eventToHandle.GetType()); 
     } 

     private void EventOccured(EventStoreCatchUpSubscription eventStoreCatchUpSubscription, 
      ResolvedEvent resolvedEvent) 
     { 
      if (resolvedEvent.OriginalEvent.EventType.StartsWith("$") || resolvedEvent.OriginalEvent.EventStreamId.StartsWith("$")) 
       return; 

      var @event = EventSerialization.DeserializeEvent(resolvedEvent.OriginalEvent); 
      if (@event != null) 
      { 
       var eventType = @event.GetType(); 
       if (_eventHandlerMapping.ContainsKey(eventType)) 
       { 
        var task = Task.Factory.StartNew(() => _eventHandlerMapping[eventType](event)); 
        Console.WriteLine("The task is running asynchronously..."); 
       } 
      } 
      if (resolvedEvent.OriginalPosition != null) _latestPosition = resolvedEvent.OriginalPosition.Value; 
     } 

     private void HandleSubscriptionDropped(EventStoreCatchUpSubscription subscription, SubscriptionDropReason dropReason, Exception ex) 
     { 
      if (dropReason == SubscriptionDropReason.ProcessingQueueOverflow) 
      { 
       //TODO: Wait and reconnect probably with back off 
      } 

      if (dropReason == SubscriptionDropReason.UserInitiated) 
       return; 

      if (SubscriptionDropMayBeRecoverable(dropReason)) 
      { 
       Start(); 
      } 
     } 

     private static bool SubscriptionDropMayBeRecoverable(SubscriptionDropReason dropReason) 
     { 
      return dropReason == SubscriptionDropReason.Unknown || dropReason == SubscriptionDropReason.SubscribingError || 
        dropReason == SubscriptionDropReason.ServerError || dropReason == SubscriptionDropReason.ConnectionClosed; 
     } 

     private static void LiveProcessingStarted(EventStoreCatchUpSubscription eventStoreCatchUpSubscription) 
     { 

     } 
    } 
} 

अपने विशेषज्ञ राय में, यह एक वैध दृष्टिकोण है? क्या आप किसी भी सुधार का सुझाव दे सकते हैं?

पुनश्च:

शायद:

Task.Run(() => _eventHandlerMapping[eventType](@event)); 

बेहतर होगा?

Answer 1

आप एक ही EventOccured प्रतिनिधि है जो जहां सभी घटनाओं है कि में पाए जाते हैं के लिए सूचित किया चाहते हैं EventStore
सबसे पहले जिस पर घटनाओं निकाल दिया जाता है के अलावा किसी अन्य डिस्पैचर में EventOccured अंदर पूर्व कोड चलाने पर विचार।
दूसरा, FakeEventBase के कार्यान्वयन के साथ इसे abstract class पर बदलना संभव होगा और फिर इसे बढ़ाएं और प्रत्येक FakeEvent प्रकार के लिए व्यक्तिगत उदाहरण बनाएं। यह बहुत साफ समाधान होगा।
तीसरे, क्यूइंग के लिए कस्टम ThreadScheduler पर विचार करें और इन Handle कार्यों को चलाएं। http://msdn.microsoft.com/en-us/library/system.threading.tasks.taskscheduler(v=vs.110).aspx

संपादित करें:
मैं एक प्रसारक वर्ग जब आपरेशन पूरा कर लिया है जानता है और समाप्त घटना को जन्म देती है जो नीचे की तरह होगा।

public class EventBroadcaster 
{ 
    public event EventHandler SomeEventOccured; 

    public async void DoLongRunningOperationAndRaiseFinishedEvent() 
    { 
     var waitingTask = Task.Delay(TimeSpan.FromSeconds(2)); 

     await waitingTask.ContinueWith(t => RaiseSomeEventOccured(), CancellationToken.None, 
      TaskContinuationOptions.OnlyOnRanToCompletion, TaskScheduler.Current); 
    } 

    private void RaiseSomeEventOccured() 
    { 
     EventHandler handler = SomeEventOccured; 
     if (handler != null) handler(this, EventArgs.Empty); 
    } 
} 

और फिर एक EventListener

public class EventListner 
{ 
    private readonly string _id; 

    public EventListner(string id) 
    { 
     _id = id; 
    } 

    public void ListenTo(EventBroadcaster broadcaster) 
    { 
     broadcaster.SomeEventOccured += OnSomeEventOccured; 
    } 

    private async void OnSomeEventOccured(object sender, EventArgs eventArgs) 
    { 
     var currentTime = DateTime.Now; 
     Console.WriteLine("EventListner {0} received at {1}", _id, 
      currentTime.ToString("dd-MM-yyyy HH:mm:ss.fffffff")); 

     //Not required just to show this does not affect other instances. 
     //await Task.Delay(TimeSpan.FromSeconds(5)); 
    } 
} 

तो यह

public static class Program 
{ 
    public static void Main(string[] args) 
    { 
     var broadcaster = new EventBroadcaster(); 

     var listners = new List<EventListner>(); 

     for (int i = 1; i < 10; i++) 
     { 
      var listner = new EventListner(i.ToString(CultureInfo.InvariantCulture)); 
      listner.ListenTo(broadcaster); 
      listners.Add(listner); 
     } 

     broadcaster.DoLongRunningOperationAndRaiseFinishedEvent(); 

     Console.WriteLine("Waiting for operation to complete"); 

     Console.ReadLine(); 

    } 
} 

परीक्षण इस उदाहरण हैंडलर प्रतिनिधियों के क्रम में एक के बाद एक के सक्रिय होने में लिए Program.cs होगा वे सब्सक्राइब किए गए थे।

अब की तरह नीचे नोट कुछ करने के लिए प्रसारणकर्ता में कोड को संशोधित: मैं कोडिंग में आसानी के लिए Action को EventHandler से विधि हस्ताक्षर बदल दिया है।

private void RaiseSomeEventOccured() 
    { 
     Action handler = SomeEventOccured; 
     if (handler != null) 
     { 
      var parallelOption = new ParallelOptions() { MaxDegreeOfParallelism = Environment.ProcessorCount }; 
      Parallel.Invoke(parallelOption, Array.ConvertAll(handler.GetInvocationList(), ConvertToAction)); 
      handler(); 
     } 
    } 

    private Action ConvertToAction(Delegate del) 
    { 
     return (Action)del; 
    } 

अब आप कि घटनाओं यादृच्छिक क्रम में निकाल दिया जाता है देखेंगे। TPL और Parallel प्रोग्रामिंग आप यकीन है कि यह के लिए जाने से पहले एक लाभ है बनाने की जरूरत है साथ हमेशा:
मैं विकल्प का उपयोग 1.
नोट बेहतर प्रदर्शन मिला है।

Answer 2

मुझे वास्तव में समांतर घटना ग्राहक बनाने में कोई बिंदु नहीं दिखता है (यदि मैं आपके इरादे को सही समझता हूं - समानांतर में ईवेंट हैंडलर चलाने में सक्षम होने के लिए, सामान्य ईवेंट के साथ एक के बाद नहीं)।

यह बहुत अधिक स्पष्ट अगर ईवेंट हैंडलर ही यह पता चलता है समानांतर में चलाने के लिए आशय व्यक्त करने के लिए है।

कुछ (बहुत आदिम) की तरह कुछ।

void SomeEventHandler(object sender, EventArgs e) 
{ 
    Task.Run(() => 
    { 
     ... // some code to run in parallel 
    }); 
} 

आप प्रबंधक की तरह बना सकते हैं (ईमानदारी से, मैं कोई सुराग नहीं कैसे सभी कोर पर कब्जा करने के लिए है, लेकिन मुझे नहीं लगता इस जटिल है, मैं सिर्फ यह करने के लिए की आवश्यकता नहीं है), लेकिन कृपया, सामान्य घटनाओं को रखें।