संक्षेप में पायथन से RabbitMQ के साथ काम करने के बारे में

ऐसा हुआ है कि मेगाफॉन में काम करने की प्रक्रिया में, एक ही कार्य को RabbitMQ के साथ काम करते समय सामना करना पड़ता है। स्वाभाविक रूप से सवाल उठता है: "ऐसे कार्यों के कार्यान्वयन को सरल और स्वचालित कैसे करें?"

दिमाग में आने वाला पहला समाधान HTTP इंटरफ़ेस का उपयोग करना है, और निश्चित रूप से, बॉक्स से बाहर, RabbitMQ में एक अच्छा वेब इंटरफ़ेस और HTTP एपीआई है। फिर भी, HTTP API का उपयोग हमेशा सुविधाजनक नहीं होता है, और कभी-कभी असंभव भी होता है (मान लीजिए कि आपके पास पर्याप्त एक्सेस अधिकार नहीं हैं, लेकिन मैं वास्तव में एक संदेश प्रकाशित करना चाहता हूं) ऐसे समय में, AMQP प्रोटोकॉल का उपयोग करके काम करना आवश्यक हो जाता है

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

• पोस्ट प्रकाशित
• प्रूफ़ देना संदेश
• मूल मार्ग तत्वों का निर्माण और संपादन

पायथन को इस तरह के कार्य को लागू करने के लिए सबसे सरल (और मेरी राय में सुंदर) उपकरण के रूप में चुना गया था। (यहां कोई बहस कर सकता है, लेकिन यह क्या बदलेगा?)

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

संक्षेप में एएमक्यूपी प्रोटोकॉल और रैबिटएमक्यू संदेश दलाल के बारे में

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

RabbitMQ एक खुला स्रोत अनुप्रयोग है जो पूरी तरह से AMQP का समर्थन करता है और कई अतिरिक्त सुविधाएँ प्रदान करता है। RabbitMQ के साथ काम करने के लिए, बड़ी संख्या में पुस्तकालयों को विभिन्न प्रकार की प्रोग्रामिंग भाषाओं में लिखा गया है, जिसमें पायथन भी शामिल है।

पायथन कार्यान्वयन

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

1. मॉड्यूल पोस्टिंग के लिए जिम्मेदार है
2. कतार से संदेशों को घटाने के लिए जिम्मेदार मॉड्यूल
3. एक मॉड्यूल जिसे RabbitMQ ब्रोकर के कॉन्फ़िगरेशन में परिवर्तन करने के लिए डिज़ाइन किया गया है
4. एक मॉड्यूल जिसमें पिछले मॉड्यूल के लिए पैरामीटर और विधियां आम हैं

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

चूंकि मेगाफ़ॉन में "खरगोशों" की संरचना में पर्याप्त संख्या में नोड्स होते हैं, उपयोग की सुविधा के लिए, नोड्स से कनेक्ट करने के लिए डेटा सामान्य पैरामीटर और विधियों rmq_common_tools.py के साथ एक मॉड्यूल में स्थानांतरित किया जाता है

पायथन में एएमक्यूपी पर काम करने के लिए, हम पिका पुस्तकालय का उपयोग करेंगे।

import pika 

इस लाइब्रेरी का उपयोग करते हुए, RabbitMQ के साथ काम करना तीन मुख्य चरणों में शामिल होगा:

1. एक कनेक्शन स्थापित करें
2. आवश्यक संचालन करना
3. कनेक्शन बंद करें

पहला और अंतिम चरण सभी मॉड्यूल के लिए समान हैं और rmq_common_tools.py में कार्यान्वित किया जाता है

कनेक्शन स्थापित करने के लिए:

 rmq_parameters = pika.URLParameters(rmq_url_connection_str) rmq_connection = pika.BlockingConnection(rmq_parameters) rmq_channel = rmq_connection.channel() 

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

 'amqp://rabbit_user:rabbit_password@host:port/vhost' 

एक कनेक्शन बंद करने के लिए:

 rmq_connection.close() 

संदेशों के प्रकाशन

एक संदेश प्रकाशित करना शायद सबसे आसान है, लेकिन एक ही समय में खरगोशों के साथ काम करते समय सबसे लोकप्रिय ऑपरेशन।

Rmq_publish.py में संकलित पोस्ट प्रकाशन उपकरण

संदेश पोस्ट करने के लिए, विधि का उपयोग करें

 rmq_channel.basic_publish(exchange = params.exch, routing_key = params.r_key, body = text) 

जहां:
विनिमय - विनिमय बिंदु का नाम जिस पर संदेश प्रकाशित किया जाएगा
routing_key - रूटिंग कुंजी जिसके साथ संदेश प्रकाशित किया जाएगा
शरीर - संदेश शरीर

rmq_publish.py प्रकाशन के लिए दो संदेश इनपुट मोड का समर्थन करता है:

1. संदेश कमांड लाइन (from_console) के माध्यम से एक पैरामीटर के रूप में दर्ज किया गया है
2. संदेश फ़ाइल से पढ़ा जाता है (from_file)

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

संदेश प्राप्त करना

संदेश प्राप्त करने का मुद्दा अब प्रकाशन जैसी मामूली बात नहीं रह गई है। जब संदेश पढ़ने की बात आती है, तो आपको समझने की आवश्यकता है:

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

संदेश पाठक rmq_consume.py फ़ाइल में कार्यान्वित किया गया

दो ऑपरेटिंग मोड प्रदान किए गए हैं:

1. मौजूदा कतार से संदेश पढ़ें
2. इस कतार से संदेश पढ़ने के लिए एक समय कतार और मार्ग बनाना

एक कतार और मार्ग बनाने के सवाल पर नीचे विचार किया जाएगा।

प्रत्यक्ष प्रूफरीडिंग इस प्रकार कार्यान्वित की जाती है:

 channel.basic_consume(on_message, queue=params.queue) try: channel.start_consuming() except KeyboardInterrupt: channel.stop_consuming() except Exception: channel.stop_consuming() rmq_tools.console_log(":\n", traceback.format_exc()) 

जहाँ
on_message - संदेश हैंडलर प्रक्रिया
params.queue - कतार का नाम जिससे घटाव बनाया जाएगा

मैसेज हैंडलर को मैसेज डिलीवरी पर कुछ ऑपरेशन करना होगा और मैसेज डिलीवर होने पर (या कन्फर्म नहीं होने पर कन्फर्म नहीं) करना होगा।

 def on_message(channel, method_frame, header_frame, body): global all_cnt, lim if all_cnt >= lim: rmq_tools.console_log('   .') raise KeyboardInterrupt body_str = body.decode("utf-8")[:4000] rk = method_frame.routing_key rmq_params.file.write(rk + '\n') rmq_params.file.write(body_str + '\n\n') all_cnt = all_cnt + 1 if (lim != 0) and (rmq_params.file == sys.stdout): sys.stdout.write(f'[{rmq_tools.time_now()}] - {all_cnt} of {lim} messages consumed.\r') channel.basic_ack(delivery_tag=method_frame.delivery_tag) 

जहाँ
all_cnt - वैश्विक काउंटर
lim - पढ़ने के लिए संदेशों की संख्या

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

डेटाबेस में रीड मैसेज लिखना भी संभव है। वर्तमान कार्यान्वयन में, ऐसा अवसर प्रस्तुत नहीं किया गया है, लेकिन इसे जोड़ना मुश्किल नहीं है।

 ora_adress = 'host:port/dbSID' ora_creds = 'user/pass' connection_ora = cx_Oracle.connect(ora_creds + '@' + ora_address) ora_cursor = connection_ora.cursor() 

 global cnt, commit_int insert_rec = 'insert into ' + tab_name + '(routing_key, text) values (:rkey, :text)' ora_cursor.execute(insert_rec, text = body_str, rkey = rk) if cnt > commit_int : ora_cursor.execute('commit') cnt = 1 cnt = cnt + 1 

 ora_cursor.execute('commit') connection_ora.close() 

का विन्यास

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

मुख्य कॉन्फ़िगरेशन ऑपरेशन हैं:

1. कतार या विनिमय बिंदु बनाना
2. एक अग्रेषण नियम बनाना (बाध्यकारी)
3. एक कतार या विनिमय बिंदु हटाना
4. एक अग्रेषण नियम (बंधन) हटाना
5. कतार समाशोधन

चूंकि उनमें से प्रत्येक के लिए पिका पुस्तकालय में तैयार प्रक्रिया है, लॉन्च की सुविधा के लिए, वे बस rmq_setup.py फ़ाइल में संकलित हैं। अगला, हम मापदंडों के बारे में कुछ टिप्पणियों के साथ पिका पुस्तकालय से प्रक्रियाओं को सूचीबद्ध करते हैं।

एक कतार बनाना

 rmq_channel.queue_declare(queue=params.queue, durable = params.durable) 

यहाँ सब कुछ सरल है
कतार - बनाने के लिए कतार का नाम
टिकाऊ - एक तार्किक पैरामीटर, ट्रू का मान मतलब होगा कि जब खरगोश रिबूट होता है, तो कतार मौजूद रहेगी। यदि गलत है, तो रिबूट पर कतार हटा दी जाएगी। दूसरा विकल्प आमतौर पर अस्थायी कतारों के लिए उपयोग किया जाता है जो भविष्य में आवश्यक नहीं होने की गारंटी देते हैं।

विनिमय बिंदु (विनिमय) बनाना

 rmq_channel.exchange_declare(exchange=params.exch, exchange_type = params.type, durable = params.durable) 

यहाँ एक नया पैरामीटर exchange_type उत्पन्न होता है - विनिमय बिंदु का प्रकार। किस प्रकार के विनिमय बिंदुओं के बारे में यहां पढ़ा गया है ।
विनिमय - निर्मित विनिमय बिंदु का नाम

एक कतार या विनिमय बिंदु हटाना

 rmq_channel.queue_delete(queue=params.queue) rmq_channel.exchange_delete(exchange=params.exch) 

एक अग्रेषण नियम बनाना (बाध्यकारी)

 rmq_channel.queue_bind(exchange=params.exch, queue=params.queue, routing_key=params.r_key) 

विनिमय - विनिमय बिंदु का नाम जिससे हस्तांतरण किया जाएगा
कतार - कतार का नाम जिसे आगे भेजा जाना है
रूटिंग_की - रूटिंग कुंजी का मुखौटा, जिसका उपयोग अग्रेषण के लिए किया जाएगा।

निम्नलिखित प्रविष्टियां मान्य हैं:

• rk.my_key। * - इस मास्क में, एक तारांकन का अर्थ है एक गैर-खाली वर्ण सेट। दूसरे शब्दों में, इस तरह के एक मुखौटा rk.my_key प्रकार की किसी भी कुंजी को छोड़ देगा । + कुछ और, लेकिन कुंजी rk.my_key को याद नहीं करेंगे
• rk.my_key। # - यह मुखौटा पिछले + कुंजी rk.my_key के रूप में सब कुछ छोड़ देगा

एक अग्रेषण नियम (बंधन) हटाना

 rmq_channel.queue_unbind(exchange=params.exch, queue=params.queue, routing_key=params.r_key) 

सब कुछ एक अग्रेषण नियम बनाने के समान है।

कतार समाशोधन

 rmq_channel.queue_purge(queue=params.queue) 

कतार - कतार का नाम साफ होने के लिए

रोजमर्रा की जिंदगी में आवेदन। सबसे ज्यादा क्या काम आया।

खैर, अब रोजमर्रा की जिंदगी में एएमक्यूपी के उपयोग के बारे में थोड़ी छाप।

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

दूसरी सबसे लोकप्रिय समय कतार से संदेशों को पढ़ने की क्षमता थी। यह सुविधा नए मार्गों और संदेश प्रवाह को कॉन्फ़िगर करने में उपयोगी है, साथ ही दुर्घटनाओं को रोकने में भी।

अन्य संभावनाओं को भी विभिन्न कार्यों में आवेदन मिला।