Redis Stream - संस्करण 5.0 की रिलीज के साथ Redis में पेश किया गया एक नया अमूर्त डेटा प्रकार है
वैचारिक रूप से, Redis Stream एक सूची है जिसमें आप प्रविष्टियाँ जोड़ सकते हैं। प्रत्येक प्रविष्टि में एक विशिष्ट पहचानकर्ता होता है। डिफ़ॉल्ट रूप से, एक पहचानकर्ता स्वचालित रूप से उत्पन्न होता है और इसमें टाइमस्टैम्प शामिल होता है। इसलिए, आप समय के अनुसार रिकॉर्डिंग रेंज का अनुरोध कर सकते हैं या नया डेटा प्राप्त कर सकते हैं क्योंकि यह स्ट्रीम में आता है, क्योंकि यूनिक्स टेल-ऑफ़ कमांड लॉग फ़ाइल को पढ़ता है और नए डेटा की प्रत्याशा में जमा देता है। कृपया ध्यान दें कि कई ग्राहक एक ही समय में स्ट्रीम सुन सकते हैं, क्योंकि कई "पूंछ-एफ" प्रक्रियाएं एक ही समय में एक फ़ाइल को पढ़ सकती हैं, बिना एक दूसरे के साथ विरोध किए।

नए डेटा प्रकार के सभी लाभों को समझने के लिए, आइए संक्षेप में लंबे समय से मौजूद Redis संरचनाओं को याद करते हैं जो Redis Stream की कार्यक्षमता को आंशिक रूप से दोहराते हैं।

ऐतिहासिक भ्रमण

रेडिस पब / उप

Redis Pub / Sub एक सरल मैसेजिंग सिस्टम है जो पहले से ही आपके कुंजी-मूल्य संग्रहण में बनाया गया है। हालांकि, सादगी के लिए आपको भुगतान करना होगा:

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

रेडिस सूची

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

संदेश एक ग्राहक को दिया जाता है। पढ़ने से अवरुद्ध पहला ग्राहक पहले डेटा प्राप्त करेगा।
क्लिंट को प्रत्येक संदेश के लिए एक रीड ऑपरेशन शुरू करना चाहिए। सूची ग्राहकों के बारे में कुछ नहीं जानता है।
संदेश तब तक संग्रहीत किए जाते हैं जब तक कोई उन्हें गिनता या हटाता नहीं है। यदि आप डिस्क को डेटा फ्लश करने के लिए एक रेडिस सर्वर सेट करते हैं, तो सिस्टम की विश्वसनीयता नाटकीय रूप से बढ़ जाती है।

स्ट्रीम का परिचय

एक धारा प्रविष्टि जोड़ना

XADD कमांड स्ट्रीम में एक नया रिकॉर्ड जोड़ता है। एक रिकॉर्ड सिर्फ एक स्ट्रिंग नहीं है, इसमें एक या अधिक कुंजी-मूल्य जोड़े शामिल हैं। इस प्रकार, प्रत्येक रिकॉर्ड पहले से ही संरचित है और एक CSV फ़ाइल की संरचना जैसा दिखता है।

> XADD mystream * sensor-id 1234 temperature 19.8 1518951480106-0

ऊपर दिए गए उदाहरण में, हम क्रमशः नाम (कुंजी) "मिस्ट्रीम": "सेंसर-आईडी" और "तापमान" के साथ दो फ़ील्ड्स को क्रमशः "1234" और "19.8" मान के साथ जोड़ते हैं। दूसरे तर्क के रूप में, कमांड उस पहचानकर्ता को स्वीकार करता है जिसे रिकॉर्ड को सौंपा जाएगा - यह पहचानकर्ता विशिष्ट रूप से स्ट्रीम में प्रत्येक रिकॉर्ड को पहचानता है। हालाँकि, इस मामले में, हम पास हो गए * क्योंकि हम चाहते हैं कि रेडिस हमारे लिए एक नया पहचानकर्ता उत्पन्न करे। प्रत्येक नए पहचानकर्ता में वृद्धि होगी। इसलिए, प्रत्येक नए रिकॉर्ड में पिछले रिकॉर्ड के संबंध में एक बड़ा पहचानकर्ता होगा।

आईडी प्रारूप

XADD कमांड द्वारा लौटाए गए रिकॉर्ड पहचानकर्ता में दो भाग होते हैं:

{millisecondsTime}-{sequenceNumber}

millisecondsTime - Unis time in milliseconds (Redis server time)। हालाँकि, यदि वर्तमान समय पिछले रिकॉर्ड के समय से कम या अधिक है, तो पिछले रिकॉर्ड के समय का उपयोग किया जाता है। इसलिए, यदि सर्वर का समय अतीत में वापस आ जाता है, तो नया पहचानकर्ता अभी भी वृद्धि की संपत्ति को बनाए रखेगा।

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

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

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

 > XADD somestream 0-1 field value 0-1 > XADD somestream 0-2 foo bar 0-2

कृपया ध्यान दें कि इस मामले में, आपको पहचानकर्ता में वृद्धि की निगरानी करनी चाहिए। हमारे उदाहरण में, न्यूनतम पहचानकर्ता "0-1" है, इसलिए टीम किसी अन्य पहचानकर्ता को स्वीकार नहीं करेगी जो "0-1" से कम या बराबर है।

 > XADD somestream 0-1 foo bar (error) ERR The ID specified in XADD is equal or smaller than the target stream top item

स्ट्रीम में रिकॉर्ड की संख्या

आप केवल XLEN कमांड का उपयोग करके किसी स्ट्रीम में रिकॉर्ड की संख्या प्राप्त कर सकते हैं। हमारे उदाहरण के लिए, यह कमांड निम्नलिखित मान लौटाएगा:

 > XLEN somestream (integer) 2

रेंज अनुरोध - XRANGE और XREVRANGE

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

 > XRANGE mystream - + 1) 1) 1518951480106-0 2) 1) "sensor-id" 2) "1234" 3) "temperature" 4) "19.8" 2) 1) 1518951482479-0 2) 1) "sensor-id" 2) "9999" 3) "temperature" 4) "18.2"

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

 > XRANGE mystream 1518951480106 1518951480107 1) 1) 1518951480106-0 2) 1) "sensor-id" 2) "1234" 3) "temperature" 4) "19.8"

इस रेंज में हमारा केवल एक रिकॉर्ड है, हालांकि वास्तविक डेटा सेट में लौटा हुआ परिणाम बहुत बड़ा हो सकता है। इस कारण से, XRANGE COUNT विकल्प का समर्थन करता है। मात्रा निर्दिष्ट करके, हम केवल पहला एन रिकॉर्ड प्राप्त कर सकते हैं। अगर हमें अगली एन प्रविष्टियाँ (पेजिनेशन) प्राप्त करने की आवश्यकता है, तो हम प्राप्त अंतिम पहचानकर्ता का उपयोग कर सकते हैं, उसके अनुक्रम को एक से बढ़ा सकते हैं और फिर से अनुरोध कर सकते हैं। आइए इसे निम्नलिखित उदाहरण में देखें। हम XADD का उपयोग करके 10 तत्वों को जोड़ना शुरू कर रहे हैं (मान लीजिए कि स्ट्रीम स्ट्रीम पहले ही 10 तत्वों से भर चुकी है)। चलना शुरू करने के लिए, प्रति कमांड 2 तत्व मिलते हैं, हम पूरी सीमा के साथ शुरू करते हैं, लेकिन COUNT 2 के बराबर।

 > XRANGE mystream - + COUNT 2 1) 1) 1519073278252-0 2) 1) "foo" 2) "value_1" 2) 1) 1519073279157-0 2) 1) "foo" 2) "value_2"

निम्नलिखित दो तत्वों के साथ पुनरावृत्ति जारी रखने के लिए, हमें प्राप्त अंतिम पहचानकर्ता का चयन करने की आवश्यकता है, जो कि 1519073279157-0 है, और अनुक्रम 1 में जोड़ें।
परिणामी पहचानकर्ता, इस मामले में 1519073279157-1, अब अगले XRANGE कॉल के लिए सीमा की शुरुआत में एक नए तर्क के रूप में इस्तेमाल किया जा सकता है:

 > XRANGE mystream 1519073279157-1 + COUNT 2 1) 1) 1519073280281-0 2) 1) "foo" 2) "value_3" 2) 1) 1519073281432-0 2) 1) "foo" 2) "value_4"

और इसी तरह। चूंकि XRANGE की जटिलता खोज करने के लिए O (लॉग (N)) है, और फिर M तत्वों को वापस करने के लिए O (M), प्रत्येक पुनरावृत्ति कदम तेज है। इस प्रकार, एक्सग्रे का उपयोग करके, कुशलता से प्रवाह को पुनरावृत्त करना संभव है।

XREVRANGE कमांड XRANGE के बराबर है, लेकिन तत्वों को रिवर्स ऑर्डर में लौटाता है:

 > XREVRANGE mystream + - COUNT 1 1) 1) 1519073287312-0 2) 1) "foo" 2) "value_10"

ध्यान दें कि XREVRANGE कमांड रिवर्स ऑर्डर में स्टार्ट और स्टॉप रेंज के तर्क लेती है।

XREAD के साथ नए रिकॉर्ड पढ़ना

अक्सर स्ट्रीम में सदस्यता लेने और केवल नए संदेश प्राप्त करने का कार्य होता है। यह अवधारणा Redis Pub / Sub या ब्लॉकिंग Redis List की तरह लग सकती है, लेकिन Redis Stream का उपयोग करने के तरीके में बुनियादी अंतर हैं:

प्रत्येक नए संदेश को प्रत्येक ग्राहक को डिफ़ॉल्ट रूप से दिया जाता है। यह व्यवहार रेडिस सूची को अवरुद्ध करने से अलग है, जहां एक नए संदेश को केवल एक ग्राहक द्वारा पढ़ा जाएगा।
जबकि Redis Pub / Sub में सभी संदेश भूल जाते हैं और कभी सहेजे नहीं जाते हैं, स्ट्रीम में सभी संदेशों को अनिश्चित काल तक संग्रहीत किया जाता है (जब तक कि ग्राहक स्पष्ट रूप से हटाने के लिए कॉल नहीं करता)।
Redis Stream आपको एक स्ट्रीम में संदेशों तक पहुंच को अलग करने की अनुमति देता है। एक विशिष्ट ग्राहक केवल अपना व्यक्तिगत संदेश इतिहास देख सकता है।

आप स्ट्रीम की सदस्यता ले सकते हैं और XREAD कमांड का उपयोग करके नए संदेश प्राप्त कर सकते हैं। यह XRANGE की तुलना में थोड़ा अधिक जटिल है, इसलिए हम पहले सरल उदाहरणों के साथ शुरू करेंगे।

 > XREAD COUNT 2 STREAMS mystream 0 1) 1) "mystream" 2) 1) 1) 1519073278252-0 2) 1) "foo" 2) "value_1" 2) 1) 1519073279157-0 2) 1) "foo" 2) "value_2"

ऊपर दिए गए उदाहरण में, एक गैर-अवरुद्ध XREAD फ़ॉर्म निर्दिष्ट किया गया है । कृपया ध्यान दें कि COUNT विकल्प वैकल्पिक है। वास्तव में, केवल आवश्यक कमांड विकल्प STREAMS विकल्प है, जो संगत अधिकतम पहचानकर्ता के साथ धाराओं की सूची सेट करता है। हमने लिखा है "STREAMS mystream 0" - हम मिस्ट्रीम स्ट्रीम के सभी रिकॉर्ड "0-0" से अधिक पहचानकर्ता के साथ प्राप्त करना चाहते हैं। जैसा कि आप उदाहरण से देख सकते हैं, कमांड स्ट्रीम का नाम देता है, क्योंकि हम एक ही समय में कई थ्रेड्स की सदस्यता ले सकते हैं। हम लिख सकते हैं, उदाहरण के लिए, "STREAMS मिस्ट्रीम अन्यस्ट्रीम 0 0"। कृपया ध्यान दें कि STREAMS विकल्प के बाद, हमें पहले सभी आवश्यक धाराओं के नाम और उसके बाद ही पहचानकर्ताओं की एक सूची प्रदान करनी होगी।

इस सरल रूप में, आदेश XRANGE की तुलना में कुछ खास नहीं करता है। हालाँकि, दिलचस्प बात यह है कि हम XREAD को ब्लॉक कमांड में आसानी से बदल सकते हैं।

> XREAD BLOCK 0 STREAMS mystream $

ऊपर दिए गए उदाहरण में, 0 मिलीसेकंड के टाइमआउट के साथ एक नया ब्लॉक विकल्प निर्दिष्ट किया गया है (इसका अर्थ है अंतहीन प्रतीक्षा)। इसके अलावा, मिस्ट्रीम स्ट्रीम के लिए सामान्य पहचानकर्ता को पारित करने के बजाय, विशेष पहचानकर्ता $ पारित किया गया था। इस विशेष पहचानकर्ता का अर्थ है कि XREAD को पहचानकर्ता के रूप में रहस्य धारा में अधिकतम पहचानकर्ता का उपयोग करना चाहिए। इसलिए हम केवल नए संदेश प्राप्त करेंगे, जिस क्षण से हमने सुनना शुरू किया था। एक तरह से, यह यूनिक्स टेल-एफ कमांड के समान है।

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

XREAD को अवरुद्ध करना एक साथ कई धाराओं को सुन सकता है, आपको बस उनके नाम निर्दिष्ट करने की आवश्यकता है। इस स्थिति में, कमांड पहली स्ट्रीम का रिकॉर्ड लौटाएगा जिसमें डेटा आया था। इस स्ट्रीम के लिए अवरुद्ध पहला ग्राहक पहले डेटा प्राप्त करेगा।

उपभोक्ता समूह

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

यदि हम कल्पना करते हैं कि हमारे पास तीन ग्राहक C1, C2, C3 और एक धारा है जिसमें 1, 2, 3, 4, 5, 6, 7 संदेश शामिल हैं, तो नीचे दिए गए चित्र में भी संदेश सेवा होगी:

1 -> C1
2 -> C2
3 -> C3
4 -> C1
5 -> C2
6 -> C3
7 -> C1

इस प्रभाव को प्राप्त करने के लिए, Redis Stream उपभोक्ता समूह नामक एक अवधारणा का उपयोग करता है। यह अवधारणा एक छद्म ग्राहक के समान है जो एक धारा से डेटा प्राप्त करता है, लेकिन वास्तव में एक समूह के भीतर कई ग्राहकों द्वारा प्रदान किया जाता है, कुछ गारंटी प्रदान करता है:

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

एक अर्थ में, एक समूह की स्थिति का प्रतिनिधित्व इस प्रकार किया जा सकता है:

 +----------------------------------------+ | consumer_group_name: mygroup | consumer_group_stream: somekey | last_delivered_id: 1292309234234-92 | | consumers: | "consumer-1" with pending messages | 1292309234234-4 | 1292309234232-8 | "consumer-42" with pending messages | ... (and so forth) +----------------------------------------+

अब उपभोक्ता समूह के लिए मुख्य टीमों से परिचित होने का समय है, अर्थात्:

XGROUP का उपयोग समूहों को बनाने, नष्ट करने और प्रबंधित करने के लिए किया जाता है।
XREADGROUP का उपयोग किसी समूह के माध्यम से एक धारा को पढ़ने के लिए किया जाता है।
XACK - यह कमांड सब्सक्राइबर को संदेश को सफलतापूर्वक संसाधित करने की अनुमति देता है

उपभोक्ता समूह का निर्माण

मान लीजिए कि एक रहस्य धारा पहले से मौजूद है। तब समूह निर्माण आदेश इस तरह दिखेगा:

> XGROUP CREATE mystream mygroup $
OK

समूह बनाते समय, हमें एक पहचानकर्ता पास करना होगा, जिसके साथ समूह संदेश प्राप्त करेगा। यदि हम केवल सभी नए संदेश प्राप्त करना चाहते हैं, तो हम विशेष पहचानकर्ता $ (जैसा कि ऊपर हमारे उदाहरण में है) का उपयोग कर सकते हैं। यदि आप एक विशेष पहचानकर्ता के बजाय 0 निर्दिष्ट करते हैं, तो स्ट्रीम के सभी संदेश समूह के लिए उपलब्ध होंगे।

अब जब समूह बनाया जाता है, तो हम तुरंत XREADGROUP कमांड का उपयोग करके संदेश पढ़ना शुरू कर सकते हैं। यह कमांड XREAD के समान है और वैकल्पिक ब्लॉक विकल्प का समर्थन करता है। हालांकि, एक अनिवार्य समूह विकल्प है, जिसे हमेशा दो तर्कों के साथ निर्दिष्ट किया जाना चाहिए: समूह का नाम और ग्राहक का नाम। COUNT विकल्प भी समर्थित है।

स्ट्रीम पढ़ने से पहले, आइए वहाँ कुछ संदेश डालें:

 > XADD mystream * message apple 1526569495631-0 > XADD mystream * message orange 1526569498055-0 > XADD mystream * message strawberry 1526569506935-0 > XADD mystream * message apricot 1526569535168-0 > XADD mystream * message banana 1526569544280-0

अब समूह के माध्यम से इस धारा को पढ़ने का प्रयास करते हैं:

 > XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream > 1) 1) "mystream" 2) 1) 1) 1526569495631-0 2) 1) "message" 2) "apple"

उपरोक्त कमांड शब्दशः इस प्रकार है:

", ऐलिस के ग्राहक, माइग्रुप के सदस्य, उस रहस्य से एक संदेश पढ़ना चाहते हैं जो पहले कभी किसी के लिए नहीं हुआ।"

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

इसके अलावा, विशेष मामलों में, आप एक वास्तविक पहचानकर्ता निर्दिष्ट कर सकते हैं, जैसे 0 या कोई अन्य मान्य पहचानकर्ता। इस स्थिति में, XREADGROUP कमांड आपके पास "लंबित" स्थिति वाले संदेशों का इतिहास लौटाएगा, जो निर्दिष्ट ग्राहक (ऐलिस) को दिए गए थे, लेकिन अभी तक XACK कमांड का उपयोग करके पुष्टि नहीं की गई है।

हम बिना COUNT विकल्प के पहचानकर्ता 0 को तुरंत निर्दिष्ट करके इस व्यवहार को सत्यापित कर सकते हैं। हम केवल लंबित संदेश देखते हैं, अर्थात सेब वाला संदेश:

 > XREADGROUP GROUP mygroup Alice STREAMS mystream 0 1) 1) "mystream" 2) 1) 1) 1526569495631-0 2) 1) "message" 2) "apple"

हालाँकि, यदि हम संदेश को सफलतापूर्वक संसाधित किए जाने की पुष्टि करते हैं, तो इसे प्रदर्शित नहीं किया जाएगा:

 > XACK mystream mygroup 1526569495631-0 (integer) 1 > XREADGROUP GROUP mygroup Alice STREAMS mystream 0 1) 1) "mystream" 2) (empty list or set)

अब कुछ पढ़ने के लिए बॉब की बारी है:

 > XREADGROUP GROUP mygroup Bob COUNT 2 STREAMS mystream > 1) 1) "mystream" 2) 1) 1) 1526569498055-0 2) 1) "message" 2) "orange" 2) 1) 1526569506935-0 2) 1) "message" 2) "strawberry"

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

इस प्रकार, ऐलिस, बॉब और किसी भी अन्य समूह के ग्राहक एक ही स्ट्रीम से विभिन्न संदेश पढ़ सकते हैं। वे अपने कच्चे संदेश के इतिहास को पढ़ सकते हैं या संसाधित किए गए संदेशों को चिह्नित कर सकते हैं।

ध्यान रखने योग्य कुछ बातें हैं:

जैसे ही ग्राहक संदेश को XREADGROUP कमांड मानता है , यह संदेश "लंबित" स्थिति में चला जाता है और इस विशेष ग्राहक को सौंपा जाता है। अन्य समूह ग्राहक इस संदेश को नहीं पढ़ पाएंगे।
सब्सक्राइबर पहले उल्लेख पर स्वचालित रूप से बनाए जाते हैं, उनके स्पष्ट निर्माण की कोई आवश्यकता नहीं है।
XREADGROUP के साथ , आप एक ही समय में कई अलग-अलग धाराओं के संदेश पढ़ सकते हैं, हालाँकि, इस कार्य के लिए, आपको पहले XGROUP का उपयोग करके प्रत्येक स्ट्रीम के लिए समान नाम वाले समूह बनाने होंगे।

क्रैश रिकवरी

ग्राहक विफलता से उबर सकता है और "लंबित" की स्थिति के साथ अपने संदेशों की सूची को फिर से पढ़ सकता है। हालांकि, वास्तविक दुनिया में, ग्राहक अंततः विफल हो सकते हैं। यदि विफलता के बाद वह ठीक नहीं हो सकता है तो ग्राहक के झूलने वाले संदेश का क्या होगा?
उपभोक्ता समूह एक सुविधा प्रदान करता है जो विशेष रूप से ऐसे मामलों के लिए उपयोग की जाती है - जब आपको संदेशों के स्वामी को बदलने की आवश्यकता होती है।

सबसे पहले, आपको XPENDING कमांड को कॉल करने की आवश्यकता है, जो "लंबित" की स्थिति के साथ समूह के सभी संदेशों को प्रदर्शित करता है। अपने सरलतम रूप में, केवल दो तर्कों के साथ एक कमांड को बुलाया जाता है: स्ट्रीम का नाम और समूह का नाम:

 > XPENDING mystream mygroup 1) (integer) 2 2) 1526569498055-0 3) 1526569506935-0 4) 1) 1) "Bob" 2) "2"

टीम ने पूरे समूह के लिए और प्रत्येक ग्राहक के लिए अप्रमाणित संदेशों की संख्या मुद्रित की। हमारे पास केवल दो असंसाधित संदेशों वाले बॉब हैं, क्योंकि ऐलिस द्वारा अनुरोध किए गए एकमात्र संदेश की पुष्टि XACK के साथ की गई थी।

हम अधिक तर्कों का उपयोग करके अतिरिक्त जानकारी का अनुरोध कर सकते हैं:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]

{स्टार्ट-आईडी} {एंड-आईडी} - पहचानकर्ताओं की श्रेणी (आप "-" और "+" का उपयोग कर सकते हैं)
{गिनती} - प्रसव के प्रयासों की संख्या
{उपभोक्ता-नाम} - समूह का नाम

 > XPENDING mystream mygroup - + 10 1) 1) 1526569498055-0 2) "Bob" 3) (integer) 74170458 4) (integer) 1 2) 1) 1526569506935-0 2) "Bob" 3) (integer) 74170458 4) (integer) 1

अब हमारे पास प्रत्येक संदेश का विवरण है: पहचानकर्ता, ग्राहक का नाम, मिलीसेकंड में डाउनटाइम, और अंत में, डिलीवरी के प्रयासों की संख्या। हमारे पास बॉब से दो संदेश हैं, और वे लगभग 20 घंटे के लिए 74170458 मिलीसेकंड के लिए निष्क्रिय हैं।

कृपया ध्यान दें कि कोई भी हमें यह जाँचने से नहीं रोक रहा है कि केवल XRANGE का उपयोग करके संदेश की सामग्री क्या है।

 > XRANGE mystream 1526569498055-0 1526569498055-0 1) 1) 1526569498055-0 2) 1) "message" 2) "orange"

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

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}

इस आदेश का उपयोग करके, हम एक "विदेशी" संदेश प्राप्त कर सकते हैं जो अभी तक स्वामी को {उपभोक्ता} में बदलकर संसाधित नहीं किया गया है। हालाँकि, हम न्यूनतम डाउनटाइम {min-idle-time} भी प्रदान कर सकते हैं। यह ऐसी स्थिति से बचने में मदद करता है जहां दो ग्राहक एक साथ एक ही संदेश के मालिक को बदलने की कोशिश करते हैं:

Client 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0
Clinet 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0

पहला ग्राहक डाउनटाइम रीसेट करेगा और डिलीवरी की संख्या के काउंटर को बढ़ाएगा। तो दूसरा ग्राहक यह अनुरोध करने में सक्षम नहीं होगा।

 > XCLAIM mystream mygroup Alice 3600000 1526569498055-0 1) 1) 1526569498055-0 2) 1) "message" 2) "orange"

संदेश का ऐलिस द्वारा सफलतापूर्वक दावा किया गया था, जो अब संदेश को संसाधित कर सकता है और इसे स्वीकार कर सकता है।

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

वितरण काउंटर

XPENDING आउटपुट में आपके द्वारा देखा गया काउंटर प्रत्येक संदेश की डिलीवरी की संख्या है। इस तरह के एक काउंटर को दो तरीकों से बढ़ाया जाता है: जब संदेश को XCLAIM के माध्यम से सफलतापूर्वक अनुरोध किया जाता है या जब XREADGROUP कॉल का उपयोग किया जाता है ।

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

धागे की स्थिति

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

 > XINFO STREAM mystream 1) length 2) (integer) 13 3) radix-tree-keys 4) (integer) 1 5) radix-tree-nodes 6) (integer) 2 7) groups 8) (integer) 2 9) first-entry 10) 1) 1524494395530-0 2) 1) "a" 2) "1" 3) "b" 4) "2" 11) last-entry 12) 1) 1526569544280-0 2) 1) "message" 2) "banana"

ऊपर दी गई कमांड निर्दिष्ट स्ट्रीम पर सामान्य जानकारी प्रदर्शित करती है। अब थोड़ा और जटिल उदाहरण:

 > XINFO GROUPS mystream 1) 1) name 2) "mygroup" 3) consumers 4) (integer) 2 5) pending 6) (integer) 2 2) 1) name 2) "some-other-group" 3) consumers 4) (integer) 1 5) pending 6) (integer) 0

ऊपर दी गई कमांड निर्दिष्ट स्ट्रीम के सभी समूहों के लिए सामान्य जानकारी प्रदर्शित करती है

 > XINFO CONSUMERS mystream mygroup 1) 1) name 2) "Alice" 3) pending 4) (integer) 1 5) idle 6) (integer) 9104628 2) 1) name 2) "Bob" 3) pending 4) (integer) 1 5) idle 6) (integer) 83841983

ऊपर दिया गया आदेश निर्दिष्ट स्ट्रीम और समूह के सभी ग्राहकों की जानकारी प्रदर्शित करता है।
यदि आप कमांड सिंटैक्स भूल जाते हैं, तो मदद के लिए कमांड से संपर्क करें:

 > XINFO HELP 1) XINFO {subcommand} arg arg ... arg. Subcommands are: 2) CONSUMERS {key} {groupname} -- Show consumer groups of group {groupname}. 3) GROUPS {key} -- Show the stream consumer groups. 4) STREAM {key} -- Show information about the stream. 5) HELP -- Print this help.

स्ट्रीम आकार सीमा

कई एप्लिकेशन हमेशा स्ट्रीम में डेटा एकत्र नहीं करना चाहते हैं। धारा में संदेशों की अधिकतम संख्या होना अक्सर उपयोगी होता है। अन्य मामलों में, निर्दिष्ट स्ट्रीम आकार तक पहुंचने पर धारा से अन्य संदेशों को दूसरे निरंतर भंडारण में स्थानांतरित करना उपयोगी होता है। आप XADD कमांड में MAXLEN पैरामीटर का उपयोग करके स्ट्रीम का आकार सीमित कर सकते हैं :

 > XADD mystream MAXLEN 2 * value 1 1526654998691-0 > XADD mystream MAXLEN 2 * value 2 1526654999635-0 > XADD mystream MAXLEN 2 * value 3 1526655000369-0 > XLEN mystream (integer) 2 > XRANGE mystream - + 1) 1) 1526654999635-0 2) 1) "value" 2) "2" 2) 1) 1526655000369-0 2) 1) "value" 2) "3"

मैक्सलेन का उपयोग करते समय, निर्दिष्ट लंबाई तक पहुंचने पर पुराने रिकॉर्ड स्वचालित रूप से हटा दिए जाते हैं, इसलिए स्ट्रीम में एक स्थिर आकार होता है। हालांकि, इस मामले में ट्रिमिंग रेडिस मेमोरी में सबसे अधिक उत्पादक तरीके से नहीं होती है। स्थिति में निम्नानुसार सुधार किया जा सकता है: ऊपर दिए गए उदाहरण में तर्क ~ का अर्थ है कि हमें एक विशिष्ट मान के लिए धारा की लंबाई को सीमित करने की आवश्यकता नहीं है। हमारे उदाहरण में, यह 1000 (उदाहरण के लिए, 1000, 1010 या 1030) से अधिक या इसके बराबर कोई भी संख्या हो सकती है। हमने केवल स्पष्ट रूप से संकेत दिया है कि हम चाहते हैं कि हमारी धारा कम से कम 1000 रिकॉर्ड संग्रहीत करें। यह मेमोरी के साथ Redis के अंदर अधिक कुशल कार्य करता है। एक अलग XTRIM कमांड भी है जो समान कार्य करता है:

XADD mystream MAXLEN ~ 1000 * ... entry fields here ...

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

लगातार भंडारण और प्रतिकृति

Redis Stream को एसिंक्रोनस रूप से दास नोड्स के रूप में दोहराया जाता है और AOF (सभी डेटा का स्नैपशॉट) और RDB (सभी लेखन कार्यों का लॉग) जैसी फ़ाइलों को सहेजा जाता है। उपभोक्ता समूह राज्य प्रतिकृति भी समर्थित है। इसलिए, यदि संदेश मास्टर नोड पर "लंबित" स्थिति में है, तो दास नोड्स पर इस संदेश की स्थिति समान होगी।

किसी स्ट्रीम से अलग-अलग आइटम हटाना

संदेशों को हटाने के लिए एक विशेष XDEL कमांड है । आदेश को स्ट्रीम का नाम मिलता है, उसके बाद उस संदेश के पहचानकर्ता को हटा दिया जाना चाहिए:

 > XRANGE mystream - + COUNT 2 1) 1) 1526654999635-0 2) 1) "value" 2) "2" 2) 1) 1526655000369-0 2) 1) "value" 2) "3" > XDEL mystream 1526654999635-0 (integer) 1 > XRANGE mystream - + COUNT 2 1) 1) 1526655000369-0 2) 1) "value" 2) "3"

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

शून्य-लंबाई की धाराएँ

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

निष्कर्ष

Redis Stream संदेश दलालों, संदेश कतारों, एकीकृत लॉग और चैट सिस्टम बनाने के लिए आदर्श है जो इतिहास को संग्रहीत करता है।

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

रेडिस स्ट्रीम - आपके मैसेजिंग सिस्टम की विश्वसनीयता और स्केलेबिलिटी