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

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

रेडिस पब/उप

रेडिस पब/सब एक सरल मैसेजिंग सिस्टम है जो पहले से ही आपके की-वैल्यू स्टोर में निर्मित है। हालाँकि, सरलता की एक कीमत चुकानी पड़ती है:

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

रेडिस सूची

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

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

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

किसी स्ट्रीम में एक प्रविष्टि जोड़ना

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

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

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

पहचानकर्ता प्रारूप

कमांड द्वारा लौटाई गई प्रविष्टि आईडी XADD, दो भागों से मिलकर बना है:

{millisecondsTime}-{sequenceNumber}
मिलीसेकंडसमय - मिलीसेकंड में यूनिक्स समय (रेडिस सर्वर समय)। हालाँकि, यदि वर्तमान समय पिछली रिकॉर्डिंग के समय के समान या कम है, तो पिछली रिकॉर्डिंग के टाइमस्टैम्प का उपयोग किया जाता है। इसलिए, यदि सर्वर का समय समय से पीछे चला जाता है, तो नया पहचानकर्ता अभी भी वृद्धि गुण को बरकरार रखेगा।

क्रम संख्या एक ही मिलीसेकंड में बनाए गए रिकॉर्ड के लिए उपयोग किया जाता है। क्रम संख्या पिछली प्रविष्टि के सापेक्ष 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 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 विकल्प का समर्थन करता है. मात्रा निर्दिष्ट करके, हम आसानी से पहला एन रिकॉर्ड प्राप्त कर सकते हैं। यदि हमें अगले एन रिकॉर्ड (पेजिनेशन) प्राप्त करने की आवश्यकता है, तो हम अंतिम प्राप्त आईडी का उपयोग कर सकते हैं, इसे बढ़ा सकते हैं क्रम संख्या एक करके फिर से पूछें। आइए इसे निम्नलिखित उदाहरण में देखें। हम 10 तत्वों को जोड़ना शुरू करते हैं XADD (यह मानते हुए कि मायस्ट्रीम पहले से ही 10 तत्वों से भरा हुआ था)। प्रति कमांड 2 तत्व प्राप्त करने की पुनरावृत्ति शुरू करने के लिए, हम पूरी श्रृंखला से शुरू करते हैं लेकिन 2 के बराबर COUNT के साथ।

> 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(log(N)) है और फिर M तत्वों को वापस करने के लिए O(M) है, तो प्रत्येक पुनरावृत्ति चरण तेज़ है। इस प्रकार, का उपयोग कर XRANGE धाराओं को कुशलतापूर्वक दोहराया जा सकता है।

टीम XREVRANGE समतुल्य है XRANGE, लेकिन तत्वों को उल्टे क्रम में लौटाता है:

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

कृपया ध्यान दें कि आदेश XREVRANGE रेंज तर्कों को उल्टे क्रम में प्रारंभ और बंद करता है।

XREAD का उपयोग करके नई प्रविष्टियाँ पढ़ना

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

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

आप कमांड का उपयोग करके किसी थ्रेड की सदस्यता ले सकते हैं और नए संदेश प्राप्त कर सकते हैं Xपढ़ें. यह उससे थोड़ा अधिक जटिल है 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"

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

इस सरल रूप में कमांड इसकी तुलना में कुछ खास काम नहीं करता है XRANGE. हालाँकि, दिलचस्प बात यह है कि हम आसानी से मुड़ सकते हैं Xपढ़ें एक ब्लॉकिंग कमांड के लिए, ब्लॉक तर्क निर्दिष्ट करते हुए:

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

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

ब्लॉक कर रहा है Xपढ़ें आप एक साथ कई थ्रेड भी सुन सकते हैं, आपको बस उनके नाम निर्दिष्ट करने होंगे। इस मामले में, कमांड डेटा प्राप्त करने वाली पहली स्ट्रीम का रिकॉर्ड लौटाएगा। किसी दिए गए थ्रेड के लिए ब्लॉक किए गए पहले ग्राहक को पहले डेटा प्राप्त होगा।

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

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

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

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
इस प्रभाव को प्राप्त करने के लिए, रेडिस स्ट्रीम उपभोक्ता समूह नामक एक अवधारणा का उपयोग करता है। यह अवधारणा एक छद्म-ग्राहक के समान है, जो एक स्ट्रीम से डेटा प्राप्त करता है, लेकिन वास्तव में एक समूह के भीतर कई ग्राहकों द्वारा परोसा जाता है, जो कुछ गारंटी प्रदान करता है:

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

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

+----------------------------------------+
| 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)                             
+----------------------------------------+

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

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

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

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

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

अब जब समूह बन गया है, तो हम तुरंत कमांड का उपयोग करके संदेश पढ़ना शुरू कर सकते हैं एक्सरीडग्रुप. यह कमांड काफी हद तक समान है Xपढ़ें और वैकल्पिक ब्लॉक विकल्प का समर्थन करता है। हालाँकि, एक आवश्यक ग्रुप विकल्प है जिसे हमेशा दो तर्कों के साथ निर्दिष्ट किया जाना चाहिए: समूह का नाम और ग्राहक का नाम। 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 या कोई अन्य वैध पहचानकर्ता निर्दिष्ट कर सकते हैं। इस मामले में आदेश एक्सरीडग्रुप आपको "लंबित" स्थिति के साथ संदेशों का इतिहास लौटाएगा जो निर्दिष्ट ग्राहक (ऐलिस) को वितरित किए गए थे लेकिन अभी तक कमांड का उपयोग करके स्वीकार नहीं किए गए हैं XACK.

हम विकल्प के बिना तुरंत आईडी 0 निर्दिष्ट करके इस व्यवहार का परीक्षण कर सकते हैं COUNT. हम केवल एक लंबित संदेश देखेंगे, यानी सेब संदेश:

> 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"

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

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

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

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

विफलता पुनर्प्राप्ति

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

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

> 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

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

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

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

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

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 विकल्प का उपयोग केवल संदेश आईडी वापस करने के लिए किया जा सकता है। यदि आप संदेश के विवरण में रुचि नहीं रखते हैं और सिस्टम प्रदर्शन बढ़ाना चाहते हैं तो यह उपयोगी है।

डिलिवरी काउंटर

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

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

धागा अवस्था

टीम 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.

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

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

> 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"

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

XADD mystream MAXLEN ~ 1000 * ... entry fields here ...
उपरोक्त उदाहरण में ~ तर्क का अर्थ है कि हमें स्ट्रीम की लंबाई को किसी विशिष्ट मान तक सीमित करने की आवश्यकता नहीं है। हमारे उदाहरण में, यह 1000 से बड़ी या उसके बराबर कोई भी संख्या हो सकती है (उदाहरण के लिए, 1000, 1010, या 1030)। हमने अभी स्पष्ट रूप से निर्दिष्ट किया है कि हम चाहते हैं कि हमारी स्ट्रीम कम से कम 1000 रिकॉर्ड संग्रहीत करे। यह Redis के अंदर मेमोरी प्रबंधन को अधिक कुशल बनाता है।

एक अलग टीम भी है एक्सटीआरआईएम, जो वही काम करता है:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

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

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

एक स्ट्रीम से अलग-अलग तत्वों को हटाना

मैसेज डिलीट करने के लिए एक खास कमांड है एक्सडीईएल. कमांड को हटाए जाने वाले संदेश आईडी के बाद थ्रेड का नाम मिलता है:

> 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 कॉल अंतिम तत्व को हटा देगा तो क्रमबद्ध सेट पूरी तरह से हटा दिया जाएगा। इसके बजाय, थ्रेड्स को बिना किसी तत्व के भी मेमोरी में बने रहने दिया जाता है।

निष्कर्ष

रेडिस स्ट्रीम संदेश दलाल, संदेश कतार, एकीकृत लॉगिंग और इतिहास-रखने वाली चैट सिस्टम बनाने के लिए आदर्श है।

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

स्रोत: www.habr.com

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