ZeroMQ। अध्याय 2: सॉकेट्स का परिचय

सभी को नमस्कार!
मैं "ZeroMQ.Use ZeroMQ पुस्तक का मुफ्त अनुवाद जारी रखता हूं और सीखता हूं कि विभिन्न संदेश पैटर्न कैसे लागू करें"। मैं पहले से माफी मांगता हूं कि मैंने इतनी देर तक अगली कड़ी प्रकाशित नहीं की, लेकिन जैसा कि वे कहते हैं: "आलस्य हमारे आगे पैदा हुआ था ..."। अच्छा, चलो गीत को अलग रखें।

सामग्री

• अध्याय 1: आरंभ करना
• अध्याय 2: सॉकेट्स का परिचय
• अध्याय 3: सॉकेट टोपोलॉजी का उपयोग करना
• अध्याय 4: अतिरिक्त पैटर्न

हमने पिछले अध्याय में ZeroMQ की बुनियादी संरचनाओं की जांच करने के बाद, इसमें हम सॉकेट्स पर विचार करेंगे:

• क्लाइंट-सर्वर पैटर्न (प्रकाशित-सदस्यता)
• पाइपलाइन पैटर्न

प्रकाशन-सदस्यता पैटर्न

सबसे पहले, आइए एक क्लासिक पैटर्न, क्लाइंट-सर्वर पैटर्न (पब्लिश-सब्सक्रिप्शन) को पेश करें, जो वितरण की प्रकृति का एकतरफा तरीका है जब सर्वर क्लाइंट की एक विशिष्ट सूची को संदेश भेजता है। यह एक से कई मॉडल है। इस पैटर्न का मुख्य विचार यह है कि सर्वर एक संदेश भेजता है और जुड़े हुए ग्राहक इस संदेश को प्राप्त करते हैं, जबकि डिस्कनेक्ट किए गए ग्राहक बस इसे छोड़ देते हैं। सर्वर शिथिल रूप से ग्राहकों से जुड़ा हुआ है, यह बिल्कुल भी परेशान नहीं करता है कि कोई भी ग्राहक मौजूद है या नहीं। यह टेलीविजन चैनलों या रेडियो स्टेशनों के काम करने के तरीके के समान है। टीवी चैनल हमेशा टेलीविजन कार्यक्रमों का प्रसारण करते हैं और केवल दर्शक ही इस प्रसारण को स्वीकार करने का निर्णय लेते हैं या नहीं। यदि आप सही समय को याद करते हैं, तो आप अपना पसंदीदा टीवी शो नहीं देख पाएंगे (यदि आपके पास TiVo या ऐसा कुछ नहीं है, लेकिन मान लें कि हमारी स्क्रिप्ट ऐसी दुनिया में होती है जहां रिकॉर्डिंग का आविष्कार नहीं किया गया था)। प्रकाशित-सदस्यता टेम्पलेट का लाभ यह है कि यह एक गतिशील नेटवर्क टोपोलॉजी प्रदान करता है।
क्लाइंट-सर्वर मॉडल को निम्नलिखित मुख्य पहलुओं से देखा जा सकता है:

• प्रकाशित: पोस्ट निर्माता द्वारा पोस्ट किया गया है
• सूचित करें: ग्राहक को एक संदेश के प्रकाशन के बारे में सूचित किया जाता है
• सदस्यता: ग्राहक को एक नया हस्ताक्षर जारी किया जाता है
• सदस्यता समाप्त: ग्राहक मौजूदा हस्ताक्षर हटाता है

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



निम्नलिखित निर्माता (सर्वर) का कोड है:

/* * Stock Market Server * Binds PUB socket to tcp://*:4040 * Publishes random stock values of random companies */ #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* publisher = zmq_socket(context, ZMQ_PUB); printf("Starting server...\n"); int conn = zmq_bind(publisher, "tcp://*:4040"); const char* companies[2] = {"Company1", "Company2"}; int count = 0; for(;;) { int price = count % 2; int which_company = count % 2; int index = strlen(companies[0]); char update[12]; snprintf(update, sizeof update, "%s", companies[which_company]); zmq_msg_t message; zmq_msg_init_size(&message, index); memcpy(zmq_msg_data(&message), update, index); zmq_msg_send(&message, publisher, 0); zmq_msg_close(&message); count++; } zmq_close(publisher); zmq_ctx_destroy(context); return 0; } 

नीचे ग्राहक कोड है:

 /* * Stock Market Client * Connects SUB socket to tcp://localhost:4040 * Collects stock exchange values */ #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* subscriber = zmq_socket(context, ZMQ_SUB); printf("Collecting stock information from the server.\n"); int conn = zmq_connect(subscriber, "tcp://localhost:4040"); conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, 0, 0); int i; for(i = 0; i < 10; i++) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, subscriber, 0); int length = zmq_msg_size(&reply); char* value = malloc(length); memcpy(value, zmq_msg_data(&reply), length); zmq_msg_close(&reply); printf("%s\n", value); free(value); } zmq_close(subscriber); zmq_ctx_destroy(context); return 0; } 

जब भी आप SUB सॉकेट का उपयोग करते हैं तो zmq_setsockopt () और subscribeis का उपयोग करके एक हस्ताक्षर सेट करना आवश्यक है, अन्यथा आपको कोई संदेश प्राप्त नहीं होगा। यह एक बहुत ही सामान्य गलती है।
ग्राहक किसी भी संदेश को कई हस्ताक्षर सेट कर सकता है जो उसे प्राप्त होता है यदि अद्यतन किसी भी सदस्यता से मेल खाता है। वह कुछ सदस्यता को मना भी कर सकता है। सदस्यता की एक निश्चित लंबाई होती है।
क्लाइंट को zmq_msg_recv () का उपयोग करके संदेश प्राप्त होता है। zmq_msg_recv () संदेश प्राप्त करता है और बचाता है। पिछला संदेश, यदि कोई हो, अनलोड किया गया है।

 int zmq_msg_recv (zmq_msg_t *msg, void *socket, int flags); 

विकल्प ध्वज केवल एक मान ले सकता है - ZMQ_DONTWAIT । यदि झंडा ZMQ_DONTWAIT , तो ऑपरेशन नॉन-ब्लॉकिंग मोड में किया जाता है। यदि संदेश प्राप्त होता है, तो zmq_msg_recv () संदेश का आकार बाइट्स में लौटाता है, अन्यथा -1 और त्रुटि संदेश ध्वज वापस आ जाता है।
क्लाइंट-सर्वर मॉडल अतुल्यकालिक है और SUB सॉकेट को संदेश भेजने से त्रुटि होती है। आप संदेश भेजने के लिए zmq_msg_send () कॉल कर सकते हैं, लेकिन आपको PUB सॉकेट पर zmq_msg_recv () कभी नहीं कॉल करना चाहिए।
निम्नलिखित ग्राहक पक्ष पर आउटपुट का एक उदाहरण है:

 Company2 570 Company2 878 Company2 981 Company2 783 Company1 855 Company1 524 Company2 639 Company1 984 Company1 158 Company2 145 

सर्वर हमेशा संदेश भेजेगा, भले ही ग्राहक न हों। आप इसका परीक्षण कर सकते हैं और परिणाम देख सकते हैं। यदि आप ऐसा करते हैं, तो आपको निम्नलिखित कुछ दिखाई देगा:

 Sending... Company2 36 Sending... Company2 215 Sending... Company2 712 Sending... Company2 924 Sending... Company2 721 Sending... Company1 668 Sending... Company2 83 Sending... Company2 209 Sending... Company1 450 Sending... Company1 940 Sending... Company1 57 Sending... Company2 3 Sending... Company1 100 Sending... Company2 947 

मान लें कि हम Company1 या किसी अन्य कंपनी के लिए परिणाम प्राप्त करना चाहते हैं जिसका नाम आप एक तर्क के रूप में निर्दिष्ट करते हैं। इस मामले में, हमें अपने ग्राहक कार्यक्रम को इस प्रकार बदलना होगा:

 // // Stock Market Client // Connects SUB socket to tcp://localhost:4040 // Collects stock exchange values // #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* subscriber = zmq_socket(context, ZMQ_SUB); const char* filter; if(argc > 1) { filter = argv[1]; } else { filter = "Company1"; } printf("Collecting stock information from the server.\n"); int conn = zmq_connect(subscriber, "tcp://localhost:4040"); conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter)); int i = 0; for(i = 0; i < 10; i++) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, subscriber, 0); int length = zmq_msg_size(&reply); char* value = malloc(length + 1); memcpy(value, zmq_msg_data(&reply), length); zmq_msg_close(&reply); printf("%s\n", value); free(value); } zmq_close(subscriber); zmq_ctx_destroy(context); return 0; } 

आउटपुट में निम्न जैसा कुछ होगा:

 Company1 575 Company1 504 Company1 513 Company1 584 Company1 444 Company1 1010 Company1 524 Company1 963 Company1 929 Company1 718 

संदेश फ़िल्टरिंग

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

 // // Stock Market Server // Binds PUB socket to tcp://*:4040 // Publishes random stock values of random companies // #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* publisher = zmq_socket(context, ZMQ_PUB); int conn = zmq_bind(publisher, "tcp://*:4040"); const char* companies[3] = {"Company1", "Company10", "Company101"}; int count = 0; for(;;) { int price = count % 17; int which_company = count % 3; int index = strlen(companies[which_company]); char update[64]; snprintf(update, sizeof update, "%s", companies[which_company]); zmq_msg_t message; zmq_msg_init_size(&message, index); memcpy(zmq_msg_data(&message), update, index); zmq_msg_send(&message, publisher, 0); zmq_msg_close(&message); count++; } zmq_close(publisher); zmq_ctx_destroy(context); return 0; } 

अब अपने ग्राहक कोड को निम्न में बदलें:

 // // Stock Market Client // Connects SUB socket to tcp://localhost:4040 // Collects stock exchange values // #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* subscriber = zmq_socket(context, ZMQ_SUB); const char* filter; if(argc > 1) { filter = argv[1]; } else { filter = "Company1"; } printf("Collecting stock information from the server.\n"); int conn = zmq_connect(subscriber, "tcp://localhost:4040"); conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter)); int i = 0; for(i = 0; i < 10; i++) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, subscriber, 0); int length = zmq_msg_size(&reply); char* value = malloc(length + 1); memcpy(value, zmq_msg_data(&reply), length); zmq_msg_close(&reply); printf("%s\n", value); free(value); } zmq_close(subscriber); zmq_ctx_destroy(context); return 0; } 

इस मामले में, आउटपुट निम्नलिखित के समान होगा:

 Collecting stock information from the server. Company101 950 Company10 707 Company101 55 Company101 343 Company10 111 Company1 651 Company10 287 Company101 8 Company1 889 Company101 536 

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

 // // Stock Market Server // Binds PUB socket to tcp://*:4040 // Publishes random stock values of random companies // #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* publisher = zmq_socket(context, ZMQ_PUB); int conn = zmq_bind(publisher, "tcp://*:4040"); conn = zmq_bind(publisher, "ipc://stock.ipc"); const char* companies[3] = {"Company1", "Company10", "Company101"}; for(;;) { int price = count % 17; int which_company = count % 3; int index = strlen(companies[which_company]); char update[64]; sprintf(update, "%s| %d", companies[which_company], price); zmq_msg_t message; zmq_msg_init_size(&message, index); memcpy(zmq_msg_data(&message), update, index); zmq_msg_send(&message, publisher, 0); zmq_msg_close(&message); count++; } zmq_close(publisher); zmq_ctx_destroy(context); return 0; } 

परिणामों को फ़िल्टर करने के लिए अद्यतन क्लाइंट कोड पर एक नज़र डालें:

 // // Stock Market Client // Connects SUB socket to tcp://localhost:4040 // Collects stock exchange values // #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* subscriber = zmq_socket(context, ZMQ_SUB); const char* filter; filter = "Company1|"; printf("Collecting stock information from the server.\n"); int conn = zmq_connect(subscriber, "tcp://localhost:4040"); conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter)); int i = 0; for(i = 0; i < 10; i++) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, subscriber, 0); int length = zmq_msg_size(&reply); char* value = malloc(length + 1); memcpy(value, zmq_msg_data(&reply), length); zmq_msg_close(&reply); printf("%s\n", value); free(value); } zmq_close(subscriber); zmq_ctx_destroy(context); return 0; } 

क्लाइंट और सर्वर कोड में किए गए परिवर्तनों के बाद, हम बिल्कुल वही परिणाम देख सकते हैं जो अपेक्षित थे।

सॉकेट विकल्प

चूंकि हम क्लाइंट-सर्वर मॉडल का उपयोग करते हैं, इसलिए हम ZMQ_SUBSCRIBE नामक पैरामीटर का उपयोग ZMQ_SUBSCRIBE ।

 int conn = zmq_connect(subscriber, "tcp://localhost:4040"); conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, option_value, strlen(option_value)); 

सॉकेट विकल्प zmq_setsockopt () फ़ंक्शन में सेट किए गए हैं। यह चार पैरामीटर लेता है:

• सॉकेट
• विकल्प का नाम
• विकल्प मान
• विकल्प का आकार

इसे निम्नलिखित पंक्ति से देखा जा सकता है:

 int zmq_setsockopt (void *socket, int option_name, const void *option_ value, size_t option_len); 

सदस्यता लें

ZMQ_SUBSCRIBE ZMQ_SUB सॉकेट पर एक नया संदेश बनाता है। यदि option_value तर्क खाली नहीं है, तो हम उन सभी संदेशों को सब्सक्राइब करते हैं, जो option_value शुरू option_value । आप एक ZMQ_SUB सॉकेट के लिए कई फ़िल्टर कॉन्फ़िगर कर सकते हैं।

करें

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

क्लाइंट-सर्वर मॉडल नोट

क्लाइंट-सर्वर मॉडल में, आपको निम्नलिखित बिंदुओं पर ध्यान देना चाहिए:

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

अध्याय 4 में, हम क्लाइंट-सर्वर मॉडल पर लौट आएंगे, अधिक जटिल उदाहरणों पर विचार करेंगे, और दिखाएंगे कि "धीमी" ग्राहकों के साथ कैसे व्यवहार करें।

पाइपलाइन पैटर्न

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

विभाजन करें और रणनीति को जीतें

इस रणनीति से विभाजित और जीतें, जब आप कार्यक्रम करते हैं तो कोई बच नहीं जाता है। याद रखें, जब आप केवल प्रोग्रामिंग का अध्ययन करना शुरू कर रहे थे, और आपके शिक्षक ने इसे लगभग मर्ज की छँटाई में इस्तेमाल किया था, और एक हफ्ते के बाद समूह के आधे लोगों ने कक्षाओं में भाग लेना बंद कर दिया। मुझे यकीन है कि यह सब अच्छी तरह से याद है। और यहाँ फिर से, विभाजित और जीत!
ZeroMQ पर कुछ समानांतर लिखते हैं। एक परिदृश्य पर विचार करें जहां हमारे पास एक जनरेटर है जो यादृच्छिक संख्या उत्पन्न करता है। हमारे पास ऐसे श्रमिक हैं जो न्यूटन पद्धति का उपयोग करके इन संख्याओं का वर्गमूल ज्ञात करते हैं। हमारे पास एक कलेक्टर भी है जो श्रमिकों से परिणाम एकत्र करता है।
नीचे सर्वर कोड है:

 #include <stdio.h> #include <stdlib.h> #include <unistd.h> #include <string.h> #include <sys/time.h> #include <time.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); // This is the socket that we send messages. void* socket = zmq_socket(context, ZMQ_PUSH); zmq_bind(socket, "tcp://*:4040"); // This is the socket that we send batch message. void* connector = zmq_socket(context, ZMQ_PUSH); zmq_connect(connector, "tcp://localhost:5050"); printf("Please press enter when workers are ready..."); getchar(); printf("Sending tasks to workers...\n"); // The first message. It's also the signal start of batch. int length = strlen("-1"); zmq_msg_t message; zmq_msg_init_size(&message, length); memcpy(zmq_msg_data(&message), "-1", length); zmq_msg_send(&message, connector, 0); zmq_msg_close(&message); // Generate some random numbers. srandom((unsigned) time(NULL)); // Send the tasks. int count; int msec = 0; for(count = 0; count < 100; count++) { int load = (int) ((double) (100) * random () / RAND_MAX); msec += load; char string[10]; sprintf(string, "%d", load); } printf("Total: %d msec\n", msec); sleep(1); zmq_close(connector); zmq_close(socket); zmq_ctx_destroy(context); return 0; } 

हम कर्मचारी कोड पर भी एक नज़र डालते हैं, जहाँ हम न्यूटन पद्धति का उपयोग करके किसी संख्या के वर्गमूल की गणना के लिए कुछ गणनाएँ करते हैं:

 #include <stdlib.h> #include <string.h> #include <unistd.h> #include "zmq.h" double square(double x) { return x * x; } double average(double x, double y) { return (x + y) / 2.0; } double good_enough(double guess, double x) { return abs(square(guess) - x) < 0.000001; } double improve(double guess, double x) { return average(guess, x / guess); } double sqrt_inner(double guess, double x) { if(good_enough(guess, x)) return guess; else return sqrt_inner(improve(guess, x), x); } double newton_sqrt(double x) { return sqrt_inner(1.0, x); } int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); // Let's initialize a socket to receive messages. void* receiver = zmq_socket(context, ZMQ_PULL); zmq_connect(receiver, "tcp://localhost:4040"); // Let's initialize a socket to send the messages. void* sender = zmq_socket(context, ZMQ_PUSH); zmq_connect(sender, "tcp://localhost:5050"); for(;;) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, receiver, 0); int length = zmq_msg_size(&reply); char* msg = malloc(length + 1); memcpy(msg, zmq_msg_data(&reply), length); zmq_msg_close(&reply); fflush(stdout); double val = atof(msg); printf("%.1f: %.1f\n", val, newton_sqrt(val)); sleep(1); free(msg); zmq_msg_t message; char* ssend = "T"; int t_length = strlen(ssend); zmq_msg_init_size(&message, t_length); memcpy(zmq_msg_data(&message), ssend, t_length); zmq_msg_send(&message, receiver, 0); zmq_msg_close(&message); } zmq_close(receiver); zmq_close(sender); zmq_ctx_destroy(context); return 0; } 

यहाँ कोड और कलेक्टर है:

 #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* receiver = zmq_socket(context, ZMQ_PULL); zmq_bind(receiver, "tcp://*:5050"); // We receive the first message and discard it since it's the // signal start of batch which is -1. zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, receiver, 0); int length = zmq_msg_size(&reply); char* msg = malloc(length + 1); memcpy(msg, zmq_msg_data(&reply), length); zmq_msg_close(&reply); free(msg); int count; for(count = 0; count < 100; count++) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, receiver, 0); int length = zmq_msg_size(&reply); char* value = malloc(length + 1); memcpy(value, zmq_msg_data(&reply), length); zmq_msg_close(&reply); free(value); if(count / 10 == 0) printf("10 Tasks have been processed."); fflush(stdout); } zmq_close(receiver); zmq_ctx_destroy(context); return 0; } 

निम्नलिखित चित्र ऊपर लिखे कोड का प्रतिनिधित्व करता है:



हमारे पास क्या है:

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

हमने उल्लेख किया कि कार्यकर्ता सर्वर और कलेक्टर दोनों से जुड़े हैं। आइए इन लिंक को अधिक विस्तार से देखें।
आइए हमारे कर्मचारी कोड से निम्नलिखित पंक्तियों को देखें:

 // Let's initialize a socket to receive messages. void* receiver = zmq_socket(context, ZMQ_PULL); zmq_connect(receiver, "tcp://localhost:4040"); 

ZMQ_PULL सॉकेट

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

ZMQ_PUSH सॉकेट

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

ZeroMQ प्रसंग प्राप्त करना

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

ZeroMQ प्रसंग विनाशक

प्रत्येक एप्लिकेशन के अंत में, आपको उस संदर्भ को नष्ट करने की आवश्यकता है जिसे आपने zmq_ctx_destroy () कॉल करके बनाया है। zmq_ctx_destroy () कॉल करने के बाद, सभी प्रोसेस एक एरर कोड ( ETERM ), zmq_ctx_destroy () ब्लॉक को सॉकेट खोलने के लिए ब्लॉक करते हैं और zmq_close () कॉल करके उन्हें बंद कर देते हैं।

सफाई

जब आप प्रोग्रामिंग भाषाओं जैसे कि पायथन या जावा में प्रोग्राम करते हैं, तो आपको स्मृति प्रबंधन के बारे में चिंता करने की आवश्यकता नहीं है, क्योंकि इन भाषाओं में अंतर्निहित कचरा कलेक्टर हैं।
उदाहरण के लिए, पायथन संदर्भ गिनती का उपयोग करता है, जब काउंटर शून्य हो जाता है, तो मेमोरी स्वचालित रूप से मुक्त हो जाती है। इसलिए, आपको पिहटन में ZeroMQ एप्लिकेशन लिखते समय कनेक्शन को स्पष्ट रूप से बंद नहीं करना चाहिए, क्योंकि यह ऑब्जेक्ट के संदर्भ गणना के शून्य होते ही स्वचालित रूप से बंद हो जाएगा। हालांकि, यह ध्यान दिया जाना चाहिए कि यह Jython, PyPy, या IronPython में काम नहीं करेगा। वैसे भी, आप पायथन प्रलेखन में पर्याप्त जानकारी पा सकते हैं। आइये अपने मुख्य कार्य पर वापस आते हैं।
जब आप सी में लिखते हैं, तो स्मृति प्रबंधन पूरी तरह से आपकी जिम्मेदारी है। अन्यथा, आपके पास एक अस्थिर एप्लिकेशन होगा जिसमें मेमोरी लीक होगी।
आपको सॉकेट्स को बंद करने, संदेशों को नष्ट करने और स्वयं ZeroMQ संदर्भ का ध्यान रखना चाहिए। आवेदन को सफलतापूर्वक पूरा करने के लिए कई बातों पर विचार करना होगा:

• जैसा कि पहले उल्लेख किया गया है, एप्लिकेशन को बंद करने के लिए आपको zmq_ctx_destroy () पर कॉल करके ZeroMQ संदर्भ को नष्ट करने की आवश्यकता है। हालांकि, अगर खुली कुर्सियां ​​हैं, तो zmq_ctx_destroy अनिश्चित काल तक उनके बंद होने की प्रतीक्षा कर सकता है। इस प्रकार, आपको पहले सभी सॉकेट्स को बंद करना होगा, और फिर संदर्भ को नष्ट करने के लिए zmq_ctx_destroy () कॉल करें।
• zmq_ctx_destroy () अनिश्चित काल तक प्रतीक्षा करेगा यदि कोई खुला कनेक्शन है या भेजने के लिए कतार में संदेश हैं।
• जब भी आप किसी संदेश को संसाधित करना समाप्त करते हैं, तो आपको zmq_msg_close () को कॉल करके इसे तुरंत बंद करना होगा, अन्यथा आपके एप्लिकेशन में मेमोरी लीक होगी।
• कई कुर्सियां ​​न खोलें। यदि आप ऐसा करते हैं, तो इसका मतलब यह है कि आप कुछ गलत कर रहे हैं और बेहतर तरीके से अपने आवेदन को खरोंच से डिजाइन कर रहे हैं।

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

मेमोरी लीक का पता लगाना

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

वेलग्रिंड का परिचय

आप डिबगिंग जानकारी प्रदर्शित करने के लिए –g विकल्प का उपयोग करके अपने एप्लिकेशन को संकलित कर सकते हैं। इस स्थिति में, त्रुटि संदेशों में सटीक लाइन नंबर होंगे।
निम्नलिखित उदाहरण पर विचार करें:

 #include <stdio.h> #include <stdlib.h> int main(int argc, char const *argv[]) { char* a = malloc(4); int b; printf("b = %d\n", b); return 0; } 

gcc –g –o test test.c लिखकर gcc –g –o test test.c में संकलित gcc –g –o test test.c अब मेमोरी लीक की जांच करने के लिए Valgrind चलाने का समय आ गया है। चलिए निम्नलिखित कमांड निष्पादित करते हैं:

 valgrind --leak-check=full --show-reachable=yes test 

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

 ==98190== Conditional jump or move depends on uninitialised value(s) ==98190== at 0x2D923: __vfprintf ==98190== by 0x4AC5A: vfprintf_l ==98190== by 0x952BE: printf ==98190== by 0x1F5E: main (test.c:8) ==98190== 4 bytes in 1 blocks are definitely lost in loss record 1 of 5 ==98190== at 0xF656: malloc (vg_replace_malloc.c:195) ==98190== by 0x1F46: main (test.c:6) ==98190== LEAK SUMMARY: ==98190== definitely lost: 4 bytes in 1 blocks ==98190== indirectly lost: 0 bytes in 0 blocks ==98190== possibly lost: 0 bytes in 0 blocks 

अब पिछले आउटपुट का थोड़ा वर्णन करते हैं:

• हम 98190 पर चर्चा नहीं करेंगे, क्योंकि यह प्रक्रिया आईडी है
• Conditional jump or move depends on uninitialised value(s) है, जिसका अर्थ है कि हमारे कोड में प्रारंभ सफल रहा था
• definitely lost मतलब है कि कोई स्मृति रिसाव है और हमें इसे ठीक करने की आवश्यकता है
• indirectly lost एक ब्लॉक को दर्शाता है जो खो सकता है
• possibly lost मतलब है कि एक संभावित मेमोरी लीक है

डिफ़ॉल्ट रूप से, Valgrind $PREFIX/lib/valgrind/default.supp का उपयोग करता है। हालाँकि, हमें ZeroMQ में उपयोग के लिए अपनी खुद की फ़ाइल बनानी चाहिए, यह कुछ इस तरह दिखाई देगा:

 { <socketcall_sendto> Memcheck:Param socketcall.sendto(msg) fun:send ... } { <socketcall_sendto> Memcheck:Param socketcall.send(msg) fun:send ... } 

फिर हम Valgrind को उन तर्कों के साथ चला सकते हैं जिनकी हमें आवश्यकता है:

 valgrind --leak-check=full --show-reachable=yes --suppressions=zeromq.supp server 

निष्कर्ष

इस अध्याय में, हमने सॉकेट्स के साथ मुलाकात की और दो नए मॉडल पेश किए, अर्थात् क्लाइंट-सर्वर मॉडल और पाइपलाइन पैटर्न। हमने इन टेम्पलेट्स का उपयोग करके कुछ समस्याओं को हल करने के तरीके पर भी चर्चा की, और इसे सरल उदाहरणों के साथ दिखाया। Valgrind.

, .