MassTransit। नेट के लिए RabbitMQ कतार सर्वर (MSMQ) के आधार पर मैसेजिंग के लिए सेवा बस

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

मासट्रांसिट क्या है।

MassTransit जाने-माने DataBus पैटर्न का एक कार्यान्वयन है। इस पैटर्न का मुख्य उद्देश्य कई वस्तुओं की बातचीत को व्यवस्थित करना है जो उनके बीच संदेशों के आदान-प्रदान के माध्यम से एक दूसरे के अस्तित्व के बारे में नहीं जानते हैं। लाइब्रेरी को ड्रू सेलर्स और क्रिस पैटरसन ने NServiceBus परियोजना के एक नि: शुल्क एनालॉग के रूप में लिखा था, जो विकल्प के लिए संदेश सर्वर परिवहन के रूप में RabbitMQ या MSMQ का उपयोग करने में सक्षम था। हमारे प्रोजेक्ट में, हमने RabbitMQ का उपयोग करना पसंद किया है, इसलिए यहां हम इस कतार सर्वर पर बस कॉन्फ़िगरेशन के लिए इंतजार कर रहे अनुभव और नुकसान का वर्णन करेंगे। इस तथ्य के बावजूद कि मासट्रांसिट एएमक्यूपी प्रोटोकॉल पर एक अमूर्त परत है और डेवलपर्स ने कार्यान्वयन विवरण छिपाने की कोशिश की ताकि कतार सर्वर डिवाइस और एएमक्यूपी प्रोटोकॉल का ज्ञान व्यावहारिक रूप से पुस्तकालय का उपयोग करने की आवश्यकता न हो, लेख को समझने और बस कॉन्फ़िगरेशन के दौरान रेक को सफलतापूर्वक पार करने के लिए, डिवाइस का एक सामान्य विचार। RabbitMQ सर्वर के लिए वांछनीय है। यह बुरी खबर है, लेकिन अच्छी खबर है - न्यूनतम ज्ञान की आवश्यकता है, यहां से पहले चार पाठों को पढ़ने के लिए पर्याप्त होगा। पाठ छोटे और समझ में आते हैं, RabbitMQ के साथ काम करने की मूल बातें सीखने में ज्यादा समय नहीं लगेगा, लेकिन यह कई लाभ ला सकता है। वैसे, पहले कुछ पाठों का अनुवाद भी हब्र पर किया गया था। पाठ एक और पाठ दो ।

व्यापार करने के लिए।

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


हम सीधे कोड पर आगे बढ़ते हैं। बस में भेजे गए संदेश साधारण डीटीओ ऑब्जेक्ट हैं। सबसे पहले, हमें संदेश का एक वर्ग बनाने की आवश्यकता है, जो प्रकाशक से ग्राहक को भेजा जाता है। चलो इसे KeyWasPressed कहते हैं।

public class KeyWasPressed { //       public ConsoleKey PressedKey { get; set; } } 

अब हम एक साधारण प्रकाशक (प्रकाशक) और ग्राहक (ग्राहक) लिखने के लिए आगे बढ़ते हैं। पुस्तकालय का एक प्रमुख तत्व सर्विसबस है। MassBransit में ServiceBus एक मैसेजिंग वातावरण है जिसमें संदेश परिवहन के लिए RabbitMQ (या MSMQ) कतार सर्वर जिम्मेदार है। हमारे सब्सक्राइबर और पब्लिशर इसके बहुत ही प्रकार के ऑब्जेक्ट होंगे - ServiceBus।

सब्सक्राइबर।

  IServiceBus subscriber = ServiceBusFactory.New(sbc => { //        rabbitMq sbc.UseRabbitMq(); //            sbc.ReceiveFrom("rabbitmq://localhost/subscriber"); //   KeyWasPressed.   //      sbc.Subscribe(subs => subs.Handler<KeyWasPressed>(msg => Console.WriteLine("{0}{1}{2}{3}",Environment.NewLine,"Key '", msg.PressedKey, "' was pressed") )); }); 

प्रकाशक।

 IServiceBus publisher = ServiceBusFactory.New(sbc => { //,        rabbitMq sbc.UseRabbitMq(); // ,       sbc.ReceiveFrom("rabbitmq://localhost/publisher"); }); 

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

 while (true) { publisher.Publish(new KeyWasPressed() { PressedKey = Console.ReadKey().Key }); } 

सब कुछ तैयार है, हमारे आवेदन को चलाएं और चाबियाँ दबाएं।

कतार सर्वर पर क्या होता है।

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


प्रत्येक संदेश जिसके लिए कम से कम एक ग्राहक मौजूद है, MT टेम्पलेट नाम के अनुसार एक नाम के साथ एक विनिमय बिंदु (एक्सचेंज) बनाता है: ClassName और उसके लिए ग्राहक कतारें बांधता है। हमारे आवेदन में, केवल एक विनिमय बिंदु है, जिसे MT कहा जाता है: KeyWasPressed, और यह विनिमय बिंदु एक कतार - ग्राहक से बंधा हुआ है।

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

नए सब्सक्राइबर जोड़ना।

चलिए KeyWasPressed मैसेज में एक और सब्सक्राइबर जोड़कर हमारे एप्लिकेशन को बदलते हैं। पहले के विपरीत, इसे एक कतार में सदस्यता लिया जाएगा जिसे एब्सबर्स्क्राइबर कहा जाता है और प्रत्येक उपयोगकर्ता-दबाए गए कुंजी का एक संख्यात्मक प्रतिनिधित्व प्रदर्शित करेगा।

 IServiceBus anotherSubscriber = ServiceBusFactory.New(sbc => { sbc.UseRabbitMq(); sbc.ReceiveFrom("rabbitmq://localhost/anothersubscriber"); sbc.Subscribe(subs => subs.Handler<KeyWasPressed>(msg => Console.WriteLine("{0}{1}{2}{3}", Environment.NewLine, "Key with code ", (int) msg.PressedKey, " was pressed") )); }); 

एप्लिकेशन लॉन्च करें और कुंजियों को धक्का दें।

अब, जब आप स्क्रीन पर प्रत्येक कुंजी दबाते हैं, तो दो लाइनें दिखाई देती हैं - प्रत्येक कुंजी के डिजिटल और प्रतीक कोड के साथ। यदि आप कतार सर्वर नियंत्रण कक्ष में जाते हैं, तो आप देख सकते हैं कि अब पहले से ही दो ग्राहक हैं और एमटी से जुड़ी एम्बरस्क्राइबर कतारें हैं: कीवसप्रेस्ड एक्सचेंज पॉइंट। और प्रत्येक को MT.KeyWasPressed प्रकार का संदेश प्राप्त हुआ, RabbitMQ कतार सर्वर दोनों कतारों में भेजता है।

संसाधन-गहन कार्यों का वितरण।

अब देखते हैं कि कैसे, MassTransit + RabbitMQ बंडल का उपयोग करके, आप कई प्रक्रियाओं के बीच संसाधन-गहन कार्यों को वितरित कर सकते हैं।
कल्पना करें कि हम वीडियो फ़ाइलों को परिवर्तित करने के लिए एक सेवा बनाने के कार्य के साथ सामना कर रहे हैं। इस कार्य के लिए, हमारे पास दो सर्वर हैं। जाहिर है, हमने पाया कि सर्वर नंबर एक के लिए इष्टतम लोड तीन समानांतर परिवर्तनीय वीडियो फाइलें हैं, सर्वर नंबर दो के लिए, एक साथ परिवर्तित वीडियो फ़ाइलों की संख्या पांच से अधिक नहीं होनी चाहिए। रूपांतरण प्रक्रिया, हम, निश्चित रूप से, अनुकरण करेंगे। कल्पना कीजिए कि हमारे पास filesToConvert नामक एक कतार है, जो रूपांतरण के लिए फ़ाइलें प्राप्त करता है। प्रत्येक फ़ाइल प्रकार VideoFile की एक वस्तु का प्रतिनिधित्व करेगी।

 public class VideoFile { public int Num { get; set; } //,       public int TimeToConvert { get; set; } } 

इस तरह के संदेश प्राप्त करने वाले ग्राहक, खेल के नियमों के अनुसार, प्राप्त संदेश के टाइमटॉन्कवर्ट क्षेत्र में निर्दिष्ट मिलीसेकंड की संख्या से सो जाते हैं।
किंवदंती के अनुसार, कोड, पहले सर्वर पर चल रहा है।

 int firstServerFilesCount = 0; IServiceBus firstServer = ServiceBusFactory.New(sbc => { sbc.UseRabbitMq(); //   ,      sbc.SetConcurrentConsumerLimit(3); sbc.Subscribe(subs => subs.Handler<VideoFile>(msg => { firstServerFilesCount++; Thread.Sleep(msg.TimeToConvert); Console.WriteLine(" 1. {0} {1}   {2} . : {3}  3. ThreadId - {4}", Environment.NewLine, msg.Num, msg.TimeToConvert, firstServerFilesCount, Thread.CurrentThread.ManagedThreadId); firstServerFilesCount--; })); //prefetch=3.   ,         sbc.ReceiveFrom("rabbitmq://localhost/filesToConvert?prefetch=3"); }); 

किंवदंती के अनुसार, हमने पहले सर्वर पर एक साथ पार्स किए गए संदेशों की संख्या को तीन तक सीमित करने का निर्णय लिया। इसलिए, हम तर्क 3 के साथ SetConcurrentConsumerLimit पद्धति को कॉल करते हैं। इसका मतलब है कि जब आप संदेश सर्वर पर पहली सर्वर ऑब्जेक्ट को कनेक्ट करते हैं, तो MassTransit सर्वर से संदेशों को संसाधित करने के लिए तीन थ्रेड्स का एक पूल तैयार रखेगा। लेकिन हमें यह याद रखना चाहिए कि RabbitMQ संदेशों के वितरण में लगा हुआ है, और वह इस तथ्य को नहीं जान सकता है कि FirstServer ऑब्जेक्ट एक बार में तीन संदेशों को पार्स करने के लिए तैयार है। हम इस जानकारी को उरी में प्रीफैच पैरामीटर निर्दिष्ट करके उसे पास कर सकते हैं, जिसके द्वारा फर्स्टसेवर संदेश सर्वर से जुड़ता है।

किंवदंती के अनुसार, कोड एक दूसरे सर्वर पर चल रहा है।

 int secondServerFilesCount = 0; IServiceBus secondServer = ServiceBusFactory.New(sbc => { sbc.UseRabbitMq(); //   ,      sbc.SetConcurrentConsumerLimit(5); sbc.Subscribe(subs => subs.Handler<VideoFile>(msg => { secondServerFilesCount++; Thread.Sleep(msg.TimeToConvert); Console.WriteLine(" 2. {0} {1}   {2} . : {3}  5. ThreadId - {4}", Environment.NewLine, msg.Num, msg.TimeToConvert, secondServerFilesCount, Thread.CurrentThread.ManagedThreadId); secondServerFilesCount--; })); //prefetch=3.   ,         sbc.ReceiveFrom("rabbitmq://localhost/filesToConvert?prefetch=5"); }); 

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

अब हम एक कोड लिखेंगे, जो एक सौ "वीडियो फ़ाइलों" के साथ यादृच्छिक द्वारा निर्दिष्ट किए गए एक यादृच्छिक रूपांतरण समय के साथ filesTonConvert कतार को भरता है।

 Random rnd = new Random(); for (int i = 1; i <= 100; i++) { publisher.Publish(new VideoFile() {Num = i, TimeToConvert = rnd.Next(100, 5000)}); } 

हम शुरू करते हैं और सुनिश्चित करते हैं कि हमारे सदस्यता सर्वर थ्रेड्स की संख्या का उपयोग करके समानांतर में काम करते हैं जिन्हें हमने सौंपा है।

मासट्रांसिट हमें कौन सी अन्य संभावनाएं प्रदान कर सकता है।

एक लेख के भीतर, मासट्रांसिट लाइब्रेरी द्वारा प्रस्तुत सभी संभावनाओं पर विचार करना असंभव है। लेकिन आप सूचीबद्ध कर सकते हैं (जो मैं करूंगा)।

• गाथा। वितरित प्रक्रियाओं के समन्वय के लिए तंत्र

• निर्धारण। Quartz.net लाइब्रेरी के साथ एकीकरण आपको निर्धारित कतारों में संदेश भेजने की अनुमति देता है

• निदान। ControlBus , PerformanceCounters के लिए समर्थन

• आरएक्स एकीकरण। Rx एक्सटेंशन का समर्थन

• एन्क्रिप्शन। भेजे गए संदेशों का एन्क्रिप्शन। Rijndael ब्लॉक सिफर का उपयोग एन्क्रिप्शन के लिए किया जाता है।

• इकाई परीक्षण। परीक्षण उद्देश्यों के लिए, मासट्रांसिट अंतर्निहित परिवहन (लूपबैक) का उपयोग कर सकता है, जिसे बाहरी एमक्यू सर्वर की आवश्यकता नहीं होती है

लेख से उदाहरणों का कोड यहां लिया जा सकता है ।