RabbitMQ ट्यूटोरियल 3 - प्रकाशित / सदस्यता लें

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

प्रकाशित / सदस्यता लें

पिछले लेख में, एक कार्य संदेश कतार के निर्माण पर चर्चा की गई थी। यह धारणा बनाई गई कि प्रत्येक संदेश एक हैंडलर (कार्यकर्ता) को भेजा जाएगा। यह लेख कार्य को जटिल करेगा - हम कई ग्राहकों को एक संदेश भेजेंगे। इस पैटर्न को " प्रकाशित / सदस्यता " के रूप में जाना जाता है।
इस पैटर्न को समझने के लिए, हम एक सरल लॉगिंग सिस्टम बनाएंगे। इसमें दो कार्यक्रम शामिल होंगे - पहला लॉग बनाएगा, दूसरा पढ़ेगा और उन्हें प्रिंट करेगा।

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

अनिवार्य रूप से, प्रत्येक ग्राहक को प्रत्येक संदेश प्रसारित किया जाएगा।

विनिमय अंक

पिछले लेखों में, हमने संदेश भेजने और प्राप्त करने के लिए कतार के साथ काम किया। अब उन्नत खरगोश मैसेजिंग मॉडल पर विचार करें।

पिछले लेख की शर्तों को याद करें:

• निर्माता (आपूर्तिकर्ता) - एक प्रोग्राम जो संदेश भेजता है
• कतार - एक बफर एक संदेश भंडारण
• उपभोक्ता (सब्सक्राइबर) - एक ऐसा कार्यक्रम जो संदेशों को स्वीकार करता है।

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

- आपूर्तिकर्ता से संदेश प्राप्त करता है;
- इन संदेशों को कतार में भेजता है।

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



कई प्रकार हैं: प्रत्यक्ष, विषय, हेडर और फैनआउट। हम अंतिम प्रकार के धूमधाम पर ध्यान केंद्रित करेंगे। इस प्रकार के साथ एक पहुंच बिंदु बनाएं और इसे कॉल करें - लॉग:

$channel->exchange_declare('logs', 'fanout', false, false, false); 

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

पहुँच बिंदुओं की सूची देखें:

सर्वर पर सभी एक्सेस पॉइंट देखने के लिए, आपको rabbitmqctl कमांड चलाना होगा:

 $ sudo rabbitmqctl list_exchanges Listing exchanges ... direct amq.direct direct amq.fanout fanout amq.headers headers amq.match headers amq.rabbitmq.log topic amq.rabbitmq.trace topic amq.topic topic logs fanout ...done. 

हम नाम के साथ पहुंच बिंदुओं की एक सूची देखते हैं। * और एक नाम के बिना एक एक्सेस बिंदु, जो डिफ़ॉल्ट रूप से उपयोग किया जाता है (यह हमारे कार्य के लिए उपयुक्त नहीं है)।

पहुँच बिंदुओं का नाम।

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

 $channel->basic_publish($msg, '', 'hello'); 

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

अब हम अपने नामित पहुँच बिंदु पर एक संदेश भेज सकते हैं:

 $channel->exchange_declare('logs', 'fanout', false, false, false); $channel->basic_publish($msg, 'logs'); 

समय कतार:

इस बार हमने कतारों के नाम का इस्तेमाल किया ("नमस्ते" या "टास्क_क्व्यू")। नाम देने की क्षमता श्रमिकों के लिए एक विशिष्ट कतार निर्दिष्ट करने में मदद करती है, साथ ही साथ उत्पादकों और ग्राहकों के बीच कतार को विभाजित करती है।

लेकिन हमारे लॉगिंग सिस्टम के लिए आवश्यक है कि सभी संदेश कतार में भेजे जाएं, न कि केवल एक भाग। हम यह भी चाहते हैं कि संदेश पुराना हो, पुराना न हो। इसके लिए हमें 2 चीजें चाहिए:

- हर बार जब हम खरगोश से जुड़ते हैं, हम एक नई कतार बनाते हैं, या सर्वर को एक यादृच्छिक नाम बनाते हैं;
- हर बार जब ग्राहक खरगोश से डिस्कनेक्ट होता है, तो हम कतार को हटा देते हैं।

Php-amqplib क्लाइंट में, जब हम एक नाम के बिना कतार का उपयोग करते हैं, तो हम एक स्वचालित रूप से उत्पन्न के साथ एक अस्थायी कतार बनाते हैं:

 list($queue_name, ,) = $channel->queue_declare(""); 

विधि स्वचालित रूप से उत्पन्न कतार नाम वापस कर देगी। यह इस तरह हो सकता है - 'amq.gen-JzTY20BRgKO-HjmUJjwwLg।'
जब दावा किया गया कनेक्शन समाप्त हो जाता है, तो कतार स्वचालित रूप से हटा दी जाती है।

बाइंडर (बाइंडिंग)

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

 $channel->queue_bind($queue_name, 'logs'); 

अब से, हमारी कतार के संदेश पहुंच बिंदु से गुजरते हैं।
आप rabbitmqctl list_bindings कमांड का उपयोग करके बाइंडिंग की सूची देख सकते हैं।

सभी कतारों में भेजना:

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

स्क्रिप्ट कोड पर विचार करें emit_log.php:

 <?php require_once __DIR__ . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPConnection; use PhpAmqpLib\Message\AMQPMessage; $connection = new AMQPConnection('localhost', 5672, 'guest', 'guest'); $channel = $connection->channel(); $channel->exchange_declare('logs', 'fanout', false, false, false); $data = implode(' ', array_slice($argv, 1)); if(empty($data)) $data = "info: Hello World!"; $msg = new AMQPMessage($data); $channel->basic_publish($msg, 'logs'); echo " [x] Sent ", $data, "\n"; $channel->close(); $connection->close(); ?> 

(emit_log.php स्रोत)

जैसा कि आप देख सकते हैं, कनेक्शन स्थापित होने के बाद, हम एक एक्सेस प्वाइंट बनाते हैं। यह कदम आवश्यक है क्योंकि गैर-मौजूद एक्सेस बिंदु का उपयोग निषिद्ध है।

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

Receive_logs.php सब्सक्राइबर कोड:

 <?php require_once __DIR__ . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPConnection; $connection = new AMQPConnection('localhost', 5672, 'guest', 'guest'); $channel = $connection->channel(); $channel->exchange_declare('logs', 'fanout', false, false, false); list($queue_name, ,) = $channel->queue_declare("", false, false, true, false); $channel->queue_bind($queue_name, 'logs'); echo ' [*] Waiting for logs. To exit press CTRL+C', "\n"; $callback = function($msg){ echo ' [x] ', $msg->body, "\n"; }; $channel->basic_consume($queue_name, '', false, true, false, false, $callback); while(count($channel->callbacks)) { $channel->wait(); } $channel->close(); $connection->close(); ?> 

(Rece_logs.php स्रोत)

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

$ php receive_logs.php > logs_from_rabbit.log
यदि आप स्क्रीन पर लॉग दिखाना चाहते हैं, तो दूसरी विंडो खोलें और टाइप करें:

$ php receive_logs.php
और, ज़ाहिर है, संदेश निर्माता का शुभारंभ:

$ php emit_log.php

Rabbitmqctl list_bindings कमांड का उपयोग करके, हम यह सत्यापित कर सकते हैं कि कोड ने कतार को सही तरीके से बनाया है और इसे एक्सेस प्वाइंट के साथ संबद्ध किया है। दो खुले get_logs.php प्रोग्राम के साथ, आपके पास निम्नलिखित होने चाहिए:

$ sudo rabbitmqctl list_bindings
Listing bindings ...
logs exchange amq.gen-JzTY20BRgKO-HjmUJj0wLg queue []
logs exchange amq.gen-vso0PVvyiRIL2WoV3i48Yg queue []
...done.

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

निम्न आलेख वर्णन करेगा कि केवल संदेशों का सबसेट कैसे सुने।