Redis स्ट्रीम हा Redis मध्ये आवृत्ती 5.0 सह सादर केलेला नवीन अमूर्त डेटा प्रकार आहे
संकल्पनेनुसार, रेडिस स्ट्रीम ही एक सूची आहे ज्यामध्ये तुम्ही नोंदी जोडू शकता. प्रत्येक एंट्रीला एक युनिक आयडेंटिफायर असतो. डीफॉल्टनुसार, आयडी स्वयंचलितपणे व्युत्पन्न केला जातो आणि त्यात टाइमस्टॅम्प समाविष्ट असतो. त्यामुळे, तुम्ही कालांतराने रेकॉर्डच्या रेंजची क्वेरी करू शकता किंवा प्रवाहात आल्यावर नवीन डेटा प्राप्त करू शकता, जसे की Unix "tail -f" कमांड लॉग फाइल वाचते आणि नवीन डेटाची प्रतीक्षा करत असताना फ्रीझ होते. लक्षात घ्या की अनेक क्लायंट एकाच वेळी थ्रेड ऐकू शकतात, ज्याप्रमाणे अनेक "tail -f" प्रक्रिया एकमेकांशी विरोधाभास न करता एकाच वेळी फाइल वाचू शकतात.

नवीन डेटा प्रकाराचे सर्व फायदे समजून घेण्यासाठी, रेडिस स्ट्रीमच्या कार्यक्षमतेची अंशतः प्रतिकृती बनवणार्‍या दीर्घकाळापासून अस्तित्वात असलेल्या रेडिस स्ट्रक्चर्सकडे एक द्रुत नजर टाकूया.

रेडिस PUB/SUB

Redis Pub/Sub ही एक साधी मेसेजिंग सिस्टीम आहे जी तुमच्या की-व्हॅल्यू स्टोअरमध्ये आधीच तयार केलेली आहे. तथापि, साधेपणा किंमतीवर येतो:

काही कारणास्तव प्रकाशक अयशस्वी झाल्यास, तो त्याचे सर्व सदस्य गमावतो
प्रकाशकाला त्याच्या सर्व सदस्यांचा अचूक पत्ता माहित असणे आवश्यक आहे
जर डेटा प्रक्रिया होण्यापेक्षा जलद प्रकाशित झाला तर प्रकाशक त्याच्या सदस्यांना कामासह ओव्हरलोड करू शकतो
किती सदस्यांना तो वितरित केला गेला आणि ते या संदेशावर किती लवकर प्रक्रिया करू शकले याची पर्वा न करता, प्रकाशनानंतर लगेचच प्रकाशकाच्या बफरमधून संदेश हटविला जातो.
सर्व सदस्यांना एकाच वेळी संदेश प्राप्त होईल. समान संदेशावर प्रक्रिया करण्याच्या क्रमावर सदस्यांनी स्वतःमध्ये कसे तरी सहमत असणे आवश्यक आहे.
सदस्याने संदेशावर यशस्वीपणे प्रक्रिया केली आहे याची पुष्टी करण्यासाठी कोणतीही अंगभूत यंत्रणा नाही. जर एखाद्या सदस्याला संदेश प्राप्त झाला आणि प्रक्रियेदरम्यान क्रॅश झाला, तर प्रकाशकाला त्याबद्दल माहिती नसते.

Redis यादी

रेडिस लिस्ट ही डेटा स्ट्रक्चर आहे जी वाचलेल्या कमांडस ब्लॉक करण्यास समर्थन देते. तुम्ही सूचीच्या सुरुवातीपासून किंवा शेवटी संदेश जोडू आणि वाचू शकता. या संरचनेवर आधारित, आपण आपल्या वितरित प्रणालीसाठी एक चांगला स्टॅक किंवा रांग बनवू शकता आणि बर्याच बाबतीत हे पुरेसे असेल. रेडिस पब/सब मधील मुख्य फरक:

संदेश एका क्लायंटला वितरित केला जातो. प्रथम रीड-ब्लॉक केलेल्या क्लायंटला प्रथम डेटा प्राप्त होईल.
क्लिंटने प्रत्येक संदेशासाठी स्वतः वाचन ऑपरेशन सुरू केले पाहिजे. यादीला ग्राहकांबद्दल काहीही माहिती नाही.
संदेश जोपर्यंत कोणीतरी वाचत नाही किंवा स्पष्टपणे हटवत नाही तोपर्यंत ते संग्रहित केले जातात. डिस्कवर डेटा फ्लश करण्यासाठी तुम्ही Redis सर्व्हर कॉन्फिगर केल्यास, सिस्टमची विश्वासार्हता नाटकीयरित्या वाढते.

प्रवाहाचा परिचय

प्रवाहात एंट्री जोडत आहे

संघ XADD प्रवाहात नवीन एंट्री जोडते. रेकॉर्ड म्हणजे फक्त स्ट्रिंग नसते, त्यात एक किंवा अधिक की-व्हॅल्यू जोड्या असतात. अशा प्रकारे, प्रत्येक एंट्री आधीच संरचित आहे आणि CSV फाइलच्या संरचनेसारखी आहे.

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

वरील उदाहरणात, आम्ही प्रवाहात दोन फील्ड जोडतो (की) “मायस्ट्रीम”: “सेन्सर-आयडी” आणि “तापमान” या मूल्यांसह अनुक्रमे “1234” आणि “19.8”. दुसरा युक्तिवाद म्हणून, कमांड एक अभिज्ञापक घेते जो एंट्रीला नियुक्त केला जाईल - हा अभिज्ञापक प्रवाहातील प्रत्येक एंट्री अद्वितीयपणे ओळखतो. तथापि, या प्रकरणात आम्ही * उत्तीर्ण झालो कारण आम्हाला रेडिसने आमच्यासाठी नवीन आयडी जनरेट करायचा आहे. प्रत्येक नवीन आयडी वाढेल. त्यामुळे, प्रत्येक नवीन एंट्रीमध्ये मागील नोंदींच्या संदर्भात उच्च अभिज्ञापक असेल.

अभिज्ञापक स्वरूप

कमांडद्वारे एंट्री आयडी परत आला XADD, दोन भागांचा समावेश आहे:

{millisecondsTime}-{sequenceNumber}
मिलिसेकंद वेळ - युनिक्स वेळ मिलीसेकंदमध्ये (रेडीस सर्व्हर वेळ). तथापि, वर्तमान वेळ मागील रेकॉर्डिंगच्या वेळेपेक्षा समान किंवा कमी असल्यास, मागील रेकॉर्डिंगचा टाइमस्टॅम्प वापरला जातो. म्हणून, सर्व्हरची वेळ वेळेत परत गेल्यास, नवीन अभिज्ञापक अद्याप वाढीव गुणधर्म राखून ठेवेल.

क्रमसंख्या समान मिलिसेकंदमध्ये तयार केलेल्या रेकॉर्डसाठी वापरले जाते. क्रमसंख्या मागील एंट्रीच्या तुलनेत 1 ने वाढवले जाईल. कारण द क्रमसंख्या 64 बिट्सचा आकार आहे, तर सराव मध्ये तुम्ही एका मिलिसेकंदात तयार होणाऱ्या रेकॉर्डच्या संख्येवर मर्यादा घालू नये.

अशा अभिज्ञापकांचे स्वरूप पहिल्या दृष्टीक्षेपात विचित्र वाटू शकते. अविश्वासू वाचकाला कदाचित आश्चर्य वाटेल की वेळ हा अभिज्ञापकाचा भाग का आहे. याचे कारण असे आहे की Redis प्रवाह ID द्वारे श्रेणी क्वेरीस समर्थन देतात. अभिज्ञापक रेकॉर्ड तयार केलेल्या वेळेशी संबंधित असल्याने, यामुळे वेळ श्रेणींची क्वेरी करणे शक्य होते. जेव्हा आपण कमांड पाहतो तेव्हा आपण विशिष्ट उदाहरण पाहू 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 पर्यायाचे समर्थन करते. प्रमाण निर्दिष्ट करून, आम्ही फक्त प्रथम N रेकॉर्ड मिळवू शकतो. आम्हाला पुढील एन रेकॉर्ड (पृष्ठांकन) मिळवायचे असल्यास, आम्ही शेवटचा प्राप्त आयडी वापरू शकतो, तो वाढवू शकतो क्रमसंख्या एक करून आणि पुन्हा विचारा. हे पुढील उदाहरणात पाहू. आम्ही 10 घटक जोडण्यास प्रारंभ करतो XADD (मायस्ट्रीम आधीच 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(log(N)) आणि नंतर M घटक परत करण्यासाठी O(M) आहे, तर प्रत्येक पुनरावृत्ती चरण जलद आहे. अशा प्रकारे, वापरणे XRANGE प्रवाह कार्यक्षमतेने पुनरावृत्ती होऊ शकतात.

संघ XREVRANGE समतुल्य आहे XRANGE, परंतु घटक उलट क्रमाने परत करते:

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

कृपया लक्षात ठेवा की आज्ञा XREVRANGE श्रेणी वितर्क उलट क्रमाने सुरू आणि थांबवते.

XREAD वापरून नवीन नोंदी वाचत आहे

बर्‍याचदा कार्य प्रवाहाची सदस्यता घेणे आणि फक्त नवीन संदेश प्राप्त करणे उद्भवते. ही संकल्पना रेडिस पब/सब किंवा रेडिस लिस्ट ब्लॉक करण्यासारखी वाटू शकते, परंतु रेडिस स्ट्रीम कसा वापरायचा यात मूलभूत फरक आहेत:

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

तुम्ही थ्रेडची सदस्यता घेऊ शकता आणि कमांड वापरून नवीन संदेश प्राप्त करू शकता 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 पर्याय ऐच्छिक आहे. खरं तर, फक्त आवश्यक कमांड पर्याय म्हणजे स्ट्रीम्स पर्याय, जो संबंधित कमाल अभिज्ञापकासह प्रवाहांची सूची निर्दिष्ट करतो. आम्ही "स्ट्रीम्स मायस्ट्रीम 0" लिहिले - आम्हाला मायस्ट्रीम स्ट्रीमचे सर्व रेकॉर्ड "0-0" पेक्षा जास्त अभिज्ञापकासह प्राप्त करायचे आहेत. जसे तुम्ही उदाहरणावरून पाहू शकता, कमांड थ्रेडचे नाव देते कारण आम्ही एकाच वेळी अनेक थ्रेड्सचे सदस्यत्व घेऊ शकतो. आम्ही लिहू शकतो, उदाहरणार्थ, "STREAMS mystream otherstream 0 0". कृपया लक्षात घ्या की स्ट्रीम्स पर्यायानंतर आम्हाला प्रथम सर्व आवश्यक प्रवाहांची नावे आणि त्यानंतरच ओळखकर्त्यांची यादी प्रदान करणे आवश्यक आहे.

या सोप्या फॉर्ममध्ये कमांड काही विशेष करत नाही XRANGE. तथापि, मनोरंजक गोष्ट अशी आहे की आपण सहजपणे वळू शकतो XREAD ब्लॉक आर्ग्युमेंट निर्दिष्ट करून, ब्लॉकिंग कमांडवर:

> XREAD BLOCK 0 STREAMS mystream $
वरील उदाहरणामध्ये, एक नवीन ब्लॉक पर्याय 0 मिलीसेकंदांच्या कालबाह्यतेसह निर्दिष्ट केला आहे (याचा अर्थ अनिश्चित काळासाठी प्रतीक्षा करणे). शिवाय, मायस्ट्रीम स्ट्रीमसाठी नेहमीचा आयडेंटिफायर पास करण्याऐवजी, एक विशेष ओळखकर्ता $ पास केला गेला. हा विशेष ओळखकर्ता म्हणजे XREAD ओळखकर्ता म्हणून mystream मध्ये जास्तीत जास्त अभिज्ञापक वापरणे आवश्यक आहे. त्यामुळे ज्या क्षणापासून आम्ही ऐकायला सुरुवात केली त्या क्षणापासून आम्हाला नवीन संदेश प्राप्त होतील. काही प्रकारे हे Unix "tail -f" कमांड सारखे आहे.

लक्षात ठेवा की BLOCK पर्याय वापरताना आम्हाला विशेष ओळखकर्ता $ वापरण्याची गरज नाही. आम्ही प्रवाहात अस्तित्वात असलेला कोणताही अभिज्ञापक वापरू शकतो. जर टीम ब्लॉक न करता आमची विनंती तात्काळ पूर्ण करू शकत असेल, तर ते तसे करेल, अन्यथा ते ब्लॉक करेल.

अवरोधित करणे 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
हा परिणाम साध्य करण्यासाठी, रेडिस स्ट्रीम कन्झ्युमर ग्रुप नावाची संकल्पना वापरते. ही संकल्पना छद्म-ग्राहक सारखीच आहे, जी प्रवाहाकडून डेटा प्राप्त करते, परंतु प्रत्यक्षात एका गटातील एकाधिक सदस्यांद्वारे सेवा दिली जाते, विशिष्ट हमी प्रदान करते:

प्रत्येक मेसेज ग्रुपमधील वेगळ्या सबस्क्राइबरला दिला जातो.
समूहामध्ये, सदस्यांना त्यांच्या नावाने ओळखले जाते, जी केस-संवेदनशील स्ट्रिंग आहे. जर एखादा सदस्य तात्पुरता गटातून बाहेर पडला, तर त्याला त्याचे स्वतःचे वेगळे नाव वापरून गटात पुनर्संचयित केले जाऊ शकते.
प्रत्येक ग्राहक गट "पहिला न वाचलेला संदेश" संकल्पनेचे अनुसरण करतो. जेव्हा एखादा सदस्य नवीन संदेशांची विनंती करतो तेव्हा तो फक्त तेच संदेश प्राप्त करू शकतो जे यापूर्वी कधीही गटातील कोणत्याही सदस्याला वितरित केले गेले नाहीत.
मेसेजवर सबस्क्राइबरने यशस्वीरित्या प्रक्रिया केली आहे याची स्पष्टपणे पुष्टी करण्यासाठी एक कमांड आहे. या आदेशाला कॉल करेपर्यंत, विनंती केलेला संदेश "प्रलंबित" स्थितीत राहील.
ग्राहक गटामध्ये, प्रत्येक सदस्य त्याला वितरित केलेल्या संदेशांच्या इतिहासाची विनंती करू शकतो, परंतु अद्याप प्रक्रिया केलेली नाही ("प्रलंबित" स्थितीत)

एका अर्थाने, गटाची स्थिती खालीलप्रमाणे व्यक्त केली जाऊ शकते:

+----------------------------------------+
| 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 आणि पर्यायी BLOCK पर्यायाला समर्थन देते. तथापि, एक आवश्यक GROUP पर्याय आहे जो नेहमी दोन युक्तिवादांसह निर्दिष्ट केला पाहिजे: गटाचे नाव आणि सदस्याचे नाव. 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.

पर्यायाशिवाय, ID 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"

मायग्रुपचा सदस्य असलेल्या बॉबने दोनपेक्षा जास्त संदेश मागितले नाहीत. कमांड स्पेशल आयडेंटिफायर ">" मुळे केवळ वितरित न झालेल्या संदेशांची तक्रार करते. जसे आपण पाहू शकता, "सफरचंद" संदेश प्रदर्शित केला जाणार नाही कारण तो आधीच अॅलिसला वितरित केला गेला आहे, म्हणून बॉबला "संत्रा" आणि "स्ट्रॉबेरी" प्राप्त होते.

अशा प्रकारे, अॅलिस, बॉब आणि गटातील इतर कोणतेही सदस्य एकाच प्रवाहातील भिन्न संदेश वाचू शकतात. ते त्यांचा प्रक्रिया न केलेल्या संदेशांचा इतिहास देखील वाचू शकतात किंवा संदेशांवर प्रक्रिया केलेले म्हणून चिन्हांकित करू शकतात.

लक्षात ठेवण्यासारख्या काही गोष्टी आहेत:

जितक्या लवकर ग्राहकाने संदेशास आज्ञा मानली 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}]]
{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:

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.

प्रवाह आकार मर्यादा

बर्‍याच ऍप्लिकेशन्सना डेटा कायमचा प्रवाहात संकलित करायचा नसतो. प्रत्येक थ्रेडला जास्तीत जास्त संदेशांची अनुमती असणे हे सहसा उपयुक्त असते. इतर प्रकरणांमध्ये, जेव्हा निर्दिष्ट थ्रेड आकार गाठला जातो तेव्हा सर्व संदेश थ्रेडमधून दुसर्‍या पर्सिस्टंट स्टोअरमध्ये हलवणे उपयुक्त आहे. कमांडमधील 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 रेकॉर्ड संग्रहित करण्‍याची आमची इच्छा आहे. हे रेडिसमध्ये मेमरी व्यवस्थापन अधिक कार्यक्षम करते.

एक वेगळी टीमही आहे XTRIM, जे समान गोष्ट करते:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

सतत स्टोरेज आणि प्रतिकृती

रेडिस स्ट्रीमची स्लेव्ह नोड्सवर अतुल्यकालिकपणे प्रतिकृती बनविली जाते आणि 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 कॉल शेवटचा घटक काढून टाकेल तेव्हा क्रमवारी केलेला संच पूर्णपणे काढून टाकला जाईल. त्याऐवजी, आत कोणतेही घटक नसतानाही थ्रेडना मेमरीमध्ये राहण्याची परवानगी आहे.

निष्कर्ष

मेसेज ब्रोकर, मेसेज क्यू, युनिफाइड लॉगिंग आणि हिस्ट्री-कीपिंग चॅट सिस्टम तयार करण्यासाठी रेडिस स्ट्रीम आदर्श आहे.

मी एकदा म्हटल्याप्रमाणे निकलॉस विर्थ, प्रोग्राम हे अल्गोरिदम आणि डेटा स्ट्रक्चर्स आहेत आणि रेडिस तुम्हाला दोन्ही आधीच देते.

स्त्रोत: www.habr.com

रेडिस स्ट्रीम - तुमच्या मेसेजिंग सिस्टमची विश्वासार्हता आणि स्केलेबिलिटी