💹 💃🏾 😔 प्रक्रिया प्रबंधक को लागू करने के लिए मासट्रांसिट, सागा और रैबिटएमक्यू 😼 🤧 🌙

एक बार, हमने एक बड़ी कंपनी में विभिन्न वर्कफ़्लोज़ को स्वचालित करने के कार्य का सामना किया। हमारे लिए, इसका मतलब लॉन्च के समय लगभग 10 सिस्टम को एक साथ रखना था। इसके अलावा, सब कुछ अतुल्यकालिक रूप से, स्केलेबल, मज़बूती से जुड़ा होना था।

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

इस समस्या को हल करने के लिए, हमने डेटा बस के माध्यम से मैसेजिंग आर्किटेक्चर का उपयोग करने का फैसला किया, और रैबिटएमक्यू के साथ अपनी गाथा के साथ मासट्रांसिट पूरी तरह से हमारे अनुकूल है।

जैसे सागा क्या है?

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

सागा आटोमेटिक बेनामी पर आधारित है। यह घोषित तौर पर MassTransitStateMachine <> से विरासत में मिली एक कक्षा में वर्णित है। सागा के लिए, आपको कुछ घटनाओं के होने पर सभी राज्यों, ली गई घटनाओं और किए गए कार्यों का वर्णन करना होगा। वर्तमान स्थिति डेटाबेस में संग्रहीत है।

सबसे पहले, हम सागा के सभी राज्यों और घटनाओं का वर्णन करते हैं और उन्हें समझने योग्य नाम देते हैं। यह इस तरह दिखता है:

public sealed partial class StateMachine { public State AwaitingTaskCreated { get; set; } public State AwaitingTaskTakedToWork { get; set; } public State AwaitingDecisionAboutTask { get; set; } public State Rejected { get; set; } public Event<IStartWorkflowCommand> StartWorkflowCommandReceived { get; set; } public Event<TaskCreatedNotification> TaskCreated { get; set; } public Event<TaskTakedToWorkNotification> TaskTakedToWork { get; set; } public Event<TaskDeclinedNotification> TaskDeclined { get; set; } public Event<TaskApprovedNotification> TaskApproved { get; set; } private void BuildStateMachine() { InstanceState(x => x.CurrentState); Event(() => StartWorkflowCommandReceived, x => x.CorrelateById(ctx => ctx.Message.CorrelationId) .SelectId(context => context.Message.CorrelationId)); Event(() => TaskCreated, x => x.CorrelateById(ctx => ctx.Message.CorrelationId)); Event(() => TaskTakedToWork, x => x.CorrelateById(ctx => ctx.Message.CorrelationId)); Event(() => TaskDeclined, x => x.CorrelateById(ctx => ctx.Message.CorrelationId)); Event(() => TaskApproved, x => x.CorrelateById(ctx => ctx.Message.CorrelationId)); } }

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

इस प्रकार, यदि कोई समस्या उत्पन्न होती है, तो हम सभी कनेक्टेड सिस्टम से लॉग द्वारा क्या हो रहा है की पूरी तस्वीर को पुनर्स्थापित कर सकते हैं। हम सागा से संदेशों में CorrelationId भेजते हैं, सिस्टम इसे सूचनाओं में वापस भेजते हैं ताकि हम एक विशिष्ट गाथा के साथ संदेश को सहसंबंधित कर सकें।

यहाँ स्टेट मशीन क्लास का ही एक उदाहरण दिया गया है:

 public sealed partial class StateMachine : MassTransitStateMachine<WorkflowSaga> { public StateMachine() { BuildStateMachine(); Initially(WhenStartWorkflowCommandReceived()); During(AwaitingTaskCreatedInPlanner, WhenTaskCreated()); During(AwaitingTaskTakedToWork, WhenTaskTakedToWork()); During(AwaitingDecisionAboutTask, WhenTaskApproved(), WhenTaskDeclined()); } private EventActivityBinder<WorkflowSaga, IStartWorkflowCommand> WhenStartWorkflowCommandReceived() { return When(StartWorkflowCommandReceived) .Then(ctx => ctx.Instance.SaveConfigurationRequestInfo(ctx.Data)) .Send(TaskManagerQueue, ctx => new CreateTaskCommand(ctx.Instance)) .TransitionTo(AwaitingTaskCreated); } private EventActivityBinder<WorkflowSaga, TaskCreatedNotification> WhenTaskCreated() { return When(DPORPApproveTaskCreatedInPlanner) .Then(ctx => ctx.Instance.SaveCreatedTaskInfo(ctx.Data)) .Send(MailServiceQueue, ctx => new NotifyRequestAuthorThatWorkflowStarted(ctx.Instance)) .TransitionTo(AwaitingTaskTakedToWork); } private EventActivityBinder<WorkflowSaga, TaskTakedToWorkNotification> WhenTaskTakedToWork() { return When(TaskTakedToWork) .Then(ctx => ctx.Instance.MarkTaskAsTakedToWork(ctx.Data)) .TransitionTo(AwaitingDecisionAboutTask); } private EventActivityBinder<WorkflowSaga, TaskApprovedNotification> WhenTaskApproved() { return When(TaskApproved) .Then(ctx => ctx.Instance.MarkTaskAsApproved(ctx.Data)) .Finalize(); } private EventActivityBinder<WorkflowSaga, TaskDeclinedNotification> WhenTaskDeclined() { return When(TaskDeclined) .Then(ctx => ctx.Instance.MarkTaskAsDeclined(ctx.Data)) .TransitionTo(Rejected); } }

कंस्ट्रक्टर राज्यों का वर्णन करता है। पठनीयता बनाए रखने के लिए प्रत्येक घटना का रिसेप्शन अलग तरीके से बनाया गया है। संदेशों के निर्माण का पूरा तर्क स्वयं संदेशों में किया जाता है, अन्यथा, सिस्टम की बढ़ती जटिलता के साथ, सागा काफी तेज़ी से सूज जाता है।

सम्मेलनों को विकसित करने और पठनीयता बनाए रखने में सावधानी बरतनी चाहिए। C # की अपूर्णता के कारण, इसमें राज्यों और कार्यों का विवरण घोषित करना बहुत मुश्किल है। यहां तक कि सरल राज्य मशीनों के लिए, असली नरक शुरू होता है।

अब कुछ शब्द SagaInstance के बारे में। SagaInstance एक वर्ग है जिसे SagaStateMachineInstance से विरासत में मिला है। इसमें ऑब्जेक्ट्स और फ़ील्ड शामिल होते हैं जो राज्य मशीन की विशेषता रखते हैं। मोटे तौर पर, यह गाथा की स्मृति है। हम सभी सागा डेटा संग्रहीत करते हैं जिसकी उसे जीवन भर आवश्यकता होगी। साथ ही इस वर्ग में कार्य के दौरान इस डेटा में परिवर्तनों के तर्क का वर्णन किया गया है।

यहाँ एक उदाहरण है:

 public class WorkflowSaga : SagaStateMachineInstance , ISagaWithState , ICreatedOnOffset , IModifiedOnOffset , ICreatedBy<string> , IModifiedBy<string> { public Guid CorrelationId { get; set; } public string CurrentState { get; set; } public string InitialRequestViewUrl { get; set; } public string RequestNumber { get; set; } public string RequestAuthor { get; set; } public string RequestText { get; set; } public byte[] RowVersion { get; set; } public string CreatedBy { get; set; } public string ModifiedBy { get; set; } public DateTimeOffset CreatedOn { get; set; } public DateTimeOffset ModifiedOn { get; set; } public DateTimeOffset CompletedOn { get; set; } public virtual ICollection<RelatedTask> RelatedTasks { get; set; } public void SaveGabrielConfigurationRequestInfo( ICreateGabrielConfigurationRequestCommand command) { CorrelationId = command.CorrelationId; RequestNumber = command.RequestNumber; RequestAuthor = command.Author; RequestText = command.RequestText; InitialRequestViewUrl = command.InitialRequestViewUrl; CreatedOn = RuntimeContext.Current.DateTimeOffset.Now; } public void SaveCreatedTaskInfo(ITaskCreationInfo taskCreationInfo) { RelatedPlannerTasks.Add(new RelatedPlannerTask(taskCreationInfo)); } public void MarkTaskAsTakedToWork(ITaskUpdatedInfo taskInfo) { UpdateTaskInfo(taskInfo, TaskStatus.TakedToWork); } public void MarkTaskAsApproved(TaskApprovedNotification taskInfo) { UpdateTaskInfo(taskInfo, TaskStatus.Completed, taskInfo.Comment); CompletedOn = RuntimeContext.Current.DateTimeOffset.Now; } public void MarkTaskAsDeclined(TaskDeclinedNotification taskInfo) { UpdateTaskInfo(taskInfo, TaskStatus.Declined, taskInfo.Comment); CompletedOn = RuntimeContext.Current.DateTimeOffset.Now; } private void UpdateTaskInfo(ITaskUpdatedInfo taskInfo, TaskStatus taskStatus, string comment = null) { var task = RelatedTasks.Single(t => t.Number == taskInfo.Number); task.ModifiedBy = taskInfo.TaskModifiedBy; task.Comment = comment; task.Status = taskStatus; } }

एक उदाहरण से पता चलता है कि SagaInstance CorrelationId में Saga और CurrentState के साथ घटनाओं के सहसंबंध के लिए संग्रहीत किया जाता है, जो Saga की वर्तमान स्थिति को संग्रहीत करने के लिए है।

हैंडलिंग में त्रुटि

संदेश प्रसंस्करण के दौरान त्रुटि होने पर सागा का क्या होता है? यह एक महत्वपूर्ण मुद्दा है, क्योंकि हर कोई चाहता है कि राज्य मशीन हमेशा संगत रहे, भले ही कुछ गलत हो गया हो। और मासट्रांसिट के साथ, सागा ठीक कर रहा है।

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

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

बेशक, अधिक जटिल परिदृश्य हैं जब आपको अपवादों को संभालने के लिए कुछ क्षतिपूर्ति करने वाले कार्यों को करने की आवश्यकता होती है। फिर "। कैच" स्टेट मशीन हैंडलर का उपयोग एक निश्चित प्रकार के अपवाद को पकड़ने के लिए किया जाता है और फिर क्षतिपूर्ति तर्क को निष्पादित किया जाता है।

और अगर आपको केवल एक अपवाद प्रतिज्ञा करने की आवश्यकता है, तो एक पर्यवेक्षक (ऑब्जर्वर) का उपयोग करना बेहतर है।

अब उस स्थिति की कल्पना करें, जिसे हमने संदेश प्रसंस्करण के दौरान पहले ही भेजें कमांड निष्पादित कर दिया है, जिसके बाद एक अपवाद हुआ है। इस चरण पर भेजे गए आदेश का क्या होगा? सब के बाद, सब कुछ जो उड़ गया है उसे वापस नहीं किया जा सकता है? लेकिन यहां सब कुछ सोच समझ कर किया जाता है।

बस को कॉन्फ़िगर करते समय, आप UseInMemoryOutbox विकल्प को सक्षम कर सकते हैं। यह विकल्प आपको वर्तमान चरण पूरा होने तक संदेश नहीं भेजने की अनुमति देता है। यदि अपवाद होता है, तो संदेश बिल्कुल नहीं भेजे जाएंगे। यहाँ प्रलेखन से एक अंश है:

 /// <summary> /// Includes an outbox in the consume filter path, which delays outgoing messages until the return path /// of the pipeline returns to the outbox filter. At this point, the message execution pipeline should be /// nearly complete with only the ack remaining. If an exception is thrown, the messages are not sent/published. /// </summary> /// <param name="configurator">The pipe configurator</param> public static void UseInMemoryOutbox(this IConsumePipeConfigurator configurator)

परीक्षण

पहली नज़र में, अतुल्यकालिक राज्य मशीन का परीक्षण अभी भी एक खुशी है। लेकिन यहां सब कुछ ठीक है। मासट्रांसिट एक अच्छा परीक्षण लेखन ढांचा प्रदान करता है जो राज्य मशीन के परीक्षण के लिए हमारी सभी आवश्यकताओं को पूरी तरह से पूरा करता है।

ढांचा डेटा बस (InMemoryTestHarness) के कार्यान्वयन के साथ InMemory प्रदान करता है, जो आपको RabbitMQ या अन्य कतार को दरकिनार करके संदेश भेजने और प्राप्त करने की अनुमति देता है।

खैर, एक उदाहरण के रूप में:

 [TestFixture] public class SagaTests : TestFixtureBase { protected const string HostName = "HostName"; protected InMemoryTestHarness Harness; protected StateMachine StateMachine; protected StateMachineSagaTestHarness<GabrielConfigurationRequestSaga, StateMachine> Saga; [SetUp] public void SetUp() { StateMachine = (StateMachine)Kernel. Get<MassTransitStateMachine<WorkflowSaga>>(); Harness = new InMemoryTestHarness(HostName); Saga = Harness .StateMachineSaga<WorkflowSaga, StateMachine>(StateMachine); } [TearDown] public async Task TearDown() { await Harness.Stop(); } protected async Task<WorkflowSaga> InitializeSaga() { await Harness.Start(); var command = new TestStartWorkflowCommand { CorrelationId = SagaId, Author = RequestAuthor, InitialRequestViewUrl = InitialRequestViewUrl, RequestText = RequestText, RequestNumber = RequestNumber, }; await Harness.InputQueueSendEndpoint .Send<IStartWorkflowCommand>(command); //    ,  consume    , // ,  Saga  ,    Assert.IsTrue(Harness.Consumed .Select<IStartWorkflowCommand>().Any()); var currentSaga = Saga.Created.Contains(SagaId); currentSaga.RelatedPlannerTasks = new List<RelatedPlannerTask>(); return currentSaga; } [Test] public async Task CheckCurrntStateWhenStartWorkflowCommand() { var saga = await InitializeSaga(); Assert.IsNotNull(saga); Assert.AreEqual(StateMachine .AwaitingORDTApproveTaskCreatedInPlanner.Name, saga.CurrentState); } } public class WhenTaskCreated : SagaTestsBase { private async Task<WorkflowSaga> InitializeState() { var saga = await InitializeSaga(true); saga.CurrentState = StateMachine.AwaitingTaskCreated.Name; InitializeRelatedTask(saga); await SendTaskCreatedNotification(); Assert.IsTrue(Harness.Consumed .Select<TaskCreatedNotification>().Any()); return saga; } [Test] public async Task SaveWorkflowDataWhenTaskCreated() { var saga = await InitializeState(); var taskInfo = saga.RelatedPlannerTasks .First(task => task.PlannerTaskType == PlannerTaskType.DPORPApprove); Assert.AreEqual(TaskNumber, taskInfo.Number); Assert.AreEqual(TaskUrl, taskInfo.TaskUrl); Assert.AreEqual(SagaId, taskInfo.SagaCorrelationId); Assert.AreEqual(TaskStatus.Created, taskInfo.Status); Assert.AreEqual(User, taskInfo.ModifiedBy); Assert.AreEqual(saga.CurrentState, StateMachine.AwaitingTaskTakedToWork.Name); } [Test] public async Task SendsMailWhenTaskCreated() { var mailConsumer = Harness .Consumer<MockConsumer<ISendEmailMessageWithTemplateCommand>> (RabbitMqRouting.QueueNames .SendEmailsQueueName); await InitializeState(); Assert.IsTrue(mailConsumer.Consumed .Select<ISendEmailMessageWithTemplateCommand>().Any()); } private async Task SendTaskCreatedNotification() { await Harness.InputQueueSendEndpoint .Send(new TaskCreatedNotification { TaskUrl = TaskUrl, Number = TaskNumber, TaskModifiedBy = User, CorrelationId = SagaId }); } }

टेस्ट काफी तेज दौड़ते हैं। उदाहरण के लिए, एक एकल डेवलपर के कंप्यूटर पर, 850 परीक्षणों में लगभग 21 सेकंड लगते हैं।

उपयोगी सुझाव

निष्कर्ष में, हम अपने अनुभव के आधार पर उपयोगी सुझावों की एक सूची देते हैं।

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

अपने सिस्टम में लॉग और मॉनिटर सेट अप करें, यह कभी भी चोट नहीं पहुंचाएगा। इस लेख में हमारा अनुभव।

प्रक्रिया प्रबंधक को लागू करने के लिए मासट्रांसिट, सागा और रैबिटएमक्यू

जैसे सागा क्या है?

हैंडलिंग में त्रुटि

परीक्षण

उपयोगी सुझाव

More articles: