Redis Stream — nowy abstrakcyjny typ danych, wprowadzony w Redis w wersji 5.0
Z koncepcyjnego punktu widzenia, Redis Stream to Lista, do której można dodawać wpisy. Każdy wpis ma unikalny identyfikator. Domyślnie identyfikator jest generowany automatycznie i zawiera znacznik czasu. Dzięki temu można żądać zakresów wpisów w czasie lub odbierać nowe dane, gdy pojawiają się w strumieniu, podobnie jak polecenie Unix „tail -f” czyta plik dziennika i zatrzymuje się w oczekiwaniu na nowe dane. Należy zauważyć, że do strumienia mogą jednocześnie nasłuchiwać wielu klientów, tak jak wiele procesów „tail -f” może jednocześnie czytać plik, nie wchodząc ze sobą w konflikt.

Aby zrozumieć wszystkie zalety nowego typu danych, przypomnijmy sobie krótko istniejące od dawna struktury Redis, które częściowo powtarzają funkcjonalność Redis Stream.

Redis PUB/SUB

Redis Pub/Sub to prosta system komunikacji, już wbudowana w twoje magazynowanie key-value. Jednak za prostotę trzeba płacić:

Jeśli wydawca z jakiegoś powodu ulegnie awarii, traci wszystkich swoich subskrybentów.
Wydawca musi znać dokładny adres wszystkich swoich subskrybentów.
Wydawca może przeciążyć swoich subskrybentów, jeśli dane publikowane są szybciej niż są przetwarzane.
Wiadomość jest usuwana z bufora wydawcy natychmiast po publikacji, niezależnie od tego, ile subskrybentów ją otrzymało i jak szybko się z nią zapoznało.
Wszyscy subskrybenci otrzymują wiadomość jednocześnie. Subskrybenci muszą jakoś uzgodnić pomiędzy sobą kolejność przetwarzania tej samej wiadomości.
Nie ma wbudowanego mechanizmu potwierdzania pomyślnego przetworzenia wiadomości przez subskrybenta. Jeśli subskrybent otrzymał wiadomość i uległ awarii podczas przetwarzania, wydawca o tym nie dowie się.

Redis List

Redis List to struktura danych, która obsługuje komendy odczytu z blokowaniem. Możesz dodawać i odczytywać wiadomości z początku lub końca listy. Na podstawie tej struktury można stworzyć przyzwoity stos lub kolejkę dla twojego systemu rozproszonego i w większości przypadków to będzie wystarczające. Główne różnice w stosunku do Redis Pub/Sub to:

Wiadomość jest dostarczana jednemu klientowi. Pierwszy zablokowany na odczycie klient otrzyma dane jako pierwszy.
Klient musi sam zainicjować operację odczytu każdej wiadomości. List nic nie wie o klientach.
Wiadomości są przechowywane do momentu, aż ktoś je odczyta lub usunie je jawnie. Jeśli skonfigurowałeś serwer Redis, aby zapisywał dane na dysku, niezawodność systemu znacznie wzrasta.

Wprowadzenie do Stream

Dodawanie wpisu do strumienia

Zespół XADD dodaje nowy wpis do strumienia. Wpis to nie tylko ciąg, składa się z jednej lub kilku par klucz-wartość. W ten sposób każdy wpis jest już usystematyzowany i przypomina strukturę pliku CSV.

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

W powyższym przykładzie dodajemy do strumienia o nazwie (kluczu) „mystream” dwa pola: „sensor-id” i „temperature” z wartościami „1234” i „19.8” odpowiednio. Jako drugi argument polecenie przyjmuje identyfikator, który będzie przypisany do wpisu — ten identyfikator jednoznacznie identyfikuje każdy wpis w strumieniu. Jednak w tym przypadku przekazaliśmy *, ponieważ chcemy, aby Redis wygenerował dla nas nowy identyfikator. Każdy nowy identyfikator będzie zwiększany. Z tego powodu każdy nowy wpis będzie miał większy identyfikator w odniesieniu do poprzednich wpisów.

Format identyfikatora

Identyfikator wpisu, zwracany przez polecenie XADD, składa się z dwóch części:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — czas Unix w milisekundach (czas serwera Redis). Jednak jeśli bieżący czas jest równy lub mniejszy niż czas poprzedniego wpisu, używana jest znacznik czasu poprzedniego wpisu. Dlatego, jeśli czas serwera wraca w przeszłość, nowy identyfikator nadal zachowa swoje właściwości zwiększania.

sequenceNumber jest używany dla wpisów utworzonych w tej samej milisekundzie. sequenceNumber zwiększy się o 1 w stosunku do poprzedniego wpisu. Ponieważ sequenceNumber ma rozmiar 64 bitów, w praktyce nie powinieneś napotkać ograniczenia co do liczby wpisów, które mogą być generowane w jednej milisekundzie.

Format takich identyfikatorów może na pierwszy rzut oka wydawać się dziwny. Niedowierzający czytelnik może zapytać, dlaczego czas jest częścią identyfikatora. Powód jest taki, że strumienie Redis obsługują zapytania zakresowe po identyfikatorach. Ponieważ identyfikator jest związany z czasem utworzenia rekordu, umożliwia to zapytania dotyczące zakresów czasowych. Rozważymy konkretny przykład, gdy przejdziemy do studiów nad komendą. XRANGE.

Jeśli z jakiegoś powodu użytkownik musi podać swój własny identyfikator, który na przykład jest związany z jakimś zewnętrznym systemem, możemy go przekazać do komendy. XADD zamiast znaku * jak pokazano poniżej:

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

Zauważ, że w tym przypadku musisz samodzielnie śledzić wzrost identyfikatora. W naszym przykładzie minimalny identyfikator wynosi „0-1”, więc komenda nie zaakceptuje kolejnego identyfikatora, który jest równy lub mniejszy niż „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

Liczba rekordów w strumieniu

Można uzyskać liczbę rekordów w strumieniu, używając po prostu komendy XLEN. Dla naszego przykładu ta komenda zwróci następującą wartość:

> XLEN somestream
(integer) 2

Zapytania zakresowe — XRANGE i XREVRANGE

Aby zażądać danych w zakresie, musimy podać dwa identyfikatory — początkowy i końcowy zakres. Zwracany zakres będzie obejmował wszystkie elementy, w tym granice. Istnieją również dwa specjalne identyfikatory „-” i „+”, oznaczające odpowiednio najmniejszy (pierwszy rekord) i największy (ostatni rekord) identyfikator w strumieniu. Poniższy przykład wyświetli wszystkie rekordy w strumieniu.

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

Każdy zwrócony rekord stanowi tablicę z dwóch elementów: identyfikator i lista par klucz-wartość. Już mówiliśmy, że identyfikatory rekordów mają związek z czasem. Dlatego możemy zapytanie zakresowe o konkretny przedział czasu. Jednak możemy wskazać w zapytaniu niepełny identyfikator, a jedynie czas Unix, pomijając część odnoszącą się do. sequenceNumberOpuszczona część identyfikatora jest automatycznie ustawiana na zero na początku zakresu i na maksymalną możliwą wartość na końcu zakresu. Poniżej znajduje się przykład, jak można zażądać zakresu równego dwóm milisekundom.

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

Mamy tylko jeden wpis w tym zakresie, jednak w rzeczywistych zbiorach danych zwracany wynik może być ogromny. Z tego powodu XRANGE obsługuje opcję COUNT. Podając liczbę, możemy po prostu uzyskać pierwsze N wpisów. Jeśli chcemy uzyskać następne N wpisów (paginacja), możemy użyć ostatnio uzyskanego identyfikatora, zwiększyć go sequenceNumber o jeden i zapytać ponownie. Przyjrzyjmy się temu w następnym przykładzie. Zaczynamy dodawać 10 elementów za pomocą XADD (załóżmy, że strumień mystream był już wypełniony 10 elementami). Aby rozpocząć iterację, otrzymując po 2 elementy na komendę, zaczynamy od pełnego zakresu, ale z COUNT równym 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"

Aby kontynuować iterację z następnymi dwoma elementami, musimy wybrać ostatnio uzyskany identyfikator, czyli 1519073279157-0, i dodać 1 do sequenceNumber.
wynikowego identyfikatora, w tym przypadku 1519073279157-1, który teraz można użyć jako nowego argumentu początku zakresu dla następnego wywołania 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"

I tak dalej. Ponieważ złożoność XRANGE wynosi O(log (N)) dla wyszukiwania, a następnie O(M) dla zwracania M elementów, każdy krok iteracji jest szybki. W ten sposób XRANGE można efektywnie iterować strumienie.

Zespół XREVRANGE jest ekwiwalentem XRANGE, ale zwraca elementy w odwrotnej kolejności:

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

Zauważ, że komenda XREVRANGE przyjmuje argumenty zakresu start i stop w odwrotnej kolejności.

Czytanie nowych wpisów za pomocą XREAD

Często pojawia się potrzeba subskrypcji strumienia i otrzymywania tylko nowych wiadomości. Ta koncepcja może wydawać się podobna do Redis Pub/Sub lub blokującej listy Redis, ale są istotne różnice w tym, jak korzystać z Redis Stream:

Każda nowa wiadomość jest domyślnie dostarczana do każdego subskrybenta. To zachowanie różni się od blokującej listy Redis, gdzie nową wiadomość przeczyta tylko jeden subskrybent.
Podczas gdy w Redis Pub/Sub wszystkie wiadomości są zapominane i nigdy nie są zapisywane, w Stream wszystkie wiadomości są przechowywane na czas nieokreślony (chyba że klient wyraźnie wezwie do usunięcia).
Redis Stream pozwala na wydzielenie dostępu do wiadomości w ramach jednego strumienia. Konkretna subskrybent może widzieć tylko swoją osobistą historię wiadomości.

Możesz subskrybować strumień i otrzymywać nowe wiadomości, używając polecenia XREAD. To jest trochę bardziej skomplikowane niż XRANGE, dlatego najpierw zaczniemy od prostszych przykładów.

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

W powyższym przykładzie podano formę nieblokującą. XREADZauważ, że opcja COUNT nie jest obowiązkowa. Właściwie jedyną obowiązkową opcją polecenia jest opcja STREAMS, która ustala listę strumieni wraz z odpowiadającym maksymalnym identyfikatorem. Napisaliśmy „STREAMS mystream 0” — chcemy otrzymać wszystkie wpisy z strumienia mystream o identyfikatorze większym niż „0-0”. Jak widać z przykładu, polecenie zwraca nazwę strumienia, ponieważ możemy subskrybować wiele strumieni jednocześnie. Moglibyśmy napisać na przykład „STREAMS mystream otherstream 0 0”. Zauważ, że po opcji STREAMS najpierw musimy podać nazwy wszystkich potrzebnych strumieni, a dopiero potem listę identyfikatorów.

W tej prostej formie polecenie nie robi nic szczególnego w porównaniu z XRANGE. Jednak interesujące jest to, że możemy łatwo przekształcić XREAD w polecenie blokujące, podając argument BLOCK:

> XREAD BLOCK 0 STREAMS mystream $
W powyższym przykładzie wskazano nową opcję BLOCK z czasem oczekiwania 0 milisekund (co oznacza nieskończone oczekiwanie). Co więcej, zamiast przekazywać zwykły identyfikator dla strumienia mystream, przekazano specjalny identyfikator $. Ten specjalny identyfikator oznacza, że XREAD powinien używać maksymalnego identyfikatora w strumieniu mystream. Tak więc będziemy otrzymywać tylko nowe wiadomości, zaczynając z momentu, gdy zaczęliśmy nasłuchiwać. W pewnym sensie przypomina to polecenie Unix „tail -f”.

Zwróć uwagę, że korzystając z opcji BLOCK, nie musimy używać specjalnego identyfikatora $. Możemy używać dowolnego istniejącego identyfikatora w strumieniu. Jeśli zespół będzie w stanie obsłużyć nasze zapytanie natychmiast, bez blokady, to to zrobi, w przeciwnym razie zostanie zablokowane.

Blokujący XREAD może również nasłuchiwać jednocześnie wielu strumieni, wystarczy podać ich nazwy. W takim przypadku zespół zwróci dane z pierwszego strumienia, do którego dotarły dane. Pierwszy subskrybent zablokowany dla danego strumienia, będzie otrzymywał dane jako pierwszy.

Grupy konsumentów

W niektórych zadaniach chcemy wydzielić dostęp subskrybentów do wiadomości w obrębie jednego strumienia. Przykładem, gdy może to być przydatne, jest kolejka wiadomości z robotnikami, którzy będą odbierać różne wiadomości strumienia, co pozwala na skalowanie przetwarzania wiadomości.

Jeśli wyobrazimy sobie, że mamy trzech subskrybentów C1, C2, C3 i strumień, który zawiera wiadomości 1, 2, 3, 4, 5, 6, 7, to obsługa wiadomości odbywać się będzie zgodnie z poniższą diagramem:

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
Aby uzyskać ten efekt, Redis Stream wykorzystuje koncepcję zwaną grupą konsumentów. Ta koncepcja jest podobna do pseudo-subskrybenta, który otrzymuje dane ze strumienia, ale jest w rzeczywistości obsługiwana przez kilku subskrybentów w obrębie grupy, zapewniając określone gwarancje:

Każda wiadomość jest dostarczana różnym subskrybentom w obrębie grupy.
W ramach grupy subskrybenci są identyfikowani po nazwie, która jest ciągiem z uwzględnieniem wielkości liter. Jeśli jakiś subskrybent tymczasowo wypadnie z grupy, może się do niej ponownie przywrócić pod swoim unikalnym identyfikatorem.
Każda grupa konsumentów przestrzega koncepcji „pierwsza nieprzeczytana wiadomość”. Kiedy subskrybent żąda nowych wiadomości, może otrzymać tylko te, które nigdy wcześniej nie były dostarczane żadnemu subskrybentowi w obrębie grupy.
Istnieje polecenie jawnego potwierdzenia pomyślnego przetworzenia wiadomości przez subskrybenta. Dopóki to polecenie nie zostanie wywołane, żądana wiadomość pozostanie w statusie „oczekująca”.
W obrębie grupy konsumentów każdy subskrybent może żądać historii wiadomości, które zostały mu dostarczone, ale jeszcze nie zostały przetworzone (w statusie „oczekująca”).

W pewnym sensie stan grupy można przedstawić następująco:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" z oczekującymi wiadomościami  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" z oczekującymi wiadomościami 
|       ... (i tak dalej)                             
+----------------------------------------+

Teraz nadszedł czas, aby zapoznać się z podstawowymi poleceniami dla grupy konsumentów, a mianowicie:

XGROUP służy do tworzenia, usuwania i zarządzania grupami
XREADGROUP służy do odczytywania strumienia przez grupę
XACK — to polecenie umożliwia subskrybentowi oznaczenie wiadomości jako pomyślnie przetworzonej

Tworzenie grupy konsumentów

Załóżmy, że strumień mystream już istnieje. W takim razie polecenie utworzenia grupy będzie miało następującą postać:

> XGROUP CREATE mystream mygroup $ OK
Podczas tworzenia grupy musimy przekazać identyfikator, od którego grupa zacznie otrzymywać wiadomości. Jeśli chcemy po prostu otrzymywać wszystkie nowe wiadomości, możemy użyć specjalnego identyfikatora $ (jak w naszym przykładzie powyżej). Jeśli zamiast specjalnego identyfikatora podamy 0, to grupa będzie miała dostęp do wszystkich wiadomości strumienia.

Teraz, gdy grupa została utworzona, możemy od razu zacząć czytać wiadomości za pomocą polecenia XREADGROUP. To polecenie jest bardzo podobne do XREAD i wspiera opcjonalną opcję BLOCK. Jednak istnieje obowiązkowa opcja GROUP, która zawsze musi być podana z dwoma argumentami: nazwa grupy i nazwa subskrybenta. Opcja COUNT jest również obsługiwana.

Zanim zaczniemy czytać strumień, umieśćmy tam kilka wiadomości:

> 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

A teraz spróbujmy odczytać ten strumień przez grupę:

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

Powyższe polecenie dosłownie brzmi:

«Ja, Alice-subskrybent, członek grupy mygroup, chcę przeczytać jedno nieprzekazane wcześniej wiadomość ze strumienia mystream.»

Za każdym razem, gdy subskrybent wykonuje operację z grupą, musi podać swoje imię, jednoznacznie go identyfikując w grupie. W powyższym poleceniu jest jeszcze jeden bardzo ważny detal — specjalny identyfikator „>”. Ten specjalny identyfikator filtruje wiadomości, pozostawiając tylko te, które nie zostały jeszcze dostarczone.

Również, w szczególnych przypadkach, możesz podać rzeczywisty identyfikator, taki jak 0 lub dowolny inny ważny identyfikator. W takim przypadku polecenie XREADGROUP zwróci historię wiadomości o statusie „pending”, które zostały dostarczone określonemu subskrybentowi (Alice), ale jeszcze nie zostały potwierdzone za pomocą polecenia XACK.

Możemy sprawdzić to zachowanie, od razu podając identyfikator 0, bez opcji LICZBA. Po prostu zobaczymy jedną oczekującą wiadomość, czyli wiadomość z jabłkiem:

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

Jednak gdy potwierdzimy wiadomość jako pomyślnie przetworzoną, nie będzie się już wyświetlać:

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

Teraz nadszedł czas, aby Bob coś przeczytał:

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

Bob, członek grupy mygroup, poprosił o nie więcej niż dwie wiadomości. Polecenie zgłasza tylko niedostarczone wiadomości z powodu specjalnego identyfikatora „>”. Jak widać, wiadomość „apple” się nie wyświetla, ponieważ została już dostarczona Alice, więc Bob otrzymuje „orange” i „strawberry”.

W ten sposób Alice, Bob i każdy inny subskrybent grupy mogą czytać różne wiadomości z tego samego strumienia. Mogą również przeglądać historię swoich nieprzetworzonych wiadomości lub oznaczać wiadomości jako przetworzone.

Jest kilka rzeczy, o których należy pamiętać:

Gdy subskrybent uznaje wiadomość za polecenie XREADGROUP, ta wiadomość przechodzi w stan „pending” i zostaje przypisana temu konkretnemu subskrybentowi. Inni subskrybenci grupy nie będą mogli przeczytać tej wiadomości.
Subskrybenci są automatycznie tworzeni przy pierwszym wzmiance, nie ma potrzeby ich wyraźnego tworzenia.
Dzięki XREADGROUP Możesz czytać wiadomości z różnych strumieni jednocześnie, ale aby to działało, musisz najpierw stworzyć grupy o tej samej nazwie dla każdego strumienia za pomocą XGROUP

Odzyskiwanie po awarii

Subskrybent może odzyskać się po awarii i ponownie przeczytać swoją listę wiadomości ze statusem „oczekujące”. Jednak w rzeczywistości subskrybenci mogą ostatecznie ponieść porażkę. Co się dzieje z zawieszonymi wiadomościami subskrybenta, jeśli nie udało mu się odzyskać po awarii?
Grupa konsumencka oferuje funkcję, która jest używana w takich przypadkach — gdy konieczna jest zmiana właściciela wiadomości.

Na początek należy wywołać polecenie XPENDING, które wyświetla wszystkie wiadomości grupy ze statusem „oczekujące”. W najprostszej formie polecenie wywoływane jest tylko z dwoma argumentami: nazwą strumienia i nazwą grupy:

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

Polecenie wyświetliło liczbę nieprzetworzonych wiadomości dla całej grupy i dla każdego subskrybenta. Mamy tylko Boba z dwiema nieprzetworzonymi wiadomościami, ponieważ jedyna wiadomość, o którą prosiła Ala, została potwierdzona za pomocą XACK.

Możemy zażądać dodatkowych informacji, używając większej liczby argumentów:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — zakres identyfikatorów (można użyć „-” i „+”)
{count} — liczba prób dostarczenia
{consumer-name} — nazwa grupy

> 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

Teraz mamy szczegóły dla każdej wiadomości: identyfikator, nazwę subskrybenta, czas oczekiwania w milisekundach oraz, w końcu, liczbę prób dostarczenia. Mamy dwie wiadomości od Boba, które oczekują 74170458 milisekund, czyli około 20 godzin.

Zauważ, że nic nie stoi na przeszkodzie, aby sprawdzić, jaka była treść wiadomości, po prostu używając XRANGE.

> XRANGE mystream 1526569498055-0 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "wiadomość"
      2) "pomarańcza"

Musimy po prostu powtórzyć ten sam identyfikator dwukrotnie w argumentach. Teraz, gdy mamy pewien pomysł, Ala może zdecydować, że po 20 godzinach oczekiwania Bob prawdopodobnie się nie odzyska, więc nadszedł czas, aby zażądać tych wiadomości i wznowić ich przetwarzanie zamiast Boba. W tym celu używamy polecenia XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
Za pomocą tej komendy możemy uzyskać «cudzą» wiadomość, która jeszcze nie została przetworzona, zmieniając właściciela na {consumer}. Możemy jednak również ustawić minimalny czas bezczynności {min-idle-time}. Pomaga to uniknąć sytuacji, w której dwóch klientów jednocześnie próbuje zmienić właściciela tych samych wiadomości:

Klient 1: XCLAIM mystream mygroup Alicja 3600000 1526569498055-0 Klient 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
Pierwszy klient zresetuje czas bezczynności i zwiększy licznik dostaw. Zatem drugi klient nie będzie mógł go zażądać.

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "wiadomość"
      2) "pomarańczowy"

Wiadomość została pomyślnie odebrana przez Alicję, która teraz może ją przetworzyć i potwierdzić.

Z powyższego przykładu widać, że pomyślne wykonanie żądania zwraca treść samej wiadomości. Jednak nie jest to konieczne. Opcja JUSTID może być używana do zwracania tylko identyfikatorów wiadomości. Jest to przydatne, jeśli nie interesują nas szczegóły wiadomości i chcemy zwiększyć wydajność systemu.

Licznik dostaw

Licznik, który obserwujesz w wyniku XPENDING — to liczba dostaw każdej wiadomości. Taki licznik zwiększa się na dwa sposoby: gdy wiadomość jest pomyślnie żądana przez XCLAIM lub gdy używane jest wywołanie XREADGROUP.

To normalne, że niektóre wiadomości są dostarczane kilka razy. Najważniejsze, aby wszystkie wiadomości zostały przetworzone. Czasami podczas przetwarzania wiadomości występują problemy z powodu uszkodzenia samej wiadomości lub przetwarzanie wiadomości powoduje błąd w kodzie przetwarzającym. W takim przypadku może się okazać, że nikt nie będzie w stanie przetworzyć tej wiadomości. Ponieważ mamy licznik prób dostawy, możemy skorzystać z tego licznika, aby wykrywać takie sytuacje. Dlatego, gdy tylko licznik dostaw osiągnie dla ciebie ustaloną dużą liczbę, prawdopodobnie rozsądniej będzie przenieść tę wiadomość do innego strumienia i powiadomić administratora systemu.

Stan strumieni

Zespół XINFO jest używane do żądania różnych informacji o strumieniu i jego grupach. Na przykład podstawowy format komendy wygląda następująco:

> 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) "wiadomość"
       2) "banan"

Powyższe polecenie wyświetla podsumowanie informacji o wskazanym strumieniu. Teraz nieco bardziej złożony przykład:

> 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

Powyższe polecenie wyświetla podsumowanie informacji o wszystkich grupach wskazanego strumienia.

> 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

Powyższe polecenie wyświetla informacje o wszystkich subskrybentach wskazanego strumienia i grupy.
Jeśli zapomnisz składni polecenia, po prostu skorzystaj z pomocy samego polecenia:

> XINFO HELP
1) XINFO {subcommand} arg arg ... arg. Podpolecenia to:
2) CONSUMERS {key} {groupname}  -- Wyświetl grupy konsumentów grupy {groupname}.
3) GROUPS {key}                 -- Wyświetl grupy konsumentów strumienia.
4) STREAM {key}                 -- Wyświetl informacje o strumieniu.
5) HELP                         -- Wydrukuj tę pomoc.

Ograniczenie rozmiaru strumienia

Wiele aplikacji nie chce gromadzić danych w strumieniu bez końca. Często warto mieć maksymalną dopuszczalną liczbę wiadomości w strumieniu. W innych przypadkach korzystne jest przeniesienie wszystkich wiadomości ze strumienia do innego trwałego magazynu po osiągnięciu zadanej wielkości strumienia. Można ograniczyć rozmiar strumienia za pomocą parametru MAXLEN w poleceniu: 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"

Gdy używasz MAXLEN, stare wpisy są automatycznie usuwane po osiągnięciu określonej długości, dzięki czemu strumień ma stały rozmiar. Jednak przycinanie w tym przypadku odbywa się w sposób nie najwydajniejszy w pamięci Redis. Można poprawić tę sytuację w następujący sposób:

XADD mystream MAXLEN ~ 1000 * ... pola wpisy tutaj ...
Argument ~ w powyższym przykładzie oznacza, że nie musimy ograniczać długości strumienia do konkretnej wartości. W naszym przykładzie może to być każda liczba większa lub równa 1000 (np. 1000, 1010 lub 1030). Po prostu wyraźnie zaznaczyliśmy, że chcemy, aby nasz strumień przechowywał nie mniej niż 1000 wpisów. To znacznie zwiększa efektywność zarządzania pamięcią wewnętrznie w Redis.

Istnieje również osobne polecenie XTRIM, które wykonuje to samo:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Trwałe przechowywanie i replikacja

Redis Stream asynchronicznie replikuje się na węzły podrzędne i jest zapisywany w plikach typu AOF (migawek wszystkich danych) i RDB (log wszystkich operacji zapisu). Obsługiwana jest również replikacja stanu grup konsumentów. Dlatego jeśli wiadomość ma status „oczekująca” na głównym węźle, to na węzłach podrzędnych ta wiadomość będzie miała taki sam status.

Usuwanie poszczególnych elementów z strumienia

Aby usunąć wiadomości, istnieje specjalna komenda XDEL. Komenda otrzymuje nazwę strumienia, po której następują identyfikatory wiadomości, które należy usunąć:

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

Korzystając z tej komendy, należy wziąć pod uwagę, że pamięć faktycznie nie zostanie zwolniona od razu.

Strumienie o zerowej długości

Różnica między strumieniami a innymi strukturami danych Redis polega na tym, że gdy inne struktury danych nie mają już wewnętrznych elementów, w efekcie ubocznym sama struktura danych zostaje usunięta z pamięci. Na przykład, posortowany zbiór zostanie całkowicie usunięty, gdy wywołanie ZREM usunie ostatni element. Zamiast tego strumieniom pozwala się pozostać w pamięci, nawet gdy nie mają żadnego elementu wewnątrz.

Podsumowanie

Redis Stream idealnie nadaje się do tworzenia brokerów wiadomości, kolejek wiadomości, zunifikowanych dzienników i systemów czatów przechowujących historię.

Jak powiedział kiedyś Niklaus Wirth, programy to algorytmy plus struktury danych, a Redis dostarcza Ci zarówno jedno, jak i drugie.

Źródło: habr.com