Strumienie Redis, konsumenci i blokowanie operacji

Redis wprowadził strumienie, które naśladują strukturę danych dziennika z wersją 5.0. Stream to struktura danych tylko w dodatku z bogatszym zestawem operacji niż w pliku dziennika. Jest to jeden z najbardziej złożonych rodzajów danych w Redis, ponieważ implementuje dodatkowe operacje blokujące, które pozwalają klientom czekać na nowe dane dotyczące strumienia. Jest to nieco podobne do zachowania Redis Pub/Sub lub list blokujących, ale istnieją podstawowe różnice, jeśli chodzi o sposób, w jaki konsumenci konsumują dane Stream Redis.

Jak pokazano na poprzedniej ilustracji, widać kilka zalet w stosunku do Redis Pub/Sub i listach blokujących. Każdy nowy element danych jest dostarczany do każdego konsumenta. W przeciwieństwie do listy usuwania elementu listy za każdym razem, gdy nazywano BLPOP lub BRPOP, elementy strumienia pozostają tak, jak w strumieniu. Komenda XREAD działa jako kandydat blokujący i nie blokujący w strumieniach Redis.

Polecenie XREAD

Polecenie XREAD może pobierać wpisy z wielu strumieni jednocześnie, podczas gdy zwrócone wpisy mają identyfikator większy niż ostatni otrzymany identyfikator dla danego konsumenta. Może działać zarówno w sposób blokujący, jak i niekaktyczny. W naturze bez blokującej polecenie zachowuje się bardzo podobnie do polecenia Xrange, ale z dodatkowymi funkcjami wymienionymi w następujący sposób:

• Może pobierać wpisy zaczynające się od najnowszego wpisu, który ma największy identyfikator niż jakikolwiek inny element w strumieniu.
• Może odczytać z wielu strumieni jednocześnie.

To polecenie ma liniową złożoność czasu, gdy N liczba elementów jest przechowywana w strumieniu. Stąd, ze stałą liczbą powrotu, złożoność czasu jest stała.

Polecenie XREAD jest zgodne z następującą składnią:

Składnia:

Xread [liczba numer_of_returned_elements] [blok blokujący_time_in_milliseConds] Streams Key [klucz…] Id [id…]

LICZYĆ : Liczba elementów, które należy zwrócić przez polecenie. Ogranicza zwrócone wiersze do określonej liczby.

BLOK : Maksymalny czas na czekanie, aż pojawi się nowy element w strumieniu.

Dwie poprzednie opcje są opcjonalne do polecenia.

Strumienie: Klucz strumienia. Jest to opcja obowiązkowa i musi być ostatnią opcją w poleceniu, ponieważ akceptuje zmienną długość klawiszy i identyfikatory wejścia.

: Identyfikator wpisu strumieniowego.

Można określić wiele kluczy, ponieważ polecenie pozwala odczytać z więcej niż jednego strumienia. Jednocześnie można dostarczyć wiele identyfikatorów.

To polecenie zwraca odpowiedź na tablicę. Każdy element tablicy składa się z dwóch elementów, jak pokazano w następującym formacie:

Przykład 1: Sprawdź dane pogodowe dla dwóch lokalizacji z niekakującym XREAD

Załóżmy, że mamy dwa strumienie zawierające dane pogodowe dla LA i NYC. Na naszej stronie publikacji danych pogodowych musimy konsumować zarówno ze strumieni i pobierać najnowsze dane pogodowe dla tych dwóch lokalizacji. Polecenie XREAD jest idealnym kandydatem do użycia w tym scenariuszu z wariantem nie blokującym.

Czas utworzyć dwa wymienione strumienie Pogoda: NYC I Pogoda: LA i wypełnij kilka wpisów niektórymi parami wartości pola, jak pokazano następująco:

Xadd pogoda: NYC * Wind 45 wilgotność 78 Temp 12
Xadd pogoda: La * Wind 12 wilgotność 45 Temp 22

Oba strumienie Pogoda: NYC I Pogoda: LC są utworzone pomyślnie, a zwrócone identyfikatory wejścia są 1658114094434-0 I 1658114110474-0, odpowiednio.

Użyjmy polecenia XREAD, aby odczytać z obu strumieni w tym samym czasie w sposób bez blokowania.

xread Streams Pogoda: NYC Pogoda: LA 0 0

Zgodnie z oczekiwaniami wyjście zawiera wpisy z obu strumieni z sekwencją identyfikacyjną od 0. Dopuszczalne jest określenie niekompletnych identyfikatorów, jak wcześniej zilustrowano, gdzie oba identyfikatory są 0, czyli znacznik czasu milisekundowy bez części numeru sekwencji. Stąd poprzednie polecenie można napisać jak w następujący sposób:

xread Streams Pogoda: NYC Pogoda: LA 0-0 0-0

Dodajmy teraz kilka wpisów do obu strumieni.

Xadd pogoda: NYC * WITR 10 wilgotność 60 Temp 10
xadd pogoda: La * wiatr 18 wilgotność 80 Temp 5

Ponieważ mamy już najnowsze identyfikatory wpisowe dla obu strumieni z poprzednich poleceń, zadzwońmy ponownie do polecenia XREAD, aby pobrać wszystkie wpisy z większymi identyfikatorami niż te, które już zapytaliśmy.

xread Streams Pogoda: NYC Pogoda: LA 1658114094434-0 1658114110474-0

Jak widać, określone identyfikatory pochodzą z poprzedniego zapytania. Teraz wywołanie poleceń zwraca wszystkie wpisy, które mają większe identyfikatory niż określone.

Jak widać, nowo dodane wpisy są zwracane z poprzedniego polecenia. Następnie możesz wziąć identyfikatory zgłoszeń zwrócone z poprzedniego polecenia i wywołać xread z tymi identyfikatorami, aż zwrócona tablica będzie pusta.

Przykład 2: Zdobądź najnowsze promocje pizzy z blokowaniem XREAD

Istnieje inny wariant polecenia XREAD, którego można użyć do oczekiwania, aż wydawcy opublikują nowe dane w strumieniu bez zakończenia jako połączenia nieklokujące. Załóżmy scenariusz, w którym faceci z pizzy chcą przekazać powiadomienia wszystkich klientów o najnowszych dostępnych promocjach. W niektórych dniach nie może być promocji. Dlatego klienci powinni poczekać, aż nowe promocje będą dostępne. Można to osiągnąć za pomocą polecenia xread z opcją bloków.

Załóżmy, że firma pizza publikuje szczegóły promocji do strumienia o nazwie Pizzapromos: Codziennie. Dlatego możemy użyć polecenia XREAD, aby poczekać, aż nowy element promocyjny zostanie dodany do strumienia.

XREAD BLOCK 50000 Streams Pizzapromosnew: Daily $

W takim przypadku określamy identyfikator wpisu jako $, który jest interpretowany jako najlepszy identyfikator wpisu. Stąd polecenie zapyta tylko nowe wpisy dodane do strumienia, a nie historyczne wpisy.

Ponieważ nie dodaliśmy nowych wpisów do strumienia, będzie on czas po 50000 milisekund z zero powrót, jak pokazano następująco:

Teraz dodajmy wpis do strumienia za pomocą XADD, podczas gdy inny konsument czeka na dane z poleceniem XREAD, jak pokazano następująco:

Zgodnie z oczekiwaniami, dodany wpis jest natychmiast zużywany przez konsumenta. Z następnego połączenia musimy upewnić się, że przekazujemy identyfikator zwracany z tego polecenia, a nie $. Jeśli nie, przegapimy wpisy dodane pomiędzy.

Jeśli wielu klientów czeka na ten sam strumień, nowo dodane dane zostaną natychmiast przeniesione do wszystkich. Polecenie XREAD jest bardzo przydatnym i zalecanym poleceniem do użycia w blokowaniu aplikacji przyrodniczych.

Wniosek

Podsumowując, polecenie XREAD jest jednym z powszechnie używanych poleceń, które działają na strumieniach Redis. Może działać zarówno w sposób blokujący, jak i nie blokujący. Jak omówiono, wariant nie blokujący jest bardzo podobny do polecenia Xrange z kilkoma różnicami. Ponadto to polecenie można użyć z opcją blokowania, aby poczekać, aż wydawcy opublikują nowe dane w strumieniu. Ogólnie rzecz biorąc, polecenie XREAD specjalizuje się w konsumpcji danych z wielu strumieni jednocześnie. Jest to pomocna funkcja, której szukają współczesne aplikacje.