Redis Stream – ein neuer abstrakter Datentyp, der mit der Version 5.0 in Redis eingeführt wurde.
Konzeptionell ist Redis Stream eine Liste, in die Sie Einträge hinzufügen können. Jeder Eintrag hat eine eindeutige Kennung, die standardmäßig automatisch generiert wird und einen Zeitstempel enthält. Daher können Sie Zeiträume von Einträgen abfragen oder neue Daten in Echtzeit erhalten, während sie in den Stream gelangen, ähnlich wie der Unix-Befehl „tail -f“ eine Protokolldatei liest und auf neue Daten wartet. Beachten Sie, dass mehrere Clients gleichzeitig auf den Stream zugreifen können, ähnlich wie viele „tail -f“-Prozesse gleichzeitig eine Datei lesen können, ohne sich gegenseitig zu stören.

Um alle Vorteile des neuen Datentyps zu verstehen, lassen Sie uns kurz die bereits bestehenden Redis-Strukturen in Erinnerung rufen, die teilweise die Funktionalität von Redis Stream wiederholen.

Redis PUB/SUB

Redis Pub/Sub ist ein einfaches Nachrichtensystem, das bereits in Ihrem Key-Value-Speicher integriert ist. Für die Einfachheit müssen Sie jedoch einen Preis zahlen:

Wenn ein Publisher aus irgendeinem Grund ausfällt, verliert er alle seine Abonnenten.
Der Publisher muss die genaue Adresse aller seiner Abonnenten kennen.
Der Publisher kann seine Abonnenten überlasten, wenn die Daten schneller veröffentlicht werden, als sie verarbeitet werden.
Die Nachricht wird sofort nach der Veröffentlichung aus dem Buffer des Publishers entfernt, unabhängig davon, wie vielen Abonnenten sie zugestellt wurde und wie schnell diese die Nachricht verarbeiten konnten.
Alle Abonnenten erhalten die Nachricht gleichzeitig. Die Abonnenten müssen selbst einen Verarbeitungsreihenfolge für dieselbe Nachricht untereinander vereinbaren.
Es gibt keinen integrierten Mechanismus zur Bestätigung der erfolgreichen Verarbeitung einer Nachricht durch den Abonnenten. Wenn der Abonnent die Nachricht erhält und während der Verarbeitung ausfällt, wird der Publisher dies nicht erfahren.

Redis-Liste

Die Redis-Liste ist eine Datenstruktur, die Befehle zum blockierenden Lesen unterstützt. Sie können Nachrichten entweder am Anfang oder am Ende der Liste hinzufügen und abrufen. Auf dieser Grundlage kann man einen soliden Stapel oder eine Warteschlange für Ihr verteiltes System erstellen, und das wird in den meisten Fällen ausreichen. Die wichtigsten Unterschiede zum Redis Pub/Sub sind:

Die Nachricht wird an einen einzelnen Client zugestellt. Der erste blockierte lesende Client erhält die Daten zuerst.
Clint muss den Lesevorgang jeder Nachricht selbst initiieren. List weiß nichts über die Kunden.
Nachrichten werden so lange gespeichert, bis sie jemand abruft oder ausdrücklich löscht. Wenn Sie den Redis-Server so konfiguriert haben, dass er Daten auf die Festplatte schreibt, erhöht sich die Zuverlässigkeit des Systems erheblich.

Einführung in Streams

Eintrag zum Stream hinzufügen

Der Befehl XADD fügt einen neuen Eintrag zum Stream hinzu. Ein Eintrag ist nicht einfach nur eine Zeile, sondern besteht aus einem oder mehreren Schlüssel-Wert-Paaren. Somit ist jeder Eintrag bereits strukturiert und ähnelt dem Aufbau einer CSV-Datei.

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

Im obigen Beispiel fügen wir dem Stream mit dem Namen (Schlüssel) „mystream“ zwei Felder hinzu: „sensor-id“ und „temperature“ mit den Werten „1234“ und „19,8“ respektive. Als zweiten Parameter nimmt der Befehl eine ID entgegen, die dem Eintrag zugewiesen wird – diese ID identifiziert jeden Eintrag im Stream eindeutig. In diesem Fall haben wir jedoch * übergeben, da wir möchten, dass Redis für uns eine neue ID generiert. Jede neue ID wird inkrementiert. Daher wird jeder neue Eintrag eine größere ID im Vergleich zu den vorherigen Einträgen haben.

Format der ID

Die von dem Befehl zurückgegebene Eintrag-ID XADD, besteht aus zwei Teilen:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — Unix-Zeit in Millisekunden (Zeit Server Redis). Sollte die aktuelle Zeit gleich oder geringer als die Zeit des vorherigen Eintrags sein, wird der Zeitstempel des vorherigen Eintrags verwendet. Falls die Zeit des Servers in die Vergangenheit zurückspringt, wird die neue ID dennoch ihre Inkrementierungsattribute behalten.

sequenceNumber wird für Einträge verwendet, die in derselben Millisekunde erstellt wurden. sequenceNumber wird um 1 im Vergleich zum vorherigen Eintrag erhöht. Da sequenceNumber hat eine Größe von 64 Bit, sodass Sie in der Praxis nicht auf die Begrenzung der Anzahl von Datensätzen stoßen sollten, die innerhalb einer Millisekunde generiert werden können.

Auf den ersten Blick mag das Format solcher Bezeichner ungewöhnlich erscheinen. Ein skeptischer Leser könnte sich fragen, warum Zeit ein Teil des Identifikators ist. Der Grund dafür ist, dass Redis-Streams die Abfrage von Bereichen basierend auf Identifikatoren unterstützen. Da der Identifikator mit dem Erstellungszeitpunkt des Datensatzes verknüpft ist, ermöglicht es die Abfrage von Zeitbereichen. Ein konkretes Beispiel werden wir betrachten, wenn wir zur Untersuchung des Befehls XRANGE.

übergehen. Wenn der Benutzer aus irgendeinem Grund seinen eigenen Identifikator angeben muss, der beispielsweise mit einem externen System verknüpft ist, können wir ihn dem Befehl übergeben. XADD statt des Zeichens * wie unten gezeigt:

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

Bitte beachten Sie, dass Sie in diesem Fall selbst auf die Erhöhung des Identifikators achten müssen. In unserem Beispiel beträgt der minimale Identifikator „0-1“, weshalb der Befehl keinen weiteren Identifikator akzeptiert, der gleich oder kleiner als „0-1“ ist.

> XADD somestream 0-1 foo bar
(error) ERR Die angegebene ID in XADD ist gleich oder kleiner als das oberste Element des Zielstreams

Anzahl der Einträge im Stream

Die Anzahl der Einträge im Stream kann einfach mit dem Befehl ermittelt werden XLEN. In unserem Beispiel wird dieser Befehl den folgenden Wert zurückgeben:

> XLEN somestream
(ganzzahlig) 2

Bereichsanfragen — XRANGE und XREVRANGE

Um Daten nach einem Bereich abzufragen, müssen wir zwei Identifikatoren angeben — den Anfang und das Ende des Bereichs. Der zurückgegebene Bereich umfasst alle Elemente, einschließlich der Grenzen. Es gibt auch zwei spezielle Identifikatoren „-“ und „+“, die den kleinsten (d ersten Eintrag) und den größten (den letzten Eintrag) Identifikator im Stream darstellen. Das folgende Beispiel gibt alle Einträge im Stream aus.

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

Jeder zurückgegebene Eintrag ist ein Array aus zwei Elementen: einer ID und einer Liste von Schlüssel-Wert-Paaren. Wir haben bereits erwähnt, dass die IDs zeitbezogen sind. Daher können wir einen spezifischen Zeitraum abfragen. Wir können jedoch im Antrag nicht die vollständige ID angeben, sondern nur die Unix-Zeit, wobei wir den Teil weglassen, der sich auf sequenceNumberbezieht. Der weggelassene Teil der ID wird automatisch am Anfang des Bereichs auf null und am Ende auf den maximal möglichen Wert gesetzt. Hier ist ein Beispiel, wie man einen Bereich von zwei Millisekunden anfragen kann.

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

In diesem Bereich haben wir nur einen Eintrag, aber in realen Datensätzen kann das Ergebnis enorm sein. Aus diesem Grund XRANGE unterstützt die COUNT-Option. Wenn wir eine Anzahl angeben, können wir einfach die ersten N Einträge erhalten. Wenn wir die nächsten N Einträge benötigen (Seitenumbruch), können wir die zuletzt erhaltene ID verwenden und sie erhöhen. sequenceNumber für jede Einheit und erneut anfordern. Lassen Sie uns das anhand folgendes Beispiels betrachten. Wir beginnen, 10 Elemente hinzuzufügen mit Hilfe von XADD (angenommen, der Stream mystream wurde bereits mit 10 Elementen gefüllt). Um die Iteration zu starten und jeweils 2 Elemente pro Befehl zu erhalten, beginnen wir mit dem vollständigen Bereich, jedoch mit COUNT gleich 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"

Um die Iteration mit den nächsten zwei Elementen fortzusetzen, müssen wir die letzte erhaltene ID auswählen, das heißt 1519073279157-0, und 1 zu sequenceNumber.
der resultierenden ID hinzufügen. In diesem Fall 1519073279157-1, nun kann sie als neues Argument für den Beginn des Bereichs beim nächsten Aufruf verwendet werden. 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"

Und so weiter. Da die Komplexität XRANGE O(log (N)) für die Suche beträgt, und dann O(M) für die Rückgabe von M Elementen, ist jeder Schritt der Iteration schnell. So kann man XRANGE effizient durch Streams iterieren.

Der Befehl XREVRANGE ist das Äquivalent XRANGE, gibt jedoch die Elemente in umgekehrter Reihenfolge zurück:

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

Bitte beachten Sie, dass der Befehl XREVRANGE nimmt die Start- und Stoppwerte in umgekehrter Reihenfolge entgegen.

Neue Einträge mit XREAD lesen

Es gibt oft die Aufgabe, einen Stream zu abonnieren und nur neue Nachrichten zu erhalten. Dieses Konzept könnte ähnlich erscheinen wie Redis Pub/Sub oder eine blockierende Redis-Liste, aber es gibt wesentliche Unterschiede in der Nutzung von Redis Stream:

Jede neue Nachricht wird standardmäßig jedem Abonnenten zugestellt. Dieses Verhalten unterscheidet sich von einer blockierenden Redis-Liste, wo eine neue Nachricht nur von einem einzigen Abonnenten gelesen wird.
Während in Redis Pub/Sub alle Nachrichten vergessen werden und niemals gespeichert sind, werden in Streams alle Nachrichten auf unbestimmte Zeit gespeichert (es sei denn, der Client fordert ausdrücklich deren Löschung an).
Redis Stream ermöglicht es, den Zugriff auf Nachrichten innerhalb eines Streams zu differenzieren. Ein bestimmter Abonnent kann nur seine persönliche Nachrichtenhistorie sehen.

Sie können einen Stream abonnieren und neue Nachrichten empfangen, indem Sie den Befehl XREAD. Das ist etwas komplizierter als XRANGE, daher beginnen wir zunächst mit einfacheren Beispielen.

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

Im obigen Beispiel ist eine nicht blockierende Form dargestellt. XREADBitte beachten Sie, dass die Option COUNT nicht zwingend erforderlich ist. Tatsächlich ist die einzige erforderliche Option des Befehls die Option STREAMS, die eine Liste von Strömen zusammen mit der entsprechenden maximalen Identifikationsnummer definiert. Wir haben „STREAMS mystream 0“ geschrieben – wir möchten alle Einträge des Streams mystream mit einer ID größer als „0-0“ erhalten. Wie aus dem Beispiel ersichtlich, gibt der Befehl den Namen des Streams zurück, da wir uns gleichzeitig auf mehrere Streams abonnieren können. Wir könnten beispielsweise „STREAMS mystream otherstream 0 0“ schreiben. Bitte beachten Sie, dass wir nach der Option STREAMS zunächst die Namen aller benötigten Streams und erst danach die Liste der Identifikatoren angeben müssen.

In dieser einfachen Form tut der Befehl im Vergleich zu XRANGEnichts Besonderes. Interessant ist jedoch, dass wir leicht XREAD in einen blockierenden Befehl umwandeln können, indem wir das Argument BLOCK angeben:

> XREAD BLOCK 0 STREAMS mystream $
Im obigen Beispiel wird die neue BLOCK-Option mit einer Wartezeit von 0 Millisekunden angegeben (was unendliches Warten bedeutet). Darüber hinaus wurde anstelle der Übertragung einer herkömmlichen ID für den Stream mystream eine spezielle ID $ übergeben. Diese spezielle ID bedeutet, dass XREAD als ID die maximale ID im Stream mystream verwendet werden muss. Dadurch erhalten wir nur neue Nachrichten, beginnend mit dem Moment, als wir mit dem Abhören begonnen haben. In gewissem Sinne ähnelt das dem Unix-Befehl „tail -f“.

Bitte beachten Sie, dass wir beim Einsatz der BLOCK-Option nicht zwingend die spezielle ID $ verwenden müssen. Wir können jede vorhandene ID im Stream nutzen. Wenn der Befehl unsere Anfrage sofort bedienen kann, ohne zu blockieren, wird er das tun, andernfalls wird er blockiert.

Blockierend XREAD kann auch gleichzeitig mehrere Streams abhören, es müssen lediglich die Namen angegeben werden. In diesem Fall gibt der Befehl den ersten Stream zurück, in den Daten eingegangen sind. Der erste Abonnent, der für diesen Stream blockiert wird, erhält die Daten zuerst.

Verbrauchergruppen

In bestimmten Szenarien möchten wir den Zugriff von Abonnenten auf Nachrichten innerhalb eines einzelnen Streams einschränken. Ein Beispiel, wann dies nützlich sein kann, ist eine Warteschlange von Nachrichten mit Workern, die unterschiedliche Nachrichten des Streams empfangen, um die Verarbeitung der Nachrichten zu skalieren.

Stellen wir uns vor, wir haben drei Abonnenten C1, C2, C3 und einen Stream, der die Nachrichten 1, 2, 3, 4, 5, 6, 7 enthält. Die Verarbeitung der Nachrichten erfolgt wie in der folgenden Grafik dargestellt:

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
Um diesen Effekt zu erzielen, verwendet Redis Stream ein Konzept, das als Verbrauchergruppe bezeichnet wird. Dieses Konzept ähnelt einem Pseudo-Abonnenten, der Daten aus dem Stream erhält, aber tatsächlich von mehreren Abonnenten innerhalb der Gruppe bedient wird, was bestimmte Garantien bietet:

Jede Nachricht wird unterschiedlichen Abonnenten innerhalb der Gruppe zugestellt.
Innerhalb der Gruppe werden Abonnenten durch einen Namen identifiziert, der eine case-sensitive Zeichenkette ist. Sollte ein Abonnent vorübergehend aus der Gruppe fallen, kann er sich mit seinem einzigartigen Namen wieder in die Gruppe einfügen.
Jede Consumer Group folgt dem Konzept «erste ungelesene Nachricht». Wenn ein Abonnent neue Nachrichten anfordert, kann er nur die Nachrichten erhalten, die zuvor keinem Abonnenten innerhalb der Gruppe zugestellt wurden.
Es gibt einen Befehl zur ausdrücklichen Bestätigung der erfolgreichen Verarbeitung einer Nachricht durch den Abonnenten. Bis dieser Befehl aufgerufen wird, bleibt die angeforderte Nachricht im Status «pending».
Innerhalb der Consumer Group kann jeder Abonnent die Historie der Nachrichten anfordern, die speziell ihm zugestellt wurden, aber noch nicht verarbeitet sind (im Status «pending»).

In gewissem Sinne könnte der Status der Gruppe so dargestellt werden:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" mit ausstehenden Nachrichten  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" mit ausstehenden Nachrichten 
|       ... (und so weiter)                             
+----------------------------------------+

Jetzt ist es an der Zeit, die Hauptbefehle für die Consumer Group kennenzulernen, nämlich:

XGROUP wird verwendet zur Erstellung, Zerstörung und Verwaltung von Gruppen
XREADGROUP wird verwendet, um einen Stream über eine Gruppe zu lesen
XACK — ist ein Befehl, der es einem Subscriber ermöglicht, eine Nachricht als erfolgreich verarbeitet zu kennzeichnen

Erstellung der Consumer-Gruppe

Angenommen, der Stream mystream existiert bereits. Der Befehl zur Erstellung der Gruppe würde wie folgt aussehen:

> XGROUP CREATE mystream mygroup $ OK
Bei der Erstellung der Gruppe müssen wir die ID angeben, ab der die Gruppe Nachrichten empfangen soll. Wenn wir einfach alle neuen Nachrichten erhalten möchten, können wir die spezielle ID $ verwenden (wie in unserem Beispiel oben). Wenn stattdessen die ID 0 angegeben wird, hat die Gruppe Zugriff auf alle Nachrichten des Streams.

Jetzt, da die Gruppe erstellt ist, können wir sofort mit dem Befehl Nachrichten lesen: XREADGROUP. Dieser Befehl ist sehr ähnlich zu XREAD und unterstützt die optionale BLOCK-Option. Es gibt jedoch eine zwingende OPTION GROUP, die immer mit zwei Argumenten angegeben werden muss: der Name der Gruppe und der Name des Subscribers. Die OPTION COUNT wird ebenfalls unterstützt.

Bevor wir den Stream lesen, lasst uns ein paar Nachrichten hineinlegen:

> 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

Jetzt versuchen wir, diesen Stream über die Gruppe zu lesen:

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

Der oben genannte Befehl besagt folgendes:

«Ich, Alice, Abonnentin der Gruppe mygroup, möchte aus dem Stream mystream eine Nachricht lesen, die bisher noch niemandem zugestellt wurde».

Jedes Mal, wenn ein Abonnent eine Operation innerhalb der Gruppe ausführt, muss er seinen Namen angeben, um sich eindeutig innerhalb der Gruppe zu identifizieren. Im vorherigen Befehl gibt es noch ein weiteres wichtiges Detail — der spezielle Identifikator „>“. Dieser spezielle Identifikator filtert die Nachrichten und lässt nur die zu, die bisher noch nie zugestellt wurden.

In bestimmten Fällen können Sie auch eine echte Identifikationsnummer wie 0 oder eine andere gültige Identifikationsnummer angeben. In diesem Fall lautet der Befehl XREADGROUP gibt Ihnen die Nachrichtengeschichte mit dem Status „ausstehend“ zurück, die dem angegebenen Abonnenten (Alice) zugestellt wurde, aber noch nicht mit dem Befehl bestätigt wurde. XACK.

Wir können dieses Verhalten überprüfen, indem wir sofort die ID 0 angeben, ohne die Option. ANZAHLWir sehen einfach die einzige ausstehende Nachricht, nämlich die Nachricht mit dem Apfel:

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

Wenn wir jedoch die Nachricht als erfolgreich verarbeitet bestätigen, wird sie nicht mehr angezeigt:

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (leere Liste oder Menge)

Jetzt ist Bob an der Reihe, etwas zu lesen:

> 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, ein Mitglied der Gruppe mygroup, hat um nicht mehr als zwei Nachrichten gebeten. Der Befehl gibt nur unzustellbare Nachrichten aufgrund der speziellen ID „>“ aus. Wie Sie sehen, wird die Nachricht „apple“ nicht angezeigt, da sie bereits an Alice zugestellt wurde, weshalb Bob „orange“ und „strawberry“ erhält.

So kann Alice, Bob und jeder andere Abonnent der Gruppe verschiedene Nachrichten aus demselben Stream lesen. Außerdem können sie ihre Historie der unverarbeiteten Nachrichten einsehen oder Nachrichten als verarbeitet kennzeichnen.

Es gibt einige Dinge, die Sie beachten sollten:

Sobald ein Abonnent eine Nachricht als Befehl betrachtet, XREADGROUPwechselt diese Nachricht in den Zustand „ausstehend“ und wird diesem bestimmten Abonnenten zugewiesen. Andere Abonnenten der Gruppe können diese Nachricht nicht lesen.
Abonnenten werden automatisch beim ersten Erwähnen erstellt, eine explizite Erstellung ist nicht erforderlich.
Mit XREADGROUP Sie können Nachrichten aus mehreren verschiedenen Streams gleichzeitig lesen. Damit dies funktioniert, müssen Sie vorher Gruppen mit demselben Namen für jeden Stream erstellen mithilfe von XGROUP

Wiederherstellung nach einem Ausfall

Der Abonnent kann sich nach einem Ausfall wiederherstellen und seine Liste der Nachrichten mit dem Status „ausstehend“ erneut lesen. Allerdings können Abonnenten in der realen Welt endgültig ausfallen. Was passiert mit den ausstehenden Nachrichten eines Abonnenten, wenn er sich nach einem Ausfall nicht wiederherstellen konnte?
Die Consumer Group bietet eine Funktion an, die genau für solche Fälle verwendet wird – wenn es notwendig ist, den Eigentümer von Nachrichten zu wechseln.

Als Erstes müssen Sie den Befehl aufrufen XPENDING, der alle Nachrichten der Gruppe mit dem Status „pending“ anzeigt. In seiner einfachsten Form wird der Befehl nur mit zwei Argumenten aufgerufen: dem Namen des Streams und dem Namen der Gruppe:

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

Der Befehl zeigt die Anzahl der unverarbeiteten Nachrichten für die gesamte Gruppe und für jeden Abonnenten an. Wir haben nur Bob mit zwei unverarbeiteten Nachrichten, da die einzige Nachricht, die von Alice angefordert wurde, mit XACK.

Wir können zusätzliche Informationen anfordern, indem wir mehr Argumente verwenden:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — Bereich von Identifikatoren (Sie können „-“ und „+“ verwenden)
{count} — Anzahl der Zustellversuche
{consumer-name} — Name der Gruppe

> 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

Jetzt haben wir die Details für jede Nachricht: die ID, den Namen des Abonnenten, die Stillstandzeit in Millisekunden und schließlich die Anzahl der Zustellversuche. Wir haben zwei Nachrichten von Bob, die jeweils 74.170.458 Millisekunden lang stillstanden, also etwa 20 Stunden.

Bitte beachten Sie, dass uns nichts daran hindert, den Inhalt der Nachricht einfach zu überprüfen, indem Sie XRANGE.

> XRANGE mystream 1526569498055-0 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

Wir müssen die gleiche ID einfach zweimal in den Argumenten wiederholen. Jetzt, da wir eine Vorstellung haben, kann Alice entscheiden, dass Bob nach 20 Stunden Stillstand wahrscheinlich nicht zurückkommt und es an der Zeit ist, diese Nachrichten anzufordern und deren Verarbeitung wieder aufzunehmen, anstelle von Bobs. Dafür verwenden wir den Befehl XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
Mit diesem Befehl können wir eine „fremde“ Nachricht, die noch nicht verarbeitet wurde, durch Übertragung des Eigentums auf {consumer} erhalten. Wir können aber auch eine minimale Stillstandszeit {min-idle-time} angeben. Dies hilft zu vermeiden, dass zwei Kunden gleichzeitig versuchen, das Eigentum an denselben Nachrichten zu übertragen:

Kunde 1: XCLAIM mystream meineGruppe Alice 3600000 1526569498055-0 Kunde 2: XCLAIM mystream meineGruppe Lora 3600000 1526569498055-0
Der erste Kunde reduziert die Ausfallzeiten und erhöht die Anzahl der Lieferungen. So kann der zweite Kunde ihn nicht anfordern.

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

Die Nachricht wurde erfolgreich von Alice angefordert, die nun die Nachricht verarbeiten und bestätigen kann.

Wie im obigen Beispiel zu sehen ist, gibt eine erfolgreiche Anfrage den Inhalt der Nachricht zurück. Dies ist jedoch nicht zwingend erforderlich. Die JUSTID-Option kann verwendet werden, um nur die Identifikatoren der Nachricht zurückzugeben. Dies ist nützlich, wenn Sie sich nicht für die Details der Nachricht interessieren und die Systemleistung steigern möchten.

Lieferzähler

Der Zähler, den Sie in der Ausgabe beobachten XPENDING — ist die Anzahl der Lieferungen jeder Nachricht. Dieser Zähler erhöht sich auf zwei Arten: wenn die Nachricht erfolgreich angefordert wird über XCLAIM oder wenn ein Aufruf verwendet wird XREADGROUP.

Es ist normal, dass einige Nachrichten mehrfach zugestellt werden. Das Wichtigste ist, dass am Ende alle Nachrichten verarbeitet werden. Manchmal treten bei der Verarbeitung von Nachrichten Probleme auf, entweder aufgrund einer Beschädigung der Nachricht selbst oder weil die Verarbeitung der Nachricht einen Fehler im Handler-Code verursacht. In solchen Fällen kann es vorkommen, dass diese Nachricht von niemandem verarbeitet werden kann. Da wir einen Zähler für die Zustellversuche haben, können wir diesen verwenden, um solche Situationen zu erkennen. Daher wäre es sinnvoll, sobald der Zustellzähler eine von Ihnen festgelegte hohe Zahl erreicht hat, diese Nachricht in einen anderen Stream zu verschieben und den Systemadministrator zu benachrichtigen.

Zustand der Streams

Der Befehl XINFO wird verwendet, um verschiedene Informationen über den Stream und seine Gruppen abzurufen. Zum Beispiel sieht der grundlegende Befehl wie folgt aus:

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

Der obige Befehl zeigt allgemeine Informationen zum angegebenen Stream an. Hier ist ein etwas komplexeres Beispiel:

> XINFO GROUPS mystream
1) 1) Name
   2) "mygroup"
   3) Verbraucher
   4) (Ganzzahl) 2
   5) ausstehend
   6) (Ganzzahl) 2
2) 1) Name
   2) "some-other-group"
   3) Verbraucher
   4) (Ganzzahl) 1
   5) ausstehend
   6) (Ganzzahl) 0

Der obige Befehl zeigt allgemeine Informationen zu allen Gruppen des angegebenen Streams an.

> XINFO CONSUMERS mystream mygroup
1) 1) Name
   2) "Alice"
   3) ausstehend
   4) (Ganzzahl) 1
   5) inaktiv
   6) (Ganzzahl) 9104628
2) 1) Name
   2) "Bob"
   3) ausstehend
   4) (Ganzzahl) 1
   5) inaktiv
   6) (Ganzzahl) 83841983

Der obige Befehl zeigt Informationen zu allen Abonnenten des angegebenen Streams und der Gruppe an.
Wenn Sie die Syntax des Befehls vergessen, wenden Sie sich einfach für Hilfe an den Befehl selbst:

> XINFO HELP
1) XINFO {Subbefehl} arg arg ... arg. Die Subbefehle sind:
2) CONSUMERS {key} {groupname}  -- Zeigt die Verbrauchergruppen der Gruppe {groupname} an.
3) GROUPS {key}                 -- Zeigt die Verbrauchergruppen des Streams an.
4) STREAM {key}                 -- Zeigt Informationen über den Stream an.
5) HELP                         -- Gibt diese Hilfe aus.

Grenze der Stream-Größe

Viele Anwendungen möchten nicht endlos Daten in einem Stream sammeln. Oft ist es nützlich, eine maximal zulässige Anzahl von Nachrichten im Stream zu haben. In anderen Fällen ist es ratsam, alle Nachrichten aus dem Stream in einen anderen dauerhaften Speicher zu verschieben, sobald eine bestimmte Stream-Größe erreicht ist. Die Größe des Streams kann mit dem Parameter MAXLEN im Befehl eingeschränkt werden. 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"

Bei der Verwendung von MAXLEN werden alte Einträge automatisch entfernt, wenn die angegebene Länge erreicht ist, wodurch der Stream eine konstante Größe hat. Allerdings erfolgt das Trimmen in diesem Fall nicht auf die effizienteste Weise im Redis-Speicher. Die Situation kann wie folgt verbessert werden:

XADD mystream MAXLEN ~ 1000 * ... Eingabefelder hier ...
Das Argument ~ im obigen Beispiel bedeutet, dass wir die Länge des Streams nicht auf einen bestimmten Wert beschränken müssen. In unserem Beispiel kann es jede Zahl größer oder gleich 1000 sein (zum Beispiel 1000, 1010 oder 1030). Wir haben lediglich festgelegt, dass wir möchten, dass unser Stream mindestens 1000 Einträge speichert. Das macht die Arbeit mit dem Speicher innerhalb von Redis deutlich effizienter.

Es gibt auch einen separaten Befehl XTRIM, der dasselbe ausführt:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Dauerhafte Speicherung und Replikation

Redis Streams werden asynchron auf Slave-Knoten repliziert und in AOF-Dateien (Snapshot aller Daten) und RDB-Dateien (Log aller Schreiboperationen) gespeichert. Auch die Replikation des Zustands von Consumer Groups wird unterstützt. Daher hat eine Nachricht, die sich im Status „pending“ auf dem Master-Knoten befindet, auf den Slave-Knoten denselben Status.

Entfernen einzelner Elemente aus dem Stream

Zum Löschen von Nachrichten gibt es einen speziellen Befehl XDEL. Der Befehl erhält den Namen des Streams, gefolgt von den IDs der Nachrichten, die gelöscht werden sollen:

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

Bei der Verwendung dieses Befehls ist zu beachten, dass der Speicher tatsächlich nicht sofort freigegeben wird.

Streams mit einer Länge von null

Der Unterschied zwischen Streams und anderen Redis-Datenstrukturen liegt darin, dass bei anderen Datenstrukturen, wenn sie keine Elemente mehr enthalten, die gesamte Datenstruktur als Nebeneffekt aus dem Speicher entfernt wird. So wird beispielsweise ein sortiertes Set vollständig gelöscht, wenn der Aufruf von ZREM das letzte Element entfernt. Bei Streams ist es jedoch erlaubt, im Speicher zu bleiben, auch wenn sie kein Element enthalten.

Fazit

Redis Streams eignen sich ideal für den Aufbau von Nachrichtenschaltern, Nachrichtenwarteschlangen, einheitlichen Protokollen und Chatsystemen, die eine Historie speichern.

Wie einmal gesagt wurde Niklaus Wirth, sind Programme — Algorithmen plus Datenstrukturen, und Redis bietet bereits beides.

Quelle: habr.com