RabbitMQ ट्यूटोरियल 2 - टास्क कतार

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

टास्क की कतारें

पहले पाठ में, हमने दो कार्यक्रम लिखे: एक ने संदेश भेजे, दूसरे ने उन्हें प्राप्त किया। इस पाठ में, हम एक कतार बनाएंगे जिसका उपयोग कई ग्राहकों के बीच संसाधन-गहन कार्यों को वितरित करने के लिए किया जाएगा।

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

ऑपरेशन का यह सिद्धांत वेब अनुप्रयोगों में उपयोग के लिए विशेष रूप से उपयोगी है जहां HTTP अनुरोध के दौरान संसाधन-गहन कार्य को संभालना असंभव है।

ट्रेनिंग

पिछले पाठ में, हमने "हैलो वर्ल्ड!" पाठ के साथ एक संदेश भेजा था। और अब हम संसाधन-गहन कार्यों के लिए संदेश भेजेंगे। हम वास्तविक कार्यों का प्रदर्शन नहीं करेंगे, जैसे कि एक छवि को आकार देना या एक पीडीएफ फाइल को रेंडर करना, चलो time.sleep () फ़ंक्शन का उपयोग करके सिर्फ एक स्टब बनाते हैं। कार्य की जटिलता संदेश लाइन में अंकों की संख्या से निर्धारित की जाएगी। प्रत्येक बिंदु एक सेकंड के लिए "चलेगा"। उदाहरण के लिए, "हैलो ..." संदेश के साथ एक कार्य 3 सेकंड के लिए चलेगा।

हम पिछले उदाहरण से सेंडहोम प्रोग्राम कोड को थोड़ा संशोधित करेंगे ताकि कमांड लाइन से मनमाने संदेश भेजना संभव हो। यह कार्यक्रम हमारी बारी में संदेश भेजेगा, नए कार्यों की योजना बनाएगा। इसे new_task.py कहें :

import sys message = ' '.join(sys.argv[1:]) or "Hello World!" channel.basic_publish(exchange='', routing_key='hello', body=message) print " [x] Sent %r" % (message,) 

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

 import time def callback(ch, method, properties, body): print " [x] Received %r" % (body,) time.sleep( body.count('.') ) print " [x] Done" 

चक्रीय वितरण

कार्य कतार का उपयोग करने के लाभों में से एक कई कार्यक्रमों के साथ समानांतर में काम करने की क्षमता है। यदि हमारे पास आने वाले सभी कार्यों को पूरा करने का समय नहीं है, तो हम बस हैंडलर्स की संख्या जोड़ सकते हैं।

सबसे पहले, एक साथ दो कार्यकर्ता-उन्मुख कार्यक्रम चलाते हैं। दोनों को कतार से संदेश प्राप्त होंगे, लेकिन वास्तव में कैसे? अब हम देखेंगे।

आपको तीन टर्मिनल विंडो खोलने की आवश्यकता है। उनमें से दो में, कार्यकर्ता काम शुरू किया जाएगा। ये दो ग्राहक होंगे - C1 और C2।

 shell1$ python worker.py [*] Waiting for messages. To exit press CTRL+C 

 shell2$ python worker.py [*] Waiting for messages. To exit press CTRL+C 

तीसरी विंडो में हम नए कार्यों को प्रकाशित करेंगे। सब्सक्राइबर शुरू होने के बाद, आप कोई भी संदेश भेज सकते हैं:

 shell3$ python new_task.py First message. shell3$ python new_task.py Second message.. shell3$ python new_task.py Third message... shell3$ python new_task.py Fourth message.... shell3$ python new_task.py Fifth message..... 

आइए देखें कि ग्राहकों को क्या दिया गया:

 shell1$ python worker.py [*] Waiting for messages. To exit press CTRL+C [x] Received 'First message.' [x] Received 'Third message...' [x] Received 'Fifth message.....' 

 shell2$ python worker.py [*] Waiting for messages. To exit press CTRL+C [x] Received 'Second message..' [x] Received 'Fourth message....' 

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

संदेश की पुष्टि

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

लेकिन हम किसी भी कार्य को खोना नहीं चाहते हैं। हमें जरूरत है कि एक हैंडलर के आपातकालीन निकास की स्थिति में, संदेश दूसरे को प्रेषित किया जाए।

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

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

संदेशों को संसाधित करने का कोई समय-निर्धारण नहीं है। RabbitMQ उन्हें दूसरे सब्सक्राइबर को तभी ट्रांसफर करेगा, जब पहले का कनेक्शन बंद हो, इसलिए मैसेज के प्रोसेसिंग टाइम पर कोई पाबंदी नहीं है।

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

 def callback(ch, method, properties, body): print " [x] Received %r" % (body,) time.sleep( body.count('.') ) print " [x] Done" ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_consume(callback, queue='hello') 

अब, भले ही आप संदेश को संसाधित करते समय Ctrl + C दबाकर हैंडलर बंद कर दें, लेकिन कुछ भी नहीं खोएगा। हैंडलर बंद होने के बाद, RabbitMQ अपुष्ट संदेशों को फिर से सबमिट करेगा।

संदेशों की पुष्टि करना न भूलें

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

 $ sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged Listing queues ... hello 0 0 ...done. 

[या अधिक सुविधाजनक निगरानी स्क्रिप्ट का उपयोग करें, जिसे मैंने पहले भाग में उद्धृत किया है]

संदेश स्थिरता

हमें पता चला कि यदि ग्राहक अप्रत्याशित रूप से काम करना बंद कर देते हैं तो कार्यों को कैसे नहीं खोना चाहिए। यदि RabbitMQ सर्वर काम करना बंद कर देता है तो कार्य खो जाएगा।

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

सबसे पहले, सुनिश्चित करें कि कतार खो नहीं है। ऐसा करने के लिए, आपको इसे टिकाऊ ( टिकाऊ ) घोषित करना होगा:

 channel.queue_declare(queue='hello', durable=True) 

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

 channel.queue_declare(queue='task_queue', durable=True) 

यह कोड प्रदाता प्रोग्राम और सब्सक्राइबर प्रोग्राम दोनों के लिए तय किया जाना चाहिए।
तो हम यह सुनिश्चित कर सकते हैं कि RabbitMQ सर्वर के पुनरारंभ होने पर task_queue कतार खो नहीं जाएगी। अब आपको लचीला के रूप में संदेशों को चिह्नित करने की आवश्यकता है। ऐसा करने के लिए, 2 के मान के साथ delivery_mode प्रॉपर्टी पास करें:

 channel.basic_publish(exchange='', routing_key="task_queue", body=message, properties=pika.BasicProperties( delivery_mode = 2, # make message persistent )) 

संदेश स्थिरता के बारे में एक नोट

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

संदेशों का समान वितरण

आपने देखा होगा कि संदेश वितरण अभी भी उतना काम नहीं करता है जितना हमें चाहिए। उदाहरण के लिए, जब दो ग्राहक काम करते हैं, यदि सभी विषम संदेशों में जटिल कार्य होते हैं [पूरा होने में बहुत समय की आवश्यकता होती है] , और यहां तक ​​कि साधारण कार्यों में भी शामिल होते हैं, तो पहला हैंडलर लगातार व्यस्त रहेगा, और दूसरा अधिकांश समय मुफ्त होगा। लेकिन RabbitMQ इसके बारे में कुछ भी नहीं जानता है और फिर भी ग्राहकों को बदले में संदेश भेजेगा।
ऐसा इसलिए है क्योंकि RabbitMQ संदेश को उस क्षण वितरित करता है जो वे कतार में आते हैं और ग्राहकों से अपुष्ट संदेशों की संख्या को ध्यान में नहीं रखते हैं। RabbitMQ बस nth ग्राहक को प्रत्येक nth संदेश भेजता है।



इस व्यवहार को बदलने के लिए, हम prefetch_count = 1 विकल्प के साथ basic_qos विधि का उपयोग कर सकते हैं। इससे RabbitMQ सब्सक्राइबर को एक बार में एक से अधिक मैसेज नहीं दे पाएगा। दूसरे शब्दों में, सब्सक्राइबर को एक नया संदेश तब तक नहीं मिलेगा जब तक कि वह पहले वाले को प्रोसेस और कन्फर्म नहीं करता है। RabbitMQ पहले फ्री सब्सक्राइबर को मैसेज भेजेगा।

 channel.basic_qos(prefetch_count=1) 

कतार आकार पर ध्यान दें

यदि सभी ग्राहक व्यस्त हैं, तो कतार का आकार बढ़ सकता है। आपको इस पर ध्यान देना चाहिए और संभवतः, ग्राहकों की संख्या में वृद्धि करनी चाहिए।

खैर, अब सब साथ हैं

New_task.py का पूरा कोड है :

 #!/usr/bin/env python import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.queue_declare(queue='task_queue', durable=True) message = ' '.join(sys.argv[1:]) or "Hello World!" channel.basic_publish(exchange='', routing_key='task_queue', body=message, properties=pika.BasicProperties( delivery_mode = 2, # make message persistent )) print " [x] Sent %r" % (message,) connection.close() 

कार्यकर्ता के लिए पूरा कोड:

 #!/usr/bin/env python import pika import time connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.queue_declare(queue='task_queue', durable=True) print ' [*] Waiting for messages. To exit press CTRL+C' def callback(ch, method, properties, body): print " [x] Received %r" % (body,) time.sleep( body.count('.') ) print " [x] Done" ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_qos(prefetch_count=1) channel.basic_consume(callback, queue='task_queue') channel.start_consuming() 

संदेश पावती और prefetch_count का उपयोग करके, आप एक कार्य कतार बना सकते हैं। रेज़लूशन को कॉन्फ़िगर करना कार्य को RabbitMQ सर्वर को पुनरारंभ करने के बाद भी जारी रखने की अनुमति देगा।

तीसरे पाठ में, हम इस बात पर ध्यान देंगे कि आप कितने ग्राहकों को एक संदेश भेज सकते हैं।