हम RabbitMQ और TypeScript का उपयोग करके ऑनलाइन स्टोर से ऑर्डर प्रोसेस करते हैं

सभी को नमस्कार! इंटरनेट वाणिज्य की लोकप्रियता लगातार बढ़ रही है, जैसा कि व्यापार से संबंधित सभी प्रकार की गतिविधियों के अनौपचारिककरण का हिस्सा है। इसके साथ ही सूचना प्रसंस्करण की जटिलता बढ़ रही है। ऑनलाइन स्टोर के ग्राहक द्वारा किया गया प्रत्येक ऑर्डर विभिन्न सेवाओं के साथ बड़ी संख्या में एकीकरण उत्पन्न करता है। ऐसी सेवाओं में भुगतान प्रसंस्करण, वितरण, लेखांकन और वफादारी सेवाएं शामिल हो सकती हैं। प्रत्येक आदेश का भुगतान, रिकॉर्ड, इकट्ठा और वितरित किया जाना चाहिए, और आगे के विश्लेषण के लिए भी उपलब्ध होना चाहिए। यह और इसलिए एक साधारण स्थिति नहीं है, इस तथ्य से जटिल है कि एक ऑनलाइन स्टोर का उपयोगकर्ता ऑर्डर देते समय किसी चीज के लिए लंबे समय तक और दर्दनाक रूप से इंतजार नहीं करना चाहता है। ऑनलाइन स्टोर से प्रतिक्रिया त्वरित होनी चाहिए, क्योंकि देरी के प्रत्येक मिलीसेकंड से ग्राहक को खोने की संभावना बढ़ जाती है, और बाद में लाभ होता है। इस आलेख में मैं RabbitMQ संदेश ब्रोकर के बारे में बात करना चाहता हूं और इसे कैसे उपयोग करना है ताकि ऑर्डर प्रोसेसिंग को Node.js और टाइपस्क्रिप्ट का उपयोग करके व्यवस्थित किया जा सके। बिल्ली के लिए आपका स्वागत है।

आवश्यक सिद्धांत

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

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

यह वर्णन करने के लिए कि विभिन्न प्रकार के एक्सचेंजर्स कैसे काम करते हैं, यह समझना आवश्यक है कि रूटिंग कुंजी क्या हैं। रूटिंग कुंजी दोनों कतार से एक्सचेंजर तक और संदेश में ही होती है। रूटिंग कुंजी बस एक स्ट्रिंग है जिसे ब्लॉकों में विभाजित किया गया है। प्रत्येक ब्लॉक को एक बिंदी द्वारा अलग किया जाता है। उदाहरण के लिए, "inform.sendEmail.sendSms"। उसी समय, विशेष वर्ण # और * का उपयोग करके संदेश मार्ग कुंजी के लिए पैटर्न सेट किया जा सकता है। * - कहते हैं कि एक बिंदु के बाद कोई भी ब्लॉक जा सकता है, लेकिन # ब्लॉक के बाद कोई भी नंबर जा सकता है। उदाहरण के लिए, "inform.sendSms। *" या "सूचित करें। #"। अब आप विनिमय बिंदुओं के प्रकारों पर आगे बढ़ सकते हैं।

एक्सचेंजर्स के चार प्रकार हैं:

• फैनआउट। इस एक्सचेंज का रूटिंग लॉजिक सरल है; यह एक आने वाले संदेश को सभी कतारों या एक्सचेंजों से पुनर्निर्देशित करता है जो इससे जुड़े होते हैं।

• प्रत्यक्ष। यह विनिमय संदेश को इस आधार पर पुनर्निर्देशित करता है कि संदेश की रूटिंग कुंजी बाइंडिंग की रूटिंग कुंजी से मेल खाती है या नहीं।

• विषय। इस प्रकार का आदान-प्रदान और साथ ही संदेश रूट रूट पर निर्भर करता है। लेकिन एक पैटर्न एक रूटिंग कुंजी के रूप में कार्य कर सकता है।

• मुखिया । यह एक्सचेंज, दूसरों के विपरीत, रूटिंग के लिए संदेश हेडर का उपयोग करता है। उसी समय, एक्सचेंजर्स की कतारें भी एक साहचर्य सरणी का उपयोग करके बाध्य होती हैं। वह तर्क जिसके द्वारा एक्सचेंजर संदेशों को रूट करेगा विशेष "x-match" कुंजी का उपयोग करके बदला जा सकता है, जो साहचर्य बंधन सरणी में सेट है। कुंजी को सभी या किसी भी दो मानों पर सेट किया जा सकता है। यदि मान सभी है, तो संदेश हेडर को पूरी तरह से साहचर्य बाइंडिंग सरणी से मेल खाना चाहिए; यदि मान किसी भी है, तो मान कम से कम एक कुंजी से मेल खाना चाहिए।

ये RabbitMQ के मुख्य घटक हैं। आप इन घटकों के बारे में एएमक्यूपी प्रोटोकॉल विनिर्देश में पढ़ सकते हैं। इसके बाद, हम एक घटक के रूप में टाइपस्क्रिप्ट का उपयोग करके एक ऑर्डर प्रोसेसिंग सिस्टम को डिज़ाइन और कार्यान्वित करेंगे, साथ ही साथ प्रत्येक घटक की सेटिंग्स को समझेंगे।

डिज़ाइन

उदाहरण को सरल बनाने के लिए, हम मानते हैं कि ऑनलाइन ऑर्डर के सफल प्रसंस्करण के लिए, हमारे पास निम्नलिखित कार्यक्षमता होनी चाहिए:

• आने वाले आदेशों को सहेजें
• ऑर्डर नंबर के साथ क्लाइंट को एसएमएस भेजें, साथ ही ऑर्डर की स्थिति भी
• यदि हमारे ग्राहक ने इस वितरण विधि को चुना है, तो हमारे ऑनलाइन स्टोर से एक नए ऑर्डर के बारे में कूरियर डिलीवरी सेवा को संदेश भेजें

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

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

अंतिम लक्ष्य का अधिक स्पष्ट रूप से प्रतिनिधित्व करने के लिए, मैं एक आरेख बनाऊंगा।

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

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

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

कार्यान्वयन

कार्यान्वयन को सत्यापित करने के लिए, आपको खरगोश की आवश्यकता होगी। मैं इस उद्देश्य के लिए डॉकटर और एक आधिकारिक ब्रोकर छवि का उपयोग करने की सलाह देता हूं। निम्नलिखित कमांड के साथ कंटेनर को स्थापित करें और चलाएं।

docker run -d --name rabbit -p 5672:5672 -e rabbitmq:3.7.15-management-alpine 

यह सुविधाजनक डिबगिंग के लिए पोर्ट 15672 पर उपलब्ध वेब इंटरफेस के साथ एक छवि है।

हम टाइपस्क्रिप्ट और एमक्लिब लाइब्रेरी (Node.js के लिए RabbitMQ क्लाइंट कार्यान्वयन) के साथ अपनी योजनाओं को लागू करेंगे, इसलिए एक शुरुआत के लिए आपको कई इंटरफेस का वर्णन करने की आवश्यकता है। हम ऑर्डर के इंटरफेस और उन संदेशों का वर्णन करते हैं जो हम खरगोश को भेजेंगे।

 //    export interface Product { id: string; name: string; price: number; } //    export interface Order { clientName: string; city: string; email?: string; phone?: string; products: Product[]; totalSum: number; deliveryAddress?: string; } //         export interface OrderWithPhone extends Order { phone: string; } //        export interface OrderWithDeliveryAddress extends Order { deliveryAddress: string; } // Types Guard'        export const isOrderWithPhone = (order: Order): order is OrderWithPhone => Boolean(order.phone); export const isOrderWithDeliveryAddress = (order: Order): order is OrderWithDeliveryAddress => Boolean(order.deliveryAddress); //    . export interface Message<O extends Order> { errors: string[]; retry: number; order: O; //         export interface FailOrder extends Message<Order> { exchange: string; routingKey: string; } 

अब हमें कतारों और एक्सचेंजों के कॉन्फ़िगरेशन इंटरफ़ेस का वर्णन करने की आवश्यकता है, जिसके आधार पर हम खरगोश में प्रसंस्करण संरचना का निर्माण करेंगे।

 import { Types, ExchangeTypes } from '../constants'; import { Options } from 'amqplib'; //   RabbitMQ       export enum Types { QUEUE = 'queue', EXCHANGE = 'exchange', } //      export enum ExchangeTypes { TOPIC = 'topic', } //    export interface Queue { name: string; options: Options.AssertQueue; } //    export interface Exchange { name: string; type: ExchangeTypes; } //    export interface Binding { type: Types; destination: string; source: string; routingKey: string; } //   RabbitMQ export interface PipelineConfig { queues: Queue[]; exchanges: Exchange[]; bindings: Binding[]; } 

सिस्टम के मुख्य घटकों का वर्णन करने के बाद, हम उस कॉन्फ़िगरेशन का वर्णन करते हैं जो ऑब्जेक्ट का उपयोग करके आरेख पर खींचा गया था।

कतार

 export default [ //        routingKey { name: 'generateRoutingKey', options: { durable: true, }, }, //   sms { name: 'sendSms', options: { durable: true, }, }, //      { name: 'delivery', options: { durable: true, }, }, //         sms { name: 'sendSmsHold', options: { durable: true, deadLetterExchange: 'notify', deadLetterRoutingKey: 'sendSms', messageTtl: 60000, }, }, //            { name: 'deliveryHold', options: { durable: true, deadLetterExchange: 'integrates', deadLetterRoutingKey: 'delivery', messageTtl: 60000, }, }, ]; 

कतारों का वर्णन करते समय, कतार के लिए निम्नलिखित विकल्पों का उपयोग किया जाता है।

• टिकाऊ । डिफ़ॉल्ट रूप से, सभी कतार संदेश मेमोरी में संग्रहीत होते हैं। इसलिए, जब ब्रोकर रिबूट होता है, तो संदेश गायब हो जाएंगे। इससे बचने के लिए, आप इस विकल्प का उपयोग कर सकते हैं। इस सेटिंग के साथ, खरगोश संदेशों को डिस्क में फ्लश कर देगा। लेकिन एक कैविएट है। ब्रोकर के पुनरारंभ होने के बाद सहेजे जाने वाले संदेशों के लिए, यह सेटिंग पर्याप्त नहीं है, संदेशों को लगातार विकल्प के साथ कतार में भेजा जाना चाहिए।
• messageTtl संदेश जीवन भर। मिलीसेकंड में दिया गया
• deadLetterExchange । एक्सचेंजर का नाम जहां संदेश समाप्त होने पर कतार से भेजा जाएगा
• deadLetterRoutingKey । RoutingKey जिसके साथ पिछले विकल्प से एक्सचेंज को संदेश भेजा जाएगा

एक्सचेंजों

 import { ExchangeTypes } from '../constants'; export default [ { name: 'postprocessing', type: ExchangeTypes.TOPIC, }, { name: 'notify', type: ExchangeTypes.TOPIC, }, { name: 'integrates', type: ExchangeTypes.TOPIC, }, ]; 

बाइंडिंग

 import { Types } from '../constants'; export default [ { type: Types.EXCHANGE, destination: 'notify', source: 'postprocessing', routingKey: '#.notify.#', }, { type: Types.EXCHANGE, destination: 'integrates', source: 'postprocessing', routingKey: '#.integrates.#', }, { type: Types.QUEUE, destination: 'generateRoutingKey', source: 'postprocessing', routingKey: 'generateRoutingKey', }, { type: Types.QUEUE, destination: 'sendSms', source: 'notify', routingKey: '#.sendSms.#', }, { type: Types.QUEUE, destination: 'delivery', source: 'integrates', routingKey: '#.delivery.#', }, { type: Types.QUEUE, destination: 'sendSmsHold', source: 'notify', routingKey: 'sendSmsHold', }, { type: Types.QUEUE, destination: 'deliveryHold', source: 'integrates', routingKey: 'deliveryHold', }, ]; 

पूर्ण विन्यास

 import { PipelineConfig } from '../interfaces'; import exchanges from './exchanges'; import queues from './queues'; import bindings from './bindigs'; export const pipelineConfig: PipelineConfig = { exchanges, queues, bindings, }; 

खरगोश से कनेक्ट करने के लिए, एक वर्ग लिखें।

 import { connect, Connection, Channel } from 'amqplib'; export class RabbitConnect { private _uri: string; private _connection: Connection; private _chanel: Channel; constructor() { //    rabbit     this._uri = process.env.RABBIT_URI || 'amqp://localhost'; } protected async connect() { this._connection = await connect(this._uri); this._chanel = await this._connection.createChannel(); } protected async disconnect() { await this._chanel.close(); return this._connection.close(); } protected get chanel() { return this._chanel; } } 

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

 import { RabbitConnect } from './RabbitConnect'; import { PipelineConfig } from './interfaces'; import { Types } from './constants'; export class Pipeline extends RabbitConnect { private _pipeline: PipelineConfig; constructor(pipelineConfig: PipelineConfig) { super(); this._pipeline = pipelineConfig; } public async create() { try { await this.connect(); //   const createQueues = this._pipeline.queues.map(queue => this.chanel.assertQueue(queue.name, queue.options), ) as PromiseLike<any>[]; //   const createExchanges = this._pipeline.exchanges.map(exchange => this.chanel.assertExchange(exchange.name, exchange.type), ) as PromiseLike<any>[]; await Promise.all([...createQueues, ...createExchanges]); //       const createBindings = this._pipeline.bindings.map(binding => { if (binding.type === Types.QUEUE) { return this.chanel.bindQueue(binding.destination, binding.source, binding.routingKey); } return this.chanel.bindExchange(binding.destination, binding.source, binding.routingKey); }); await Promise.all(createBindings); return this.disconnect(); } catch (error) { console.error(error); throw new Error(error); } } } 

अब हम सभी श्रमिकों के लिए एक सामान्य कार्यशीलता के साथ एक अमूर्त कार्यकर्ता वर्ग लिखेंगे, जिसमें से विरासत में यह संभव होगा।

 import { RabbitConnect } from './RabbitConnect'; import { Message, Order, FailOrder } from './interfaces'; import { ConsumeMessage } from 'amqplib'; export interface WorkerParams { maxRetry?: number; //     active: string; //    exchange: string; //       holdKey?: string; //      } export abstract class Worker<M extends Order> extends RabbitConnect { private _maxRetry: number; private _active: string; private _holdKey: string | undefined; protected exchange: string; private _currentMessage: Message<M>; private _currentConsumeMessage: ConsumeMessage; constructor({ active, holdKey, exchange, maxRetry }: WorkerParams) { super(); this._maxRetry = maxRetry || 0; this._active = active; this._holdKey = holdKey; this.exchange = exchange; } public async subscribe() { await this.connect(); this.chanel.consume(this._active, this.checkMessage.bind(this)); } //          private async checkMessage(message: ConsumeMessage) { this._currentConsumeMessage = message; const orderMessage: Message<M> = JSON.parse(message.content.toString()); if (orderMessage.retry >= this._maxRetry) { await this.sendToErrorStorage('  '); } this._currentMessage = orderMessage; //           await this.handler(orderMessage.order || orderMessage); } //       protected async sendToErrorStorage(error: string) { const message: FailOrder = { order: this._currentMessage.order, errors: [...this._currentMessage.errors, error], retry: this._currentMessage.retry + 1, exchange: this.exchange, routingKey: this._active }; console.log('   ', message); this.ack(); } //       protected async hold(error: string) { if (!this._holdKey) { return; } const orderMessage = { order: this._currentMessage.order, errors: [...this._currentMessage.errors, error], retry: this._currentMessage.retry + 1, }; const orderData = Buffer.from(JSON.stringify(orderMessage)); return this.chanel.publish(this.exchange, this._holdKey, orderData); } //      protected async ack() { return this.chanel.ack(this._currentConsumeMessage); } protected abstract handler(message: M): void; } 

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

अब हम आरेख से कुछ सरल श्रमिकों को लिख सकते हैं।

एक रूटिंग कुंजी उत्पन्न करने वाला कार्यकर्ता।

 import { Worker } from '../Worker'; import { isOrderWithPhone, isOrderWithDeliveryAddress, Order, Message, } from '../interfaces'; import { Keys } from '../constants'; export class GenerateRoutingKey extends Worker<Order> { constructor() { super({ active: 'generateRoutingKey', exchange: 'postprocessing', }); } protected async handler(order: Order) { try { const routingKey: string[] = []; if (isOrderWithPhone(order)) { routingKey.push(Keys.SEND_SMS); } if (isOrderWithDeliveryAddress(order)) { routingKey.push(Keys.SEND_TO_DELIVERY); } const message: Message<Order> = { retry: 0, errors: [], order, }; await this.chanel.publish( this.exchange, routingKey.join('.'), Buffer.from(JSON.stringify(message)), ); await this.ack(); } catch (error) { console.error(error); await this.sendToErrorStorage(error); } } } 

एसएमएस भेजने वाले मजदूर।

 import { Worker } from '../Worker'; import { OrderWithPhone } from '../interfaces'; export class SendSms extends Worker<OrderWithPhone> { constructor() { super({ active: 'sendSms', exchange: 'notify', holdKey: 'sendSmsHold', maxRetry: process.env.MAX_RETRY ? parseInt(process.env.MAX_RETRY) : 5, }); } protected async handler(message: OrderWithPhone) { try { console.log(' sms  : ', message.phone); this.ack(); } catch (error) { console.error(error); await this.hold(error); } } } 

वितरण सेवा के साथ कार्यकर्ता एकीकरण।

 import { Worker } from '../Worker'; import { OrderWithDeliveryAddress } from '../interfaces'; export class Delivery extends Worker<OrderWithDeliveryAddress> { constructor() { super({ active: 'delivery', exchange: 'interates', holdKey: 'deliveryHold', maxRetry: process.env.MAX_RETRY ? parseInt(process.env.MAX_RETRY) : 5, }); } protected async handler(message: OrderWithDeliveryAddress) { try { console.log('      : ', message.deliveryAddress); this.ack(); } catch (error) { console.error(error); await this.hold(error); } } } 

आवेदन के लिए प्रवेश बिंदु।

 import { Pipeline } from './Pipeline'; import { pipelineConfig } from './pipeline'; import { GenerateRoutingKey } from './workers/GenerateRoutingKey'; import { SendSms } from './workers/SendSms'; import { Delivery } from './workers/Delivery'; (async () => { try { const pipeline = new Pipeline(pipelineConfig); const generateRoutingKey = new GenerateRoutingKey(); const sendSms = new SendSms(); const delivery = new Delivery(); await pipeline.create(); await Promise.all([generateRoutingKey.subscribe(), sendSms.subscribe(), delivery.subscribe()]); } catch (error) { console.error(error); process.exit(1); } })(); 

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

निष्कर्ष

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

मुझे उम्मीद है कि आपको लेख पसंद आया होगा। मुझे किसी भी टिप्पणी और आलोचना पर खुशी होगी। सभी प्रस्तुत कोड जीथब पर पाए जा सकते हैं