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

नयाँ डेटा प्रकारका सबै फाइदाहरू बुझ्नको लागि, रेडिस स्ट्रिमको कार्यक्षमतालाई आंशिक रूपमा नक्कल गर्ने लामो-अवस्थित Redis संरचनाहरूमा द्रुत नजर राखौं।

Redis PUB/SUB

Redis Pub/Sub एक साधारण सन्देश प्रणाली हो जुन पहिले नै तपाइँको मुख्य मूल्य स्टोर मा निर्मित छ। यद्यपि, सादगी लागतमा आउँछ:

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

Redis सूची

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

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

स्ट्रिमको परिचय

स्ट्रिममा प्रविष्टि थप्दै

टोली XADD स्ट्रिममा नयाँ प्रविष्टि थप्छ। रेकर्ड एउटा स्ट्रिङ मात्र होइन, यसमा एक वा बढी कुञ्जी-मान जोडीहरू हुन्छन्। यसरी, प्रत्येक प्रविष्टि पहिले नै संरचित छ र CSV फाइलको संरचनासँग मिल्दोजुल्दो छ।

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

माथिको उदाहरणमा, हामीले नाम (कुञ्जी) "mystream" को साथ स्ट्रिममा दुई क्षेत्रहरू थप्छौं: "सेन्सर-आईडी" र "तापमान" क्रमशः "1234" र "19.8" मानहरूसँग। दोस्रो तर्कको रूपमा, आदेशले एक पहिचानकर्ता लिन्छ जुन प्रविष्टिमा तोकिन्छ - यो पहिचानकर्ताले स्ट्रिममा प्रत्येक प्रविष्टिलाई विशिष्ट रूपमा पहिचान गर्दछ। यद्यपि, यस अवस्थामा हामीले * पास गर्यौं किनभने हामी Redis ले हाम्रो लागि नयाँ ID उत्पन्न गरोस् भन्ने चाहन्छौं। प्रत्येक नयाँ आईडी बढ्नेछ। त्यसकारण, प्रत्येक नयाँ प्रविष्टिमा अघिल्लो प्रविष्टिहरूको सम्बन्धमा उच्च पहिचानकर्ता हुनेछ।

पहिचानकर्ता ढाँचा

आदेशद्वारा फर्काइएको प्रविष्टि ID XADD, दुई भागहरु मिलेर बनेको छ:

{millisecondsTime}-{sequenceNumber}
मिलिसेकेन्डसमय - मिलिसेकेन्डमा युनिक्स समय (Redis सर्भर समय)। यद्यपि, यदि हालको समय अघिल्लो रेकर्डिङको समय भन्दा समान वा कम छ भने, त्यसपछि अघिल्लो रेकर्डिङको टाइमस्ट्याम्प प्रयोग गरिन्छ। त्यसकारण, यदि सर्भर समय समयमा फिर्ता जान्छ भने, नयाँ पहिचानकर्ताले अझै पनि वृद्धि गुण कायम गर्नेछ।

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

त्यस्ता पहिचानकर्ताहरूको ढाँचा पहिलो नजरमा अनौठो लाग्न सक्छ। एक अविश्वासी पाठकले किन समय पहिचानकर्ताको भाग हो भनेर सोच्न सक्छ। कारण यो हो कि Redis स्ट्रिमहरूले ID द्वारा दायरा प्रश्नहरूलाई समर्थन गर्दछ। पहिचानकर्ता रेकर्ड सिर्जना भएको समयसँग सम्बन्धित भएकोले, यसले समय दायराहरू क्वेरी गर्न सम्भव बनाउँछ। हामी एक विशिष्ट उदाहरण हेर्नेछौं जब हामी आदेश हेर्छौं XRANGE.

यदि कुनै कारणको लागि प्रयोगकर्ताले आफ्नै पहिचानकर्ता निर्दिष्ट गर्न आवश्यक छ, जुन, उदाहरणका लागि, केहि बाह्य प्रणालीसँग सम्बन्धित छ, तब हामी यसलाई आदेशमा पास गर्न सक्छौं। XADD सट्टा * तल देखाइएको रूपमा:

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

कृपया ध्यान दिनुहोस् कि यस अवस्थामा तपाईंले आईडी वृद्धि आफैलाई निगरानी गर्नुपर्छ। हाम्रो उदाहरणमा, न्यूनतम पहिचानकर्ता "०-१" हो, त्यसैले आदेशले "०-१" बराबर वा कम भएको अर्को पहिचानकर्ता स्वीकार गर्दैन।

> 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 रेकर्डहरू प्राप्त गर्न सक्छौं। यदि हामीले अर्को 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"

अर्को दुई तत्वहरूसँग पुनरावृत्ति जारी राख्न, हामीले प्राप्त गरेको अन्तिम ID चयन गर्न आवश्यक छ, अर्थात् 1519073279157-0, र 1 मा थप्नुहोस्। अनुक्रम संख्या.
परिणामस्वरूप ID, यस अवस्थामा 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)) खोज्न र त्यसपछि O(M) ले M तत्वहरू फर्काउँछ, त्यसपछि प्रत्येक पुनरावृत्ति चरण छिटो हुन्छ। यसरी, प्रयोग गर्दै XRANGE स्ट्रिमहरू कुशलतापूर्वक दोहोर्याउन सकिन्छ।

टोली XREVRANGE बराबर छ XRANGE, तर उल्टो क्रममा तत्वहरू फर्काउँछ:

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

कृपया ध्यान दिनुहोस् कि आदेश XREVRANGE दायरा आर्गुमेन्टहरू सुरु हुन्छ र उल्टो क्रममा रोकिन्छ।

XREAD प्रयोग गरी नयाँ प्रविष्टिहरू पढ्दै

प्रायः कार्य स्ट्रिमको सदस्यता लिने र नयाँ सन्देशहरू मात्र प्राप्त गर्ने हो। यो अवधारणा Redis Pub/Sub वा Redis सूची अवरुद्ध गर्ने जस्तो लाग्न सक्छ, तर Redis स्ट्रिम कसरी प्रयोग गर्ने भन्नेमा आधारभूत भिन्नताहरू छन्:

प्रत्येक नयाँ सन्देश प्रत्येक ग्राहकलाई पूर्वनिर्धारित रूपमा डेलिभर गरिन्छ। यो व्यवहार अवरुद्ध Redis सूची भन्दा फरक छ, जहाँ एक नयाँ सन्देश केवल एक सदस्य द्वारा पढ्न सकिन्छ।
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 विकल्प वैकल्पिक छ। वास्तवमा, केवल आवश्यक आदेश विकल्प स्ट्रिम्स विकल्प हो, जसले सम्बन्धित अधिकतम पहिचानकर्ताको साथ स्ट्रिमहरूको सूची निर्दिष्ट गर्दछ। हामीले "स्ट्रिम्स माइस्ट्रिम ०" लेख्यौं - हामी माइस्ट्रिम स्ट्रिमका सबै रेकर्डहरू "०-०" भन्दा ठूलो पहिचानकर्ताको साथ प्राप्त गर्न चाहन्छौं। तपाईंले उदाहरणबाट देख्न सक्नुहुन्छ, आदेशले थ्रेडको नाम फर्काउँछ किनभने हामी एकै समयमा धेरै थ्रेडहरूको सदस्यता लिन सक्छौं। हामी लेख्न सक्छौं, उदाहरणका लागि, "स्ट्रीम्स माईस्ट्रीम अन्यस्ट्रीम ० ०"। कृपया ध्यान दिनुहोस् कि स्ट्रिम्स विकल्प पछि हामीले पहिले सबै आवश्यक स्ट्रिमहरूको नाम र त्यसपछि मात्र पहिचानकर्ताहरूको सूची प्रदान गर्न आवश्यक छ।

यस सरल फारममा आदेशले तुलनामा केहि विशेष गर्दैन XRANGE। यद्यपि, चाखलाग्दो कुरा यो हो कि हामी सजिलै घुम्न सक्छौं XREAD अवरुद्ध आदेशमा, BLOCK तर्क निर्दिष्ट गर्दै:

> XREAD BLOCK 0 STREAMS mystream $
माथिको उदाहरणमा, नयाँ ब्लक विकल्प ० मिलिसेकेन्डको टाइमआउटसँग निर्दिष्ट गरिएको छ (यसको अर्थ अनिश्चित कालसम्म पर्खनु हो)। यसबाहेक, स्ट्रिम mystream को लागि सामान्य पहिचानकर्ता पास गर्नुको सट्टा, एक विशेष पहिचानकर्ता $ पारित गरियो। यो विशेष पहिचानकर्ता भनेको हो XREAD mystream मा पहिचानकर्ताको रूपमा अधिकतम पहिचानकर्ता प्रयोग गर्नुपर्छ। त्यसैले हामीले सुन्न थालेको क्षणबाट मात्र नयाँ सन्देशहरू प्राप्त गर्नेछौं। केहि तरिकामा यो युनिक्स "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
यो प्रभाव प्राप्त गर्न, 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 - यो आदेशले ग्राहकलाई सन्देशलाई सफलतापूर्वक प्रशोधन गरिएको रूपमा चिन्ह लगाउन अनुमति दिन्छ

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

मानौं कि mystream पहिले नै अवस्थित छ। त्यसपछि समूह निर्माण आदेश यस्तो देखिन्छ:

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

अब जब समूह सिर्जना गरिएको छ, हामी तुरुन्तै आदेश प्रयोग गरेर सन्देशहरू पढ्न सुरु गर्न सक्छौं 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"

माथिको आदेश निम्नानुसार शब्दशः पढ्छ:

"म, सब्सक्राइबर एलिस, माइग्रुपको सदस्य, मेरो स्ट्रिमबाट एउटा सन्देश पढ्न चाहन्छु जुन पहिले कसैलाई पुर्‍याइएको थिएन।"

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

साथै, विशेष अवस्थामा, तपाईले वास्तविक पहिचानकर्ता निर्दिष्ट गर्न सक्नुहुन्छ जस्तै ० वा कुनै अन्य मान्य पहिचानकर्ता। यस मामला मा आदेश 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

अब हामीसँग प्रत्येक सन्देशको विवरणहरू छन्: ID, सदस्यको नाम, मिलिसेकेन्डमा निष्क्रिय समय र अन्तमा डेलिभरी प्रयासहरूको संख्या। हामीसँग बबबाट दुईवटा सन्देशहरू छन् र तिनीहरू 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 विकल्प सन्देश ID हरू मात्र फर्काउन प्रयोग गर्न सकिन्छ। यदि तपाइँ सन्देशको विवरणहरूमा रुचि राख्नुहुन्न र प्रणाली प्रदर्शन बढाउन चाहनुहुन्छ भने यो उपयोगी छ।

डेलिभरी काउन्टर

तपाईंले आउटपुटमा देख्नुभएको काउन्टर 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 प्रयोग गर्दा, पुराना रेकर्डहरू निर्दिष्ट लम्बाइमा पुग्दा स्वचालित रूपमा मेटाइन्छ, त्यसैले स्ट्रिमको आकार स्थिर हुन्छ। यद्यपि, यस अवस्थामा छाँट्ने काम Redis मेमोरीमा सबैभन्दा प्रभावकारी तरिकामा हुँदैन। तपाईं निम्नानुसार स्थिति सुधार गर्न सक्नुहुन्छ:

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

छुट्टै टोली पनि छ XTRIM, जसले एउटै कुरा गर्छ:

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

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

शून्य लम्बाइ स्ट्रिमहरू

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

निष्कर्षमा

Redis Stream सन्देश ब्रोकरहरू, सन्देश लाइनहरू, एकीकृत लगिङ, र इतिहास राख्ने च्याट प्रणालीहरू सिर्जना गर्नका लागि आदर्श हो।

मैले एकपटक भनेझैं Niklaus Wirth, प्रोग्रामहरू एल्गोरिदमहरू र डेटा संरचनाहरू हुन्, र Redis ले तपाईंलाई दुबै दिन्छ।

स्रोत: www.habr.com

Redis Stream - तपाईंको सन्देश प्रणालीको विश्वसनीयता र स्केलेबिलिटी