Redis Stream — ein neuer abstrakter Datentyp, der mit der Veröffentlichung von 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 Identifikation. Standardmäßig wird die Identifikation automatisch generiert und enthält einen Zeitstempel. Daher können Sie Zeitbereiche von Einträgen abfragen oder neue Daten empfangen, während sie in den Stream gelangen, ganz ähnlich wie der Unix-Befehl „tail -f“ eine Protokolldatei liest und auf neue Daten wartet. Beachten Sie, dass mehrere Clients gleichzeitig auf den Stream hören können, so 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 lange bestehenden Redis-Strukturen Revue passieren, die teilweise die Funktionalität von Redis Stream widerspiegeln.

Redis PUB/SUB

Redis Pub/Sub ist ein einfaches Nachrichtensystem, das bereits in Ihrem Key-Value-Speicher eingebaut ist. Doch der Einfachheit ist ein Preis beigefügt:

Wenn der Publisher aus irgendeinem Grund ausfällt, verliert er alle seine Abonnenten.
Der Publisher muss die genauen Adressen aller seiner Abonnenten kennen.
Der Publisher kann die Abonnenten überlasten, wenn Daten schneller veröffentlicht werden, als sie verarbeitet werden können.
Nach der Veröffentlichung wird die Nachricht sofort aus dem Puffer des Publishers entfernt, unabhängig davon, wie vielen Abonnenten sie zugestellt wurde und wie schnell diese die Nachricht verarbeitet haben.
Alle Abonnenten erhalten die Nachricht gleichzeitig. Die Abonnenten müssen selbst eine Vereinbarung über die Reihenfolge der Verarbeitung derselben Nachricht treffen.
Es gibt keinen eingebauten Mechanismus zur Bestätigung der erfolgreichen Verarbeitung einer Nachricht durch den Abonnenten. Wenn der Abonnent die Nachricht erhält und während der Verarbeitung abstürzt, erfährt der Publisher nichts davon.

Redis List

Redis List ist eine Datenstruktur, die Leseoperationen mit Blockierung unterstützt. Sie können Nachrichten am Anfang oder am Ende der Liste hinzufügen und auslesen. Auf Basis dieser Struktur können Sie einen ordentlichen Stapel oder eine Warteschlange für Ihr verteiltes System erstellen, was in den meisten Fällen ausreichend sein wird. Die wichtigsten Unterschiede zu Redis Pub/Sub sind:

Die Nachricht wird an einen Client zugestellt. Der erste Client, der durch das Lesen blockiert wird, erhält die Daten zuerst.
Clint muss die Leseoperation jeder Nachricht selbst einleiten. List weiß nichts über die Kunden.
Nachrichten werden so lange gespeichert, bis sie jemand liest oder ausdrücklich gelöscht wird. Wenn Sie den Redis-Server so konfiguriert haben, dass er Daten auf die Festplatte speichert, erhöht sich die Zuverlässigkeit des Systems erheblich.

Einführung in Stream

Eintrag zum Stream hinzufügen

Team XADD fügt einen neuen Eintrag zum Stream hinzu. Ein Eintrag ist nicht nur eine Zeichenkette, sondern besteht aus einem oder mehreren Schlüssel-Wert-Paaren. Somit ist jeder Eintrag bereits strukturiert und erinnert an die Struktur 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“. Als zweiten Argument nimmt der Befehl die Identifikation, die dem Eintrag zugewiesen wird — diese Identifikation identifiziert jeden Eintrag im Stream eindeutig. In diesem Fall haben wir jedoch * übergeben, weil wir möchten, dass Redis eine neue Identifikation für uns generiert. Jede neue Identifikation wird inkrementell sein. Daher wird jeder neue Eintrag eine höhere Identifikation im Vergleich zu vorhergehenden Einträgen haben.

Format der Identifikation

Die Identifikation des Eintrags, die vom Befehl zurückgegeben wird, XADD, besteht aus zwei Teilen:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — Unix-Zeit in Millisekunden (Zeit Server Redis). Wenn die aktuelle Zeit jedoch genauso oder kleiner ist als die Zeit des vorhergehenden Eintrags, wird der Zeitstempel des vorhergehenden Eintrags verwendet. Daher bleibt die neue Identifikation immer noch aufsteigend, selbst wenn die Uhr des Servers in die Vergangenheit zurückspringt.

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

Das Format solcher Identifikatoren mag auf den ersten Blick seltsam erscheinen. Ein misstrauischer Leser könnte sich fragen, warum die Zeit Teil des Identifikators ist. Der Grund liegt darin, dass Redis-Streams Bereiche von Identifikatoren abfragen können. Da der Identifikator mit dem Erstellungszeitpunkt des Eintrags verknüpft ist, ermöglicht dies, Zeiträume abzufragen. Wir werden ein konkretes Beispiel betrachten, wenn wir zur Untersuchung des Befehls übergehen. XRANGE.

Wenn der Benutzer aus einem bestimmten Grund seinen eigenen Identifikator angeben muss, der beispielsweise mit einem externen System verknüpft ist, können wir ihn an den Befehl übergeben. XADD anstatt 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 darauf achten müssen, den Identifikator zu erhöhen. In unserem Beispiel beträgt der minimale Identifikator „0-1“, daher wird der Befehl keinen weiteren Identifikator annehmen, der gleich oder kleiner als „0-1“ ist.

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

Anzahl der Einträge im Stream

Man kann die Anzahl der Einträge im Stream ganz einfach mit dem Befehl XLENerhalten. Für unser Beispiel wird dieser Befehl den folgenden Wert zurückgeben:

> XLEN somestream
(integer) 2

Bereichsabfragen — XRANGE und XREVRANGE

Um Daten im 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 (erster Eintrag) und den größten (letzter Eintrag) Identifikator im Stream bezeichnen. Das folgende Beispiel wird alle Einträge des Streams ausgeben.

> 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 besteht aus einem Array von zwei Elementen: dem Identifikator und einer Liste von Schlüssel-Wert-Paaren. Wir haben bereits erwähnt, dass die Identifikatoren der Einträge mit der Zeit verknüpft sind. Daher können wir einen Bereich für einen bestimmten Zeitraum abfragen. Wir können jedoch im Anfrage nicht den vollständigen Identifikator angeben, sondern nur die Unix-Zeit, indem wir den Teil, der sich auf ... sequenceNumberDer weggelassene Teil der Kennung wird automatisch zu null am Anfang des Bereichs und zu dem maximal möglichen Wert am Ende des Bereichs zugewiesen. Nachfolgend ein Beispiel, wie man einen Bereich von zwei Millisekunden anfordern kann.

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

Wir haben nur einen Datensatz in diesem Bereich, jedoch kann das zurückgegebene Ergebnis in echten Datensätzen erheblich sein. Aus diesem Grund XRANGE unterstützt die OPTION COUNT. Indem wir eine Anzahl angeben, können wir einfach die ersten N Datensätze erhalten. Wenn wir die nächsten N Datensätze (Pagination) erhalten müssen, können wir die zuletzt erhaltene Kennung verwenden, um diese sequenceNumber um eins zu erhöhen und erneut anzufordern. Lassen Sie uns dies im folgenden Beispiel betrachten. Wir beginnen damit, 10 Elemente hinzuzufügen mit XADD (angenommen, der Stream mystream wurde bereits mit 10 Elementen gefüllt). Um mit der Iteration zu beginnen 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 mit der Iteration der nächsten beiden Elemente fortzufahren, müssen wir die zuletzt erhaltene Kennung auswählen, also 1519073279157-0, und 1 zu sequenceNumber.
der resultierenden Kennung hinzufügen, in diesem Fall 1519073279157-1, die nun als neues Argument für den Startbereich im nächsten Aufruf verwendet werden kann. 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 das Zurückgeben von M Elementen, ist jeder Schritt der Iteration schnell. Somit kann man mit XRANGE effektiv durch Streams iterieren.

Team XREVRANGE ist das Gegenstück 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 Argumente start und stop in umgekehrter Reihenfolge an.

Das Lesen neuer Datensätze mit XREAD

stellt oft die Aufgabe dar, sich auf einen Stream zu abonnieren und nur neue Nachrichten zu erhalten. Dieses Konzept mag dem Redis Pub/Sub oder der blockierenden Redis-Liste ähnlich erscheinen, jedoch gibt es grundlegende Unterschiede in der Nutzung von Redis Stream:

Jede neue Nachricht wird standardmäßig an jeden Abonnenten übermittelt. Dieses Verhalten unterscheidet sich von der blockierenden Redis-Liste, bei der eine neue Nachricht nur von einem bestimmten Abonnenten gelesen wird.
Während in Redis Pub/Sub alle Nachrichten vergessen werden und niemals gespeichert werden, werden in Streams alle Nachrichten auf unbestimmte Zeit gespeichert (es sei denn, der Kunde fordert ausdrücklich die Löschung an).
Redis Streams ermöglichen es, den Zugang zu Nachrichten innerhalb eines Streams zu differenzieren. Ein bestimmter Abonnent kann nur seine persönliche Nachrichtenhistorie sehen.

Sie können sich auf einen Stream abonnieren und neue Nachrichten mit dem Befehl XREADerhalten. Dies 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 handelt es sich um die nicht-blockierende Form. XREADBitte beachten Sie, dass die Option COUNT nicht obligatorisch ist. Tatsächlich besteht die einzige zwingende Option des Befehls aus der Option STREAMS, die eine Liste von Streams zusammen mit der entsprechenden maximalen ID angibt. Wir haben „STREAMS mystream 0“ geschrieben – wir möchten alle Datensätze des Streams mystream mit einer ID größer als „0-0“ erhalten. Wie aus dem Beispiel hervorgeht, 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 zuerst die Namen aller benötigten Streams und dann die Liste der IDs angeben müssen.

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

> XREAD BLOCK 0 STREAMS mystream $
Im obigen Beispiel wurde die neue Option BLOCK mit einer Wartezeit von 0 Millisekunden angegeben (das bedeutet unbegrenztes Warten). Darüber hinaus wurde anstelle einer normalen ID für den Stream mystream eine spezielle ID $ übergeben. Diese spezielle ID bedeutet, dass XREAD als maximalen Identifikator im Stream mystream verwendet werden soll. So erhalten wir nur neue Nachrichten, beginnend ab dem Zeitpunkt, an dem wir das Abhören gestartet haben. In gewisser Hinsicht ähnelt dies dem Unix-Befehl „tail -f“.

Bitte beachten Sie, dass wir bei der Verwendung der BLOCK-Option keinen speziellen Identifier $ benötigen. Wir können jeden bestehenden Identifier im Stream verwenden. Wenn der Befehl unsere Anfrage sofort bedienen kann, ohne zu blockieren, wird er dies tun, andernfalls wird er blockiert.

Blockierend XREAD kann auch mehrere Streams gleichzeitig abhören, indem sie einfach deren Namen angegeben werden. In diesem Fall gibt der Befehl den ersten Stream zurück, in dem Daten eingegangen sind. Der erste Subscriber, der für diesen Stream blockiert ist, erhält die Daten zuerst.

Consumer Groups

In bestimmten Aufgaben möchten wir den Zugriff der Subscriber auf Nachrichten innerhalb eines Streams einschränken. Ein Beispiel, wo dies nützlich sein kann, ist eine Nachrichtenwarteschlange mit Workern, die verschiedene Nachrichten des Streams empfangen, was eine skalierte Verarbeitung von Nachrichten ermöglicht.

Wenn wir uns vorstellen, dass wir drei Subscriber C1, C2, C3 haben und ein Stream, der die Nachrichten 1, 2, 3, 4, 5, 6, 7 enthält, dann erfolgt die Nachrichtenbearbeitung wie im folgenden Diagramm:

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 Consumer Group bezeichnet wird. Dieses Konzept ähnelt einem Pseudo-Subscriber, der Daten aus dem Stream erhält, aber tatsächlich von mehreren Subscribern innerhalb der Gruppe bedient wird, wobei bestimmte Garantien bereitgestellt werden:

Jede Nachricht wird unterschiedlichen Subscribers innerhalb der Gruppe zugestellt.
Innerhalb der Gruppe werden Subscribers nach ihrem Namen identifiziert, der eine case-sensitive Zeichenkette ist. Wenn ein Subscriber vorübergehend aus der Gruppe ausfällt, kann er sich unter seinem eigenen einzigartigen Namen wieder in die Gruppe einfügen.
Jede Consumer Group folgt dem Konzept der „ersten ungelesenen Nachricht“. Wenn ein Subscriber neue Nachrichten anfordert, kann er nur die Nachrichten erhalten, die zuvor keinem Subscriber innerhalb der Gruppe zugestellt wurden.
Es gibt einen Befehl zur expliziten Bestätigung der erfolgreichen Verarbeitung einer Nachricht durch den Subscriber. Solange dieser Befehl nicht aufgerufen wird, bleibt die angeforderte Nachricht im Status „pending“.
Innerhalb der Consumer Group kann jeder Subscriber die Historie der Nachrichten abfragen, die ihm zugestellt wurden, aber noch nicht bearbeitet wurden (im Status „pending“).

In gewissem Sinne kann der Zustand der Gruppe wie folgt 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 grundlegenden Befehle für die Consumer-Gruppe kennenzulernen:

XGROUP wird verwendet, um Gruppen zu erstellen, zu löschen und zu verwalten
XREADGROUP wird verwendet, um den Stream über die Gruppe zu lesen
XACK – dieser Befehl ermöglicht es dem Abonnenten, eine Nachricht als erfolgreich verarbeitet zu markieren

Erstellung der Consumer-Gruppe

Angenommen, der Stream mystream existiert bereits. Dann würde der Befehl zur Erstellung der Gruppe folgendermaßen aussehen:

> XGROUP CREATE mystream mygroup $ OK
Bei der Erstellung der Gruppe müssen wir die ID übergeben, ab der die Gruppe Nachrichten empfangen wird. Wenn wir nur alle neuen Nachrichten erhalten möchten, können wir eine spezielle ID $ verwenden (wie in unserem obigen Beispiel). Wenn anstelle der speziellen ID 0 angegeben wird, sind der Gruppe alle Nachrichten des Streams zugänglich.

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

Bevor wir den Stream lesen, lassen Sie uns einige Nachrichten dort 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 obige Befehl besagt wörtlich folgendes:

„Ich, Alice der Abonnent, Mitglied der Gruppe mygroup, möchte eine Nachricht aus dem Stream mystream lesen, die zuvor noch nie zugestellt wurde.“

Jedes Mal, wenn ein Abonnent eine Operation mit einer Gruppe durchführt, muss er seinen Namen angeben, um sich innerhalb der Gruppe eindeutig zu identifizieren. In dem oben genannten Befehl gibt es noch ein weiteres sehr wichtiges Detail – die spezielle Kennung „>“. Diese spezielle Kennung filtert die Nachrichten und zeigt nur die an, die bisher noch nie zugestellt wurden.

In besonderen Fällen können Sie auch eine echte Kennung wie 0 oder jede andere gültige Kennung angeben. In diesem Fall wird der Befehl XREADGROUP Ihnen eine Historie von Nachrichten mit dem Status „pending“ zurückgeben, die an den angegebenen Abonnenten (Alice) zugestellt wurden, jedoch noch nicht mit dem Befehl bestätigt wurden. XACK.

Wir können dieses Verhalten überprüfen, indem wir sofort die Kennung 0 ohne Option angeben. COUNT. Wir 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 es an Bob, 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, bat um nicht mehr als zwei Nachrichten. Der Befehl meldet nur nicht zugestellte Nachrichten aufgrund der speziellen Kennung „>“. Wie Sie sehen, wird die Nachricht „apple“ nicht angezeigt, da sie bereits an Alice zugestellt wurde, sodass Bob „orange“ und „strawberry“ erhält.

So können Alice, Bob und jeder andere Abonnent der Gruppe verschiedene Nachrichten aus demselben Stream lesen. Sie können auch ihre Historie der unbearbeiteten Nachrichten lesen oder Nachrichten als verarbeitet markieren.

Es gibt einige Dinge, die man beachten sollte:

Sobald ein Abonnent eine Nachricht mit dem Befehl XREADGROUP, wird diese Nachricht in den Status „pending“ versetzt und an diesen spezifischen Abonnenten gebunden. Andere Abonnenten der Gruppe können diese Nachricht nicht lesen.
Abonnenten werden automatisch beim ersten Bezug erstellt, es ist keine explizite Erstellung erforderlich.
Mit Hilfe von XREADGROUP Sie können Nachrichten aus mehreren verschiedenen Streams gleichzeitig lesen, jedoch müssen Sie dafür zuvor Gruppen mit denselben Namen für jeden Stream erstellen mit XGROUP

Wiederherstellung nach einem Ausfall

Ein Subscriber kann sich nach einem Fehler wiederherstellen und seine Liste der Nachrichten mit dem Status „pending“ erneut lesen. In der realen Welt können Subscribers jedoch endgültig scheitern. Was passiert mit den schwebenden Nachrichten eines Subscribers, wenn er nach einem Fehler nicht wiederhergestellt werden kann?
Die Consumer-Gruppe bietet eine Funktion, die genau für solche Fälle genutzt wird — wenn es notwendig ist, den Besitzer von Nachrichten zu wechseln.

Zuerst muss der Befehl XPENDING, aufgerufen werden, der alle Nachrichten der Gruppe mit dem Status „pending“ anzeigt. In seiner einfachsten Form wird der Befehl nur mit zwei Argumenten aufgerufen: dem Streamnamen und dem Gruppennamen:

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

Der Befehl gab die Anzahl der unbearbeiteten Nachrichten für die gesamte Gruppe und für jeden Subscriber aus. Wir haben nur Bob mit zwei unbearbeiteten 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 der Identifikatoren (es können „-“ und „+“ verwendet werden)
{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: Identifikator, Name des Subscribers, Zeit in Millisekunden und schließlich die Anzahl der Zustellversuche. Wir haben zwei Nachrichten von Bob, und sie stehen seit 74170458 Millisekunden still, etwa 20 Stunden.

Beachten Sie, dass uns niemand daran hindert zu überprüfen, wie der Inhalt der Nachricht war, indem wir einfach XRANGE.

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

Wir müssen nur denselben Identifikator zwei Mal in den Argumenten wiederholen. Jetzt, wo wir eine Vorstellung haben, kann Alice entscheiden, dass Bob nach 20 Stunden Stillstand wahrscheinlich nicht wiederhergestellt wird und es an der Zeit ist, diese Nachrichten abzurufen und ihre Verarbeitung anstelle von Bob wieder aufzunehmen. Dazu 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 erhalten, die noch nicht verarbeitet wurde, indem wir den Besitzer auf {consumer} wechseln. Wir können jedoch auch die minimale Leerlaufzeit {min-idle-time} bereitstellen. Dies hilft, Situationen zu vermeiden, in denen zwei Kunden gleichzeitig versuchen, den Besitzer derselben Nachrichten zu wechseln.

Kunde 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Kunde 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
Der erste Kunde setzt die Leerlaufzeit zurück und erhöht den Zustellzähler. 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.

Aus dem obigen Beispiel geht hervor, dass die erfolgreiche Ausführung der Anfrage den Inhalt der Nachricht selbst zurückgibt. Das ist jedoch nicht zwingend erforderlich. Die Option JUSTID 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.

Zustellzähler

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

Es ist in Ordnung, dass einige Nachrichten mehrmals zugestellt werden. Wichtig ist, dass am Ende alle Nachrichten verarbeitet werden. Manchmal treten bei der Verarbeitung einer Nachricht Probleme auf, weil die Nachricht selbst beschädigt ist oder die Verarbeitung der Nachricht einen Fehler im Handler-Code verursacht. In solchen Fällen könnte es sein, dass niemand in der Lage ist, diese Nachricht zu verarbeiten. Da wir einen Zustellversuchszähler haben, können wir diesen Zähler verwenden, um solche Situationen zu erkennen. Daher wird es, sobald der Zustellzähler eine von Ihnen festgelegte hohe Zahl erreicht, wahrscheinlich klüger sein, diese Nachricht in einen anderen Stream zu verschieben und den Systemadministrator zu benachrichtigen.

Zustandsströme

Team XINFO wird verwendet, um verschiedene Informationen über den Stream und dessen Gruppen abzufragen. Beispielsweise 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. Jetzt ein etwas komplizierteres Beispiel:

> 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

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

> 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

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

> XINFO HELP
1) XINFO {subcommand} 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.

Stream-Größenbeschränkung

Viele Anwendungen möchten keine Daten ewig in einem Stream sammeln. Oft ist es nützlich, die maximal zulässige Anzahl von Nachrichten im Stream zu haben. In anderen Fällen ist es sinnvoll, alle Nachrichten aus dem Stream in einen anderen permanenten Speicher zu übertragen, wenn die vorgegebene Größe des Streams 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 gelöscht, wenn die angegebene Länge erreicht ist, sodass der Stream eine konstante Größe hat. Allerdings erfolgt das Trimmen in diesem Fall nicht auf die leistungsfähigste Weise im Speicher von Redis. Die Situation kann wie folgt verbessert werden:

XADD mystream MAXLEN ~ 1000 * ... Eintragsfelder 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 könnte dies jede Zahl größer oder gleich 1000 sein (zum Beispiel 1000, 1010 oder 1030). Wir haben nur ausdrücklich angegeben, dass wir möchten, dass unser Stream mindestens 1000 Einträge speichert. Das macht die Arbeit mit dem Speicher innerhalb von Redis wesentlich effizienter.

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

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Permanente Speicherung und Replikation

Redis Stream wird asynchron auf Slave-Knoten repliziert und in AOF-Dateien (Snapshots aller Daten) und RDB-Dateien (Protokoll aller Schreiboperationen) gespeichert. Außerdem wird die Replikation des Status von Consumer Groups unterstützt. Daher hat eine Nachricht, die sich im Status „ausstehend“ auf dem Master-Knoten befindet, auf den Slave-Knoten denselben Status.

Entfernung einzelner Elemente aus dem Stream

Um Nachrichten zu löschen, gibt es einen speziellen Befehl XDEL. Der Befehl erhält den Namen des Streams, gefolgt von den Identifikatoren 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 besteht darin, dass, wenn andere Datenstrukturen keine Elemente mehr enthalten, die Datenstruktur selbst als Nebeneffekt aus dem Speicher entfernt wird. Ein sortierter Satz wird beispielsweise vollständig entfernt, wenn der Aufruf von ZREM das letzte Element löscht. Streams hingegen dürfen im Speicher bleiben, selbst wenn sie kein Element enthalten.

Fazit

Redis Stream eignet sich ideal für die Erstellung von Message Brokern, Nachrichtenwarteschlangen, einheitlichen Protokollen und Chatsystemen, die eine Historie speichern.

Wie Nikolaj Wirth einmal sagte, Nikolaj Wirth, sind Programme Algorithmen plus Datenstrukturen, und Redis bietet Ihnen bereits beides.

Quelle: habr.com