Redis Stream არის ახალი აბსტრაქტული მონაცემთა ტიპი, რომელიც დაინერგა Redis-ში 5.0 ვერსიით
კონცეპტუალურად, Redis Stream არის სია, რომელშიც შეგიძლიათ დაამატოთ ჩანაწერები. თითოეულ ჩანაწერს აქვს უნიკალური იდენტიფიკატორი. ნაგულისხმევად, ID ავტომატურად გენერირდება და მოიცავს დროის ნიშანს. ამიტომ, თქვენ შეგიძლიათ მოითხოვოთ ჩანაწერების დიაპაზონი დროთა განმავლობაში, ან მიიღოთ ახალი მონაცემები ნაკადში მისვლისას, ისევე როგორც Unix-ის "tail -f" ბრძანება კითხულობს ჟურნალის ფაილს და იყინება ახალი მონაცემების მოლოდინში. გაითვალისწინეთ, რომ რამდენიმე კლიენტს შეუძლია მოისმინოს თემა ერთდროულად, ისევე როგორც ბევრ "tail -f" პროცესს შეუძლია წაიკითხოს ფაილი ერთდროულად, ერთმანეთთან კონფლიქტის გარეშე.

მონაცემთა ახალი ტიპის ყველა უპირატესობის გასაგებად, მოდით გადავხედოთ დიდი ხნის განმავლობაში არსებულ Redis სტრუქტურებს, რომლებიც ნაწილობრივ იმეორებენ Redis Stream-ის ფუნქციონირებას.

Redis PUB/SUB

Redis Pub/Sub არის მარტივი შეტყობინებების სისტემა, რომელიც უკვე ჩაშენებულია თქვენს გასაღების ღირებულების მაღაზიაში. თუმცა, სიმარტივე ფასიანია:

• თუ გამომცემელი რაიმე მიზეზით მარცხდება, მაშინ ის კარგავს თავის ყველა აბონენტს
• გამომცემელმა უნდა იცოდეს მისი ყველა აბონენტის ზუსტი მისამართი
• გამომცემელმა შეიძლება გადატვირთოს თავისი აბონენტები სამუშაოთი, თუ მონაცემები უფრო სწრაფად გამოქვეყნდება, ვიდრე დამუშავებულია
• შეტყობინება წაიშლება გამომცემლის ბუფერიდან გამოქვეყნებისთანავე, მიუხედავად იმისა, თუ რამდენ აბონენტს მიეწოდა და რამდენად სწრაფად შეძლეს ამ შეტყობინების დამუშავება.
• ყველა აბონენტი მიიღებს შეტყობინებას ერთდროულად. თავად აბონენტებმა როგორმე უნდა შეთანხმდნენ ერთსა და იმავე შეტყობინების დამუშავების თანმიმდევრობაზე.
• არ არსებობს ჩაშენებული მექანიზმი, რომელიც ადასტურებს, რომ აბონენტმა წარმატებით დაამუშავა შეტყობინება. თუ აბონენტი მიიღებს შეტყობინებას და ავარია დამუშავების დროს, გამომცემელმა არ იცის ამის შესახებ.

რედის სია

Redis List არის მონაცემთა სტრუქტურა, რომელიც მხარს უჭერს წაკითხვის ბრძანებების დაბლოკვას. შეგიძლიათ დაამატოთ და წაიკითხოთ შეტყობინებები სიის დასაწყისიდან ან ბოლოდან. ამ სტრუქტურიდან გამომდინარე, შეგიძლიათ შექმნათ კარგი სტეკი ან რიგი თქვენი განაწილებული სისტემისთვის და უმეტეს შემთხვევაში ეს საკმარისი იქნება. ძირითადი განსხვავებები Redis Pub/Sub-ისგან:

• შეტყობინება მიეწოდება ერთ კლიენტს. პირველი წაკითხული დაბლოკილი კლიენტი მიიღებს მონაცემებს.
• კლინტმა თავად უნდა წამოიწყოს წაკითხვის ოპერაცია თითოეული შეტყობინებისთვის. სიამ არაფერი იცის კლიენტების შესახებ.
• შეტყობინებები ინახება მანამ, სანამ ვინმე არ წაიკითხავს მათ ან აშკარად არ წაშლის მათ. თუ თქვენ დააკონფიგურირებთ Redis სერვერს მონაცემების დისკზე გადატანაზე, მაშინ სისტემის საიმედოობა მკვეთრად იზრდება.

სტრიმის შესავალი

ჩანაწერის დამატება ნაკადში

გუნდი XADD ამატებს ახალ ჩანაწერს ნაკადში. ჩანაწერი არ არის მხოლოდ სტრიქონი, ის შედგება ერთი ან მეტი გასაღები-მნიშვნელობის წყვილისაგან. ამრიგად, თითოეული ჩანაწერი უკვე სტრუქტურირებულია და წააგავს CSV ფაილის სტრუქტურას.

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

ზემოთ მოყვანილ მაგალითში ჩვენ ვამატებთ ნაკადს ორ ველს სახელწოდებით (გასაღები) „mystream“: „sensor-id“ და „temperature“ მნიშვნელობებით შესაბამისად „1234“ და „19.8“. როგორც მეორე არგუმენტი, ბრძანება იღებს იდენტიფიკატორს, რომელიც მიენიჭება ჩანაწერს - ეს იდენტიფიკატორი ცალსახად განსაზღვრავს თითოეულ ჩანაწერს ნაკადში. თუმცა, ამ შემთხვევაში ჩვენ გავიარეთ * იმიტომ, რომ გვსურს Redis-მა ახალი ID-ის გენერირება. ყოველი ახალი ID გაიზრდება. ამიტომ, ყოველ ახალ ჩანაწერს ექნება უფრო მაღალი იდენტიფიკატორი წინა ჩანაწერებთან მიმართებაში.

იდენტიფიკატორის ფორმატი

ბრძანებით დაბრუნებული შესვლის ID XADD, შედგება ორი ნაწილისაგან:

{millisecondsTime}-{sequenceNumber}

მილიწამის დრო — Unix დრო მილიწამებში (Redis სერვერის დრო). თუმცა, თუ მიმდინარე დრო იგივეა ან ნაკლებია, ვიდრე წინა ჩანაწერის დრო, მაშინ გამოიყენება წინა ჩანაწერის დრო. ამიტომ, თუ სერვერის დრო დროში ბრუნდება, ახალი იდენტიფიკატორი კვლავ შეინარჩუნებს ნამატი თვისებას.

რიგითი ნომერი გამოიყენება იმავე მილიწამში შექმნილი ჩანაწერებისთვის. რიგითი ნომერი გაიზრდება 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"

თითოეული დაბრუნებული ჩანაწერი არის ორი ელემენტის მასივი: იდენტიფიკატორი და გასაღები-მნიშვნელობის წყვილების სია. ჩვენ უკვე ვთქვით, რომ ჩანაწერების იდენტიფიკატორები დაკავშირებულია დროსთან. ამიტომ, ჩვენ შეგვიძლია მოვითხოვოთ გარკვეული პერიოდის დიაპაზონი. თუმცა, ჩვენ შეგვიძლია მოთხოვნაში მივუთითოთ არა სრული იდენტიფიკატორი, არამედ მხოლოდ Unix დრო, გამოვტოვოთ დაკავშირებული ნაწილი რიგითი ნომერი. იდენტიფიკატორის გამოტოვებული ნაწილი ავტომატურად დაყენდება ნულამდე დიაპაზონის დასაწყისში და მაქსიმალურ შესაძლო მნიშვნელობამდე დიაპაზონის ბოლოს. ქვემოთ მოცემულია მაგალითი იმისა, თუ როგორ შეგიძლიათ მოითხოვოთ ორი მილიწამის დიაპაზონი.

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

ჩვენ გვაქვს მხოლოდ ერთი ჩანაწერი ამ დიაპაზონში, თუმცა რეალურ მონაცემთა ნაკრებებში დაბრუნებული შედეგი შეიძლება იყოს უზარმაზარი. Ამ მიზეზით XRANGE მხარს უჭერს COUNT ვარიანტს. რაოდენობის მითითებით, ჩვენ შეგვიძლია უბრალოდ მივიღოთ პირველი N ჩანაწერები. თუ დაგვჭირდება შემდეგი N ჩანაწერების მიღება (პაგირება), შეგვიძლია გამოვიყენოთ ბოლო მიღებული ID, გავზარდოთ რიგითი ნომერი ერთი და ისევ იკითხე. მოდით შევხედოთ ამას შემდეგ მაგალითში. ჩვენ ვიწყებთ 10 ელემენტის დამატებას XADD (ვივარაუდოთ, რომ mystream უკვე სავსე იყო 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 List-ის დაბლოკვა, მაგრამ არსებობს ფუნდამენტური განსხვავებები Redis Stream-ის გამოყენებისას:

1. ყოველი ახალი შეტყობინება სტანდარტულად მიეწოდება ყველა აბონენტს. ეს ქცევა განსხვავდება Redis List-ის დაბლოკვისგან, სადაც ახალ შეტყობინებას წაიკითხავს მხოლოდ ერთი აბონენტი.
2. მიუხედავად იმისა, რომ Redis Pub/Sub-ში ყველა შეტყობინება დავიწყებულია და არასდროს რჩება, Stream-ში ყველა შეტყობინება ინახება განუსაზღვრელი ვადით (თუ კლიენტი აშკარად არ იწვევს წაშლას).
3. 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“ - გვინდა მივიღოთ mystream ნაკადის ყველა ჩანაწერი „0-0“-ზე მეტი იდენტიფიკატორით. როგორც მაგალითზე ხედავთ, ბრძანება აბრუნებს თემის სახელს, რადგან ჩვენ შეგვიძლია გამოვწეროთ რამდენიმე თემა ერთდროულად. ჩვენ შეგვიძლია დავწეროთ, მაგალითად, "STREAMS mystream otherstream 0 0". გთხოვთ გაითვალისწინოთ, რომ STREAMS ოფციის შემდეგ ჯერ უნდა მოგვაწოდოთ ყველა საჭირო ნაკადის სახელები და მხოლოდ ამის შემდეგ იდენტიფიკატორების სია.

ამ მარტივ ფორმაში ბრძანება არ აკეთებს რაიმე განსაკუთრებულს XRANGE. თუმცა, საინტერესო ის არის, რომ ჩვენ მარტივად შეგვიძლია შემობრუნება XREAD დაბლოკვის ბრძანებას, სადაც მითითებულია BLOCK არგუმენტი:

> XREAD BLOCK 0 STREAMS mystream $

ზემოთ მოყვანილ მაგალითში, ახალი BLOCK ოფცია მითითებულია 0 მილიწამის დროით (ეს ნიშნავს განუსაზღვრელი ვადით ლოდინს). უფრო მეტიც, ნაკადის mystream-ისთვის ჩვეულებრივი იდენტიფიკატორის გადაცემის ნაცვლად, გადაეცა სპეციალური იდენტიფიკატორი $. ეს სპეციალური იდენტიფიკატორი ნიშნავს იმას XREAD იდენტიფიკატორად უნდა გამოიყენოს მაქსიმალური იდენტიფიკატორი mystream-ში. ასე რომ, ჩვენ მივიღებთ მხოლოდ ახალ შეტყობინებებს მოსმენის დაწყების მომენტიდან. გარკვეულწილად, ეს ჰგავს Unix-ის "კუდი -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. ეს კონცეფცია ჰგავს ფსევდო აბონენტს, რომელიც იღებს მონაცემებს ნაკადიდან, მაგრამ რეალურად მას ემსახურება რამდენიმე აბონენტი ჯგუფის შიგნით, რაც უზრუნველყოფს გარკვეულ გარანტიებს:

1. თითოეული შეტყობინება მიეწოდება სხვადასხვა აბონენტს ჯგუფის შიგნით.
2. ჯგუფში, აბონენტები იდენტიფიცირებულია მათი სახელით, რომელიც არის რეგისტრის მგრძნობიარე სტრიქონი. თუ აბონენტი დროებით გამოდის ჯგუფიდან, ის შეიძლება აღდგეს ჯგუფში საკუთარი უნიკალური სახელის გამოყენებით.
3. ყველა მომხმარებელთა ჯგუფი მიჰყვება „პირველი წაუკითხავი შეტყობინების“ კონცეფციას. როდესაც აბონენტი ითხოვს ახალ შეტყობინებებს, მას შეუძლია მიიღოს მხოლოდ შეტყობინებები, რომლებიც ადრე არასოდეს მიუწოდებიათ რომელიმე აბონენტს ჯგუფში.
4. არსებობს ბრძანება ცალსახად დაადასტუროს, რომ შეტყობინება წარმატებით დამუშავდა აბონენტის მიერ. ამ ბრძანების გამოძახებამდე, მოთხოვნილი შეტყობინება დარჩება "მოლოდინის" სტატუსში.
5. მომხმარებელთა ჯგუფის ფარგლებში, თითოეულ აბონენტს შეუძლია მოითხოვოს მისთვის მიწოდებული, მაგრამ ჯერ არ დამუშავებული შეტყობინებების ისტორია („მოლოდინში“ სტატუსში)

გარკვეული გაგებით, ჯგუფის მდგომარეობა შეიძლება გამოიხატოს შემდეგნაირად:

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

ჯგუფის შექმნისას უნდა გადავცეთ იდენტიფიკატორი, საიდანაც ჯგუფი მიიღებს შეტყობინებებს. თუ ჩვენ უბრალოდ გვინდა ყველა ახალი შეტყობინების მიღება, მაშინ შეგვიძლია გამოვიყენოთ სპეციალური იდენტიფიკატორი $ (როგორც ჩვენს მაგალითში ზემოთ). თუ სპეციალური იდენტიფიკატორის ნაცვლად მიუთითებთ 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"

ზემოთ მოყვანილი ბრძანება სიტყვასიტყვით იკითხება შემდეგნაირად:

"მე, აბონენტ ალისს, mygroup-ის წევრს, მინდა წავიკითხო ერთი შეტყობინება mystream-დან, რომელიც აქამდე არავის მიუწოდებია."

ყოველ ჯერზე, როდესაც აბონენტი ასრულებს ოპერაციას ჯგუფზე, მან უნდა მიუთითოს თავისი სახელი, ცალსახად განსაზღვროს საკუთარი თავი ჯგუფში. ზემოხსენებულ ბრძანებაში არის კიდევ ერთი ძალიან მნიშვნელოვანი დეტალი - სპეციალური იდენტიფიკატორი ">". ეს სპეციალური იდენტიფიკატორი ფილტრავს შეტყობინებებს და ტოვებს მხოლოდ მათ, რომლებიც აქამდე არასოდეს ყოფილა მიწოდებული.

ასევე, განსაკუთრებულ შემთხვევებში, შეგიძლიათ მიუთითოთ რეალური იდენტიფიკატორი, როგორიცაა 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"

ბობმა, mygroup-ის წევრმა, მოითხოვა არაუმეტეს ორი შეტყობინებისა. ბრძანება იტყობინება მხოლოდ მიუწოდებელ შეტყობინებებს სპეციალური იდენტიფიკატორის ">" გამო. როგორც ხედავთ, შეტყობინება "ვაშლი" არ გამოჩნდება, რადგან ის უკვე მიწოდებულია ალისზე, ამიტომ ბობი იღებს "ფორთოხლის" და "მარწყვის".

ამ გზით, ალისს, ბობს და ჯგუფის ნებისმიერ სხვა აბონენტს შეუძლია წაიკითხოს სხვადასხვა შეტყობინებები იმავე ნაკადიდან. მათ ასევე შეუძლიათ წაიკითხონ დაუმუშავებელი შეტყობინებების ისტორია ან მონიშნონ შეტყობინებები დამუშავებულად.

გასათვალისწინებელია რამდენიმე რამ:

• როგორც კი აბონენტი შეტყობინებას ბრძანებად ჩათვლის XREADGROUP, ეს შეტყობინება გადადის „მოლოდინში“ და მიეკუთვნება კონკრეტულ აბონენტს. ჯგუფის სხვა აბონენტები ამ შეტყობინების წაკითხვას ვერ შეძლებენ.
• აბონენტები ავტომატურად იქმნება პირველი მოხსენიებისთანავე, არ არის საჭირო მათი ცალსახად შექმნა.
• ერთად XREADGROUP თქვენ შეგიძლიათ წაიკითხოთ შეტყობინებები რამდენიმე სხვადასხვა თემიდან ერთდროულად, მაგრამ ამისთვის, პირველ რიგში, თქვენ უნდა შექმნათ ჯგუფები ერთი და იგივე სახელით თითოეული თემისთვის. XGROUP

აღდგენა წარუმატებლობის შემდეგ

აბონენტს შეუძლია გამოჯანმრთელდეს წარუმატებლობისგან და ხელახლა წაიკითხოს შეტყობინებების სია „მოლოდინში“ სტატუსით. თუმცა, რეალურ სამყაროში, აბონენტებმა შეიძლება საბოლოოდ ვერ შეძლონ. რა ბედი ეწევა აბონენტის ჩარჩენილ შეტყობინებებს, თუ აბონენტი ვერ გამოჯანმრთელდება წარუმატებლობისგან?
მომხმარებელთა ჯგუფი გთავაზობთ ფუნქციას, რომელიც გამოიყენება სწორედ ასეთი შემთხვევებისთვის - როცა გჭირდებათ შეტყობინებების მფლობელის შეცვლა.

პირველი რაც უნდა გააკეთოთ არის ბრძანების გამოძახება მოსალოდნელია, რომელიც აჩვენებს ჯგუფში ყველა შეტყობინებას სტატუსით „მოლოდინში“. უმარტივესი ფორმით, ბრძანება გამოძახებულია მხოლოდ ორი არგუმენტით: ძაფის სახელი და ჯგუფის სახელი:

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

ამ ბრძანების გამოყენებით, ჩვენ შეგვიძლია მივიღოთ „უცხო“ შეტყობინება, რომელიც ჯერ არ არის დამუშავებული მფლობელის {consumer}-ზე გადაცვლით. თუმცა, ჩვენ შეგვიძლია ასევე მოგაწოდოთ უმოქმედობის მინიმალური დრო {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-ების დასაბრუნებლად. ეს სასარგებლოა იმ შემთხვევაში, თუ არ გაინტერესებთ შეტყობინების დეტალები და გსურთ სისტემის მუშაობის ამაღლება.

მიტანის მრიცხველი

მრიცხველი, რომელსაც ხედავთ გამომავალში მოსალოდნელია არის თითოეული შეტყობინების მიწოდების რაოდენობა. ასეთი მრიცხველი იზრდება ორი გზით: როდესაც შეტყობინება წარმატებით მოითხოვება მეშვეობით 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. ბრძანება იღებს თემის სახელს, რასაც მოჰყვება წასაშლელი შეტყობინების ID:

> 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 იდეალურია შეტყობინებების ბროკერების, შეტყობინებების რიგების, ერთიანი აღრიცხვისა და ისტორიის შესანახი ჩატის სისტემების შესაქმნელად.

როგორც ერთხელ ვთქვი ნიკლაუს ვირტი, პროგრამები არის ალგორითმები პლუს მონაცემთა სტრუქტურები და Redis უკვე გაძლევთ ორივეს.

წყარო: www.habr.com