CQRS + इवेंट सोर्सिंग का परिचय: भाग 2

पिछले लेख में, मैंने CQRS + इवेंट सोर्सिंग की मूल बातें शुरू की। इस बार मैं ईएस को जारी रखने और एक करीब से देखने का प्रस्ताव करता हूं।

उदाहरण में, जो मैंने अपने पिछले लेख के साथ अपलोड किया था, इवेंट सोर्सिंग का जादू IRepository अमूर्त और दो तरीकों IRepository.Save () और IRepository.GetById <> () के पीछे छिपा हुआ था।
अधिक विस्तार से समझने के लिए कि क्या हो रहा है, मैंने रिनैट अब्दुलिन से लोकद आईडीडीडी नमूना परियोजना के उदाहरण का उपयोग करके इवेंट स्टोर से यूनिट को बचाने और लोड करने की प्रक्रिया के बारे में बात करने का फैसला किया। उनकी एप्लिकेशन सेवाओं में, अतिरिक्त अमूर्तता के बिना, इवेंट स्टोर के लिए एक सीधी अपील है, इसलिए सब कुछ बहुत स्पष्ट दिखता है। एप्लीकेशन सर्विस कमांडहैंडलर का एक एनालॉग है, लेकिन जो एक यूनिट के सभी कमांड को प्रोसेस करता है। यह दृष्टिकोण बहुत सुविधाजनक है और हमने इसे एक परियोजना में बदल दिया है।

ApplicationService

IApplicationService इंटरफ़ेस बेहद सरल है।

public interface IApplicationService { void Execute(ICommand cmd); } 

Execute पद्धति में, हम कोई भी कमांड पास करते हैं और हम आशा करते हैं कि सेवा उन्हें वांछित हैंडलर पर पुनर्निर्देशित करेगी।

चूंकि रिनाट के उदाहरण में केवल एक ग्राहक समुच्चय है, इसलिए सेवा में केवल एक ही ग्राहकअनुप्रयोग सेवा है। दरअसल, इसलिए, किसी तर्क को आधार वर्ग में स्थानांतरित करने की आवश्यकता नहीं है। मेरी राय में एक उदाहरण के लिए महान समाधान।

एक्सैच्यूट विधि कमांड के प्रसंस्करण को पास करता है जब हस्ताक्षर से मेल खाता ओवरलोड हो जाता है। एक्सक्यूट विधि का कार्यान्वयन वक्ताओं का उपयोग करके बहुत सरल है, और आपको सभी तरीकों पर प्रतिबिंब चलाने की आवश्यकता नहीं है।

 public void Execute(ICommand cmd) { // pass command to a specific method named when // that can handle the command ((dynamic)this).When((dynamic)cmd); } 

चलिए Create कमांड - CreateCustomer से शुरू करते हैं।

 [Serializable] public sealed class CreateCustomer : ICommand { public CustomerId Id { get; set; } public string Name { get; set; } public Currency Currency { get; set; } public override string ToString() { return string.Format("Create {0} named '{1}' with {2}", Id, Name, Currency); } } 

एक वास्तविक परियोजना में, आपके पास UI और ApplicationService के बीच एक संदेश कतार होगी, लेकिन उदाहरण के लिए, Rinat ने खुद को सीधे सेवा एप्लिकेशन ऑब्जेक्ट (क्लास ApplicationServer देखें) के लिए एक कमांड भेजने के लिए सीमित कर दिया।
CreateCustomer कमांड Execute मेथड में प्रवेश करने के बाद, इसे व्हेन मेथड पर रीडायरेक्ट किया जाता है।

 public void When(CreateCustomer c) { Update(c.Id, a => a.Create(c.Id,c.Name, c.Currency, _pricingService, DateTime.UtcNow)); } 

अपडेट विधि में, हम एग्रीगेट की पहचानकर्ता और एक्शन को पास करते हैं, जो एग्रीगेट की स्थिति को बदलने का तरीका कहता है। सामान्य तौर पर, मेरी राय में, एग्रीगेट पर क्रिएट मेथड बनाना बेहतर नहीं है, लेकिन दूसरे कंस्ट्रक्टर को बनाना, क्योंकि इस संदर्भ में क्रिएट मेथड को कॉल करना थोड़ा अजीब लगता है। हम एक समुच्चय का निर्माण करते दिखते हैं, लेकिन किसी कारण से हम राज्य को बदलने की एक विधि के रूप में विधि बनाएँ। डिजाइनर के साथ, यह नहीं हुआ होगा।

अपडेट विधि पर वापस लौटें, उसके पास निम्न कार्य हैं: 1) वर्तमान एग्रीगेट के लिए सभी घटनाओं को लोड करें, 2) एक एग्रीगेट बनाएं और लोड की गई घटनाओं का उपयोग करके अपने राज्य को पुनर्स्थापित करें, 3) एग्रीगेट पर पारित एक्शन एग्जीक्यूट एक्शन निष्पादित करें, और 4 नई घटनाओं को बचाने के लिए यदि कोई हो।

 void Update(CustomerId id, Action<Customer> execute) { // Load event stream from the store EventStream stream = _eventStore.LoadEventStream(id); // create new Customer aggregate from the history Customer customer = new Customer(stream.Events); // execute delegated action execute(customer); // append resulting changes to the stream _eventStore.AppendToStream(id, stream.Version, customer.Changes); } 

राज्य की वसूली

पिछले लेख में मैंने जो उदाहरण दिखाया था, उस एग्रीगेट की स्थिति को एग्रीगेट क्लास में निजी क्षेत्रों के रूप में संग्रहीत किया गया था। यदि आप स्नैपशॉट जोड़ना चाहते हैं, तो यह बहुत सुविधाजनक नहीं है किसी भी तरह हर बार राज्य को चूसना या प्रतिबिंब का उपयोग करना होगा। रिनट के पास अधिक सुविधाजनक दृष्टिकोण है - राज्य के पास ग्राहकस्टैट नामक एक अलग वर्ग है, जो समुच्चय से प्रक्षेपण विधियों को निकालना संभव बनाता है और स्नैपशॉट को सहेजना और लोड करना बहुत आसान है, हालांकि वे उदाहरण में नहीं हैं।
जैसा कि आप देख सकते हैं, एक ही एग्रीगेट की घटनाओं की सूची को कंस्ट्रक्टर को स्थानांतरित कर दिया जाता है, क्योंकि इसके राज्य को पुनर्स्थापित करने के लिए यह अनुमान लगाना मुश्किल नहीं है।
बदले में इकाई ग्राहक वसूली वर्ग के लिए राज्य की वसूली को दर्शाता है।

 /// <summary> /// Aggregate state, which is separate from this class in order to ensure, /// that we modify it ONLY by passing events. /// </summary> readonly CustomerState _state; public Customer(IEnumerable<IEvent> events) { _state = new CustomerState(events); } 

मैं संपूर्ण CustomerState वर्ग के लिए कोड दे दूंगा, मैं कुछ जब प्रक्षेपण विधियों को हटा दूंगा।

 /// <summary> /// This is the state of the customer aggregate. /// It can be mutated only by passing events to it. /// </summary> public class CustomerState { public string Name { get; private set; } public bool Created { get; private set; } public CustomerId Id { get; private set; } public bool ConsumptionLocked { get; private set; } public bool ManualBilling { get; private set; } public Currency Currency { get; private set; } public CurrencyAmount Balance { get; private set; } public int MaxTransactionId { get; private set; } public CustomerState(IEnumerable<IEvent> events) { foreach (var e in events) { Mutate(e); } } ... public void When(CustomerCreated e) { Created = true; Name = e.Name; Id = e.Id; Currency = e.Currency; Balance = new CurrencyAmount(0, e.Currency); } public void When(CustomerRenamed e) { Name = e.Name; } public void Mutate(IEvent e) { // .NET magic to call one of the 'When' handlers with // matching signature ((dynamic) this).When((dynamic)e); } } 

जैसा कि आप कंस्ट्रक्टर में देख सकते हैं, हम उत्तीर्ण घटनाओं पर चलते हैं और उन्हें म्यूट विधि से पास करते हैं, जो बदले में उन्हें एक उपयुक्त प्रक्षेपण विधि में बदल देती है।
आप देख सकते हैं कि सभी संपत्तियों में एक निजी सेटर विधि है, जो गारंटी देती है कि हम केवल इसी घटना को पारित करके बाहर से राज्य को बदल सकते हैं।

राज्य को बहाल कर दिया गया था, अब आप इसे बदलने का प्रयास कर सकते हैं। राज्य परिवर्तन विधि में थोड़ा पीछे चलते हैं। जब से मैंने CreateCustomer कमांड को समझना शुरू किया, हम यूनिट पर क्रिएट विधि को भी देखेंगे।

 public void Create(CustomerId id, string name, Currency currency, IPricingService service, DateTime utc) { if (_state.Created) throw new InvalidOperationException("Customer was already created"); Apply(new CustomerCreated { Created = utc, Name = name, Id = id, Currency = currency }); var bonus = service.GetWelcomeBonus(currency); if (bonus.Amount > 0) AddPayment("Welcome bonus", bonus, utc); } 

यह हमारे व्यावसायिक नियमों की जांच करने का स्थान है, क्योंकि हमारे पास इकाई की वर्तमान स्थिति तक पहुंच है। हमारे पास एक व्यावसायिक नियम है कि ग्राहक केवल एक बार बनाया जा सकता है (हालांकि, एक तकनीकी सीमा भी है), इसलिए जब हम पहले से निर्मित इकाई पर कॉल करने का प्रयास करते हैं, तो हम कार्रवाई को छोड़ देते हैं।
यदि सभी व्यावसायिक नियम संतुष्ट हैं, तो हम ग्राहक विधि से लागू पद्धति को पास करते हैं। लागू करने की विधि बहुत सरल है, यह घटना को केवल राज्य वस्तु पर ले जाती है और इसे वर्तमान परिवर्तनों की सूची में जोड़ देती है।

 public readonly IList<IEvent> Changes = new List<IEvent>(); readonly CustomerState _state; void Apply(IEvent e) { // pass each event to modify current in-memory state _state.Mutate(e); // append event to change list for further persistence Changes.Add(e); } 

ज्यादातर मामलों में, एक घटना एक एग्रीगेट के साथ प्रति ऑपरेशन होती है। लेकिन अभी बनाएँ विधि के मामले में, दो घटनाएँ हो सकती हैं।
जब हम कस्टमर क्रिएट ईवेंट को लागू करने की विधि में पास कर लेते हैं, तो हम वर्तमान ग्राहक में एक स्वागत योग्य बोनस जोड़ सकते हैं यदि व्यावसायिक नियम यह है कि बोनस राशि शून्य से अधिक होनी चाहिए। इस मामले में, AddPayment कुल विधि कहा जाता है, जो किसी चेक को एकत्र नहीं करता है, लेकिन केवल CustomerPaymentAdded घटना को ट्रिगर करता है।

 public void AddPayment(string name, CurrencyAmount amount, DateTime utc) { Apply(new CustomerPaymentAdded() { Id = _state.Id, Payment = amount, NewBalance = _state.Balance + amount, PaymentName = name, Transaction = _state.MaxTransactionId + 1, TimeUtc = utc }); } 

अब आपको नई ईवेंट्स को सहेजना है और उन्हें रीड मॉडल में प्रकाशित करना है। मुझे संदेह है कि यह अगली पंक्ति है।

 // append resulting changes to the stream _eventStore.AppendToStream(id, stream.Version, customer.Changes); 

सुनिश्चित करें ...

 public void AppendToStream(IIdentity id, long originalVersion, ICollection<IEvent> events) { if (events.Count == 0) return; var name = IdentityToString(id); var data = SerializeEvent(events.ToArray()); try { _appendOnlyStore.Append(name, data, originalVersion); } catch(AppendOnlyStoreConcurrencyException e) { // load server events var server = LoadEventStream(id, 0, int.MaxValue); // throw a real problem throw OptimisticConcurrencyException.Create(server.Version, e.ExpectedStreamVersion, id, server.Events); } // technically there should be a parallel process that queries new changes // from the event store and sends them via messages (avoiding 2PC problem). // however, for demo purposes, we'll just send them to the console from here Console.ForegroundColor = ConsoleColor.DarkGreen; foreach (var @event in events) { Console.WriteLine(" {0} r{1} Event: {2}", id,originalVersion, @event); } Console.ForegroundColor = ConsoleColor.DarkGray; } 

खैर, लगभग प्रसन्न। घटनाओं को क्रमबद्ध और केवल स्टोर में सहेजा जाता है (हम उन्हें हटाने और संशोधित करने नहीं जा रहे हैं)। लेकिन उन्हें रीड-मॉडल (प्रस्तुति स्तर पर) पर भेजने के बजाय, रिनैट उन्हें केवल कंसोल पर प्रदर्शित करता है।
हालांकि, उदाहरण के लिए, यह पर्याप्त है।
यदि आप यह देखना चाहते हैं कि यह संदेश कतार के साथ कैसे काम करेगा, तो आप मेरे पिछले लेख के गिठब पर एक उदाहरण देख सकते हैं। मैंने इसे थोड़ा बदल दिया और समाधान में इवेंट सोर्सिंग बुनियादी ढांचे का हिस्सा भी पेश किया। इस उदाहरण में, मैं बताना चाहता हूं कि आप स्नैपशॉट कैसे जोड़ सकते हैं।

स्नैपशॉट्स

यूनिट की स्थिति के मध्यवर्ती स्नैपशॉट लेने के लिए स्नैपशॉट की आवश्यकता होती है। यह हमें यूनिट के घटनाओं के पूरे प्रवाह को डाउनलोड करने की अनुमति नहीं देता है, लेकिन केवल उन लोगों के लिए जो हमने पिछले स्नैपशॉट बनाने के बाद हुए थे।
तो आइए कार्यान्वयन को देखें।
UserCommandHandler वर्ग में रिनैट के समान एक अपडेट विधि है, लेकिन मेरे लिए यह अभी भी स्नैपशॉट को सहेजता है और लोड करता है। हम हर 100 संस्करणों में स्नैपशॉट करते हैं।

 private const int SnapshotInterval = 100; 

सबसे पहले, हम भंडार से एक स्नैपशॉट बढ़ाते हैं, फिर हम इस स्नैपशॉट को बनाने के बाद होने वाली घटनाओं की धारा को लोड करते हैं। फिर हम यूनिट के डिजाइनर को यह सब देते हैं।

 private void Update(string userId, Action<UserAR> updateAction) { var snapshot = _snapshotRepository.Load(userId); var startVersion = snapshot == null ? 0 : snapshot.StreamVersion + 1; var stream = _eventStore.OpenStream(userId, startVersion, int.MaxValue); var user = new UserAR(snapshot, stream); updateAction(user); var originalVersion = stream.GetVersion(); _eventStore.AppendToStream(userId, originalVersion, user.Changes); var newVersion = originalVersion + 1; if (newVersion % SnapshotInterval == 0) { _snapshotRepository.Save(new Snapshot(userId, newVersion,user.State)); } } 

कंस्ट्रक्टर में, हम स्नैपशॉट से राज्य प्राप्त करने या स्नैपशॉट न होने पर एक नया राज्य बनाने का प्रयास करते हैं। प्राप्त स्थिति पर, हम सभी लापता घटनाओं को खो देते हैं, और परिणामस्वरूप हमें इकाई का वर्तमान संस्करण मिलता है।

 public UserAR(Snapshot snapshot, TransitionStream stream) { _state = snapshot != null ? (UserState) snapshot.Payload : new UserState(); foreach (var transition in stream.Read()) { foreach (var @event in transition.Events) { _state.Mutate((IEvent) @event.Data); } } } 

समुच्चय में हेरफेर करने के बाद, हम जाँचते हैं कि क्या कुल संस्करण एक अंतराल है जिसके माध्यम से हम स्नैपशॉट लेते हैं, और यदि ऐसा है, तो नए स्नैपशॉट को रिपॉजिटरी में सहेजें। UserCommandHandler से एग्रीगेट की स्थिति प्राप्त करने के लिए, हमें इसके लिए स्टेट प्रॉपर्टी को इंटरनल गेट्टर बनाना होगा।

यही सब है, अब हमारे पास स्नैपशॉट हैं, जिसने हमें बहुत तेजी से घटनाओं की स्थिति में इकाई की स्थिति को बहुत तेजी से बहाल करने की अनुमति दी है।

प्रतिक्रिया

यदि आप CQRS + ES के विषय में रुचि रखते हैं, तो कृपया बेझिझक टिप्पणी में सवाल पूछ सकते हैं। यदि आप हब पर कोई उर्फ ​​नहीं है, तो आप मुझे स्काइप पर भी लिख सकते हैं हाल ही में, चेल्याबिंस्क के एक दोस्त ने मेरे स्काइप पर दस्तक दी और उनके सवालों के कारण मुझे अगले लेख के लिए बहुत सारे विचार मिले। चूंकि मेरे पास अब अपने निपटान में अधिक खाली समय है, इसलिए मैं उन्हें अधिक बार लिखने की योजना बनाता हूं।