So verwenden Sie Redis-Streams

Origin

Seit der Einführung von Modulen in Redis 4.0 begannen Benutzer darüber nachzudenken, wie diese Probleme gelöst werden können. Einer der Benutzer, Timothy Downs, sagte mir über IRC:

\<forkfork> 我计划给这个模块增加一个事务日志式的数据类型 &mdash;&mdash; 这意味着大量的订阅者可以在不导致 redis 内存激增的情况下做一些像发布/订阅那样的事情\<forkfork> 订阅者持有他们在消息队列中的位置，而不是让 Redis 必须维护每个消费者的位置和为每个订阅者复制消息

Seine Ideen haben mich inspiriert. Nachdem ich ein paar Tage darüber nachgedacht hatte, wurde mir klar, dass dies unsere Chance sein könnte, alle unsere Probleme auf einen Schlag zu lösen. Ich muss mir das Konzept von „Protokoll“ neu vorstellen. Die Protokollierung ist ein grundlegendes Programmierelement, das jeder verwendet hat, da es einfach eine Datei im Anhängemodus öffnet und Daten in einem bestimmten Format schreibt. Allerdings müssen Redis-Datenstrukturen abstrakt sein. Sie befinden sich im Speicher, und wir nutzen den Speicher nicht, weil wir faul sind, sondern weil wir mithilfe einiger Zeiger Datenstrukturen konzeptualisieren und sie von expliziten Einschränkungen abstrahieren können. Beispielsweise gibt es bei der Protokollierung im Allgemeinen mehrere Probleme: Offsets sind nicht logisch, sondern echte Byte-Offsets. Was ist, wenn Sie logische Offsets benötigen, die sich auf den Zeitpunkt beziehen, zu dem der Eintrag eingefügt wurde? Wir bieten Bereichsabfragen an. Ebenso ist es oft schwierig, Protokolle mit Müll zu sammeln: Wie löscht man alte Elemente in einer Datenstruktur, die nur angehängt werden können? Nun, in unserem idealen Protokoll würden wir einfach sagen: Ich möchte den Eintrag mit der Nummer *** und keines der alten Elemente usw.

Als ich von Timothys Idee inspiriert wurde und versuchte, eine Spezifikation zu schreiben, habe ich den Radix-Baum im Redis-Cluster verwendet, um sie umzusetzen, und einige interne Teile davon optimiert. Dies bildet die Grundlage für die Implementierung eines platzsparenden Protokolls, das dennoch den Zugriff auf den Bereich in logarithmischer Zeit ermöglicht. Gleichzeitig habe ich begonnen, mich über Kafkas Stream-bezogene Inhalte zu informieren, um zusätzliche Inspiration zu erhalten. ***Ich habe mir das Konzept von KafkaVerbrauchergruppen ausgeliehen und es erneut für Redis optimiert , um auf die Situation anzuwenden, in der Redis im Speicher verwendet wird. Allerdings blieb die Spezifikation auf dem Papier und im Laufe der Zeit habe ich sie fast von Anfang bis Ende neu geschrieben, um viele der Vorschläge, die ich aus Gesprächen mit anderen erhalten habe, in das Redis-Upgrade einfließen zu lassen. Ich hoffe, dass Redis-Streaming zu einer nützlichen Funktion für Zeitreihen werden kann und nicht nur zu einer üblichen Ereignis- und Nachrichtenanwendung. Lass uns etwas Code schreiben

Nach meiner Rückkehr von der Redis-Konferenz habe ich den Sommer damit verbracht, eine Bibliothek namens Listpack zu implementieren. Diese Bibliothek ist der Nachfolger von ziplist.c, einer Datenstruktur, die eine Liste von Zeichenfolgenelementen in einer einzelnen Zuordnung darstellte. Es handelt sich um ein ganz besonderes Serialisierungsformat, das die Möglichkeit bietet, auch in umgekehrter Reihenfolge (von rechts nach links) zu analysieren: um Ziplists in verschiedenen Anwendungsfällen zu ersetzen.

Durch die Kombination der Funktionen von Radix-Bäumen und Listpacks ist es einfach, ein platzsparendes Protokoll zu erstellen, das auch indizierbar ist, was bedeutet, dass ein wahlfreier Zugriff nach ID und Zeit zulässig ist. Nachdem dies geschehen war, begann ich, Code zu schreiben, um die Streaming-Datenstruktur zu implementieren. Ich bin immer noch dabei, die Implementierung abzuschließen, aber sie läuft jetzt im Streams-Zweig von Redis auf Github. Ich behaupte nicht, dass die API zu 100 % endgültig ist, aber es gibt zwei interessante Fakten: Erstens fehlte damals nur die Verbrauchergruppe, plus einige weniger wichtige Befehle zur Manipulation des Flusses, aber alle großen Aspekte waren vorhanden erreicht. Zweitens habe ich beschlossen, sobald alle Aspekte relativ stabil sind, alle Streaming-Funktionen in etwa zwei Monaten auf den 4.0-Zweig zurückzuportieren. Dies bedeutet, dass Redis-Benutzer, die Streams nutzen möchten, nicht auf die Veröffentlichung von Redis 4.2 warten müssen, sondern diese sofort in der Produktion verfügbar sind. Dies ist möglich, da als neue Datenstruktur fast alle Codeänderungen im neuen Code erscheinen. Zusätzlich zu Blockierungslistenoperationen: Der Code wurde überarbeitet. Wir verwenden denselben Code für Stream- und Listenblockierungsoperationen, was die interne Implementierung von Redis erheblich vereinfacht. ziplist.c 的继任者，那是一个表示在单个分配中的字符串元素列表的数据结构。它是一个非常特殊的序列化格式，其特点在于也能够以逆序（从右到左）解析：以便在各种用例中替代 ziplists。

结合 radix 树和 listpacks 的特性，它可以很容易地去构建一个空间高效的日志，并且还是可索引的，这意味着允许通过 ID  和时间进行随机访问。自从这些就绪后，我开始去写一些代码以实现流数据结构。我还在完成这个实现，不管怎样，现在在 Github 上的 Redis 的  streams 分支里它已经可以跑起来了。我并没有声称那个 API 是 100%  的最终版本，但是，这有两个有意思的事实：一，在那时只有消费群组是缺失的，加上一些不太重要的操作流的命令，但是，所有的大的方面都已经实现了。二，一旦各个方面比较稳定了之后，我决定大概用两个月的时间将所有的流的特性向后移植backport到  4.0 分支。这意味着 Redis 用户想要使用流，不用等待 Redis 4.2  发布，它们在生产环境马上就可用了。这是可能的，因为作为一个新的数据结构，几乎所有的代码改变都出现在新的代码里面。除了阻塞列表操作之外：该代码被重构了，我们对于流和列表阻塞操作共享了相同的代码，而极大地简化了  Redis 内部实现。

教程：欢迎使用 Redis 的 streams

在某些方面，你可以认为流是 Redis 列表的一个增强版本。流元素不再是一个单一的字符串，而是一个字段field和值value组成的对象。范围查询更适用而且更快。在流中，每个条目都有一个 ID，它是一个逻辑偏移量。不同的客户端可以阻塞等待blocking-wait比指定的 ID 更大的元素。Redis 流的一个基本的命令是 XADD。是的，所有的 Redis 流命令都是以一个 X

Tutorial: Willkommen bei Redis-Streams🎜In gewisser Weise können Sie sich Streams als eine erweiterte Version von Redis-Listen vorstellen. Das Stream-Element ist keine einzelne Zeichenfolge mehr, sondern ein Objekt bestehend aus 🎜field🎜field🎜🎜 und 🎜value🎜value🎜🎜. Bereichsabfragen sind anwendbarer und schneller. In einem Stream hat jeder Eintrag eine ID, die einen logischen Offset darstellt. Verschiedene Clients können auf Elemente mit einer größeren ID als der angegebenen 🎜blockieren-warten🎜blockieren-warten🎜🎜. Ein grundlegender Befehl für Redis-Streaming ist XADD. Ja, allen Redis-Streaming-Befehlen wird ein X vorangestellt. 🎜

> XADD mystream * sensor-id 1234 temperature 10.51506871964177.0

这个 XADD 命令将追加指定的条目作为一个指定的流 —— “mystream” 的新元素。上面示例中的这个条目有两个字段：sensor-id 和 temperature，每个条目在同一个流中可以有不同的字段。使用相同的字段名可以更好地利用内存。有意思的是，字段的排序是可以保证顺序的。XADD 仅返回插入的条目的 ID，因为在第三个参数中是星号（*），表示由命令自动生成 ID。通常这样做就够了，但是也可以去强制指定一个 ID，这种情况用于复制这个命令到从服务器slave server和AOFappend-only file文件。

这个 ID 是由两部分组成的：一个毫秒时间和一个序列号。1506871964177 是毫秒时间，它只是一个毫秒级的 UNIX 时间戳。圆点（.）后面的数字 0  是一个序号，它是为了区分相同毫秒数的条目增加上去的。这两个数字都是 64  位的无符号整数。这意味着，我们可以在流中增加所有想要的条目，即使是在同一毫秒中。ID 的毫秒部分使用 Redis 服务器的当前本地时间生成的  ID 和流中的***一个条目 ID 两者间的***的一个。因此，举例来说，即使是计算机时间回跳，这个 ID  仍然是增加的。在某些情况下，你可以认为流条目的 ID 是完整的 128  位数字。然而，事实上它们与被添加到的实例的本地时间有关，这意味着我们可以在毫秒级的精度的范围随意查询。

正如你想的那样，快速添加两个条目后，结果是仅一个序号递增了。我们可以用一个 MULTI/EXEC 块来简单模拟“快速插入”：

> MULTIOK> XADD mystream * foo 10QUEUED> XADD mystream * bar 20QUEUED> EXEC1) 1506872463535.02) 1506872463535.1

在上面的示例中，也展示了无需指定任何初始模式schema的情况下，对不同的条目使用不同的字段。会发生什么呢？就像前面提到的一样，只有每个块（它通常包含  50-150  个消息内容）的***个消息被使用。并且，相同字段的连续条目都使用了一个标志进行了压缩，这个标志表示与“它们与这个块中的***个条目的字段相同”。因此，使用相同字段的连续消息可以节省许多内存，即使是字段集随着时间发生缓慢变化的情况下也很节省内存。

为了从流中检索数据，这里有两种方法：范围查询，它是通过 XRANGE 命令实现的；流播streaming，它是通过 XREAD 命令实现的。XRANGE 命令仅取得包括从开始到停止范围内的全部条目。因此，举例来说，如果我知道它的 ID，我可以使用如下的命名取得单个条目：

> XRANGE mystream 1506871964177.0 1506871964177.01) 1) 1506871964177.0   2) 1) "sensor-id"      2) "1234"      3) "temperature"      4) "10.5"

不管怎样，你都可以使用指定的开始符号 - 和停止符号 + 表示最小和***的 ID。为了限制返回条目的数量，也可以使用 COUNT 选项。下面是一个更复杂的 XRANGE 示例：

> XRANGE mystream - + COUNT 21) 1) 1506871964177.0   2) 1) "sensor-id"      2) "1234"      3) "temperature"      4) "10.5"2) 1) 1506872463535.0   2) 1) "foo"      2) "10"

这里我们讲的是 ID 的范围，然后，为了取得在一个给定时间范围内的特定范围的元素，你可以使用 XRANGE，因为 ID 的“序号” 部分可以省略。因此，你可以只指定“毫秒”时间即可，下面的命令的意思是：“从 UNIX 时间 1506872463 开始给我 10 个条目”：

127.0.0.1:6379> XRANGE mystream 1506872463000 + COUNT 101) 1) 1506872463535.0   2) 1) "foo"      2) "10"2) 1) 1506872463535.1   2) 1) "bar"      2) "20"

关于 XRANGE 需要注意的最重要的事情是，假设我们在回复中收到 ID，随后连续的 ID 只是增加了序号部分，所以可以使用 XRANGE 遍历整个流，接收每个调用的指定个数的元素。Redis 中的*SCAN 系列命令允许迭代 Redis 数据结构，尽管事实上它们不是为迭代设计的，但这样可以避免再犯相同的错误。

使用 XREAD 处理流播：阻塞新的数据

当我们想通过 ID 或时间去访问流中的一个范围或者是通过 ID 去获取单个元素时，使用 XRANGE 是非常***的。然而，在使用流的案例中，当数据到达时，它必须由不同的客户端来消费时，这就不是一个很好的解决方案，这需要某种形式的汇聚池pooling。（对于 某些 应用程序来说，这可能是个好主意，因为它们仅是偶尔连接查询的）

XREAD 命令是为读取设计的，在同一个时间，从多个流中仅指定我们从该流中得到的***条目的  ID。此外，如果没有数据可用，我们可以要求阻塞，当数据到达时，就解除阻塞。类似于阻塞列表操作产生的效果，但是这里并没有消费从流中得到的数据，并且多个客户端可以同时访问同一份数据。

这里有一个典型的 XREAD 调用示例：

> XREAD BLOCK 5000 STREAMS mystream otherstream $ $

它的意思是：从 mystream 和 otherstream 取得数据。如果没有数据可用，阻塞客户端 5000 毫秒。在 STREAMS 选项之后指定我们想要监听的关键字，***的是指定想要监听的 ID，指定的 ID 为 $ 的意思是：假设我现在需要流中的所有元素，因此，只需要从下一个到达的元素开始给我。

如果我从另一个客户端发送这样的命令：

> XADD otherstream * message “Hi There”

在 XREAD 侧会出现什么情况呢？

1) 1) "otherstream"   2) 1) 1) 1506935385635.0         2) 1) "message"            2) "Hi There"

与收到的数据一起，我们也得到了数据的关键字。在下次调用中，我们将使用接收到的***消息的 ID：

> XREAD BLOCK 5000 STREAMS mystream otherstream $ 1506935385635.0

依次类推。然而需要注意的是使用方式，客户端有可能在一个非常大的延迟之后再次连接（因为它处理消息需要时间，或者其它什么原因）。在这种情况下，期间会有很多消息堆积，为了确保客户端不被消息淹没，以及服务器不会因为给单个客户端提供大量消息而浪费太多的时间，使用 XREAD 的 COUNT 选项是非常明智的。

流封顶

目前看起来还不错……然而，有些时候，流需要删除一些旧的消息。幸运的是，这可以使用 XADD 命令的 MAXLEN 选项去做：

> XADD mystream MAXLEN 1000000 * field1 value1 field2 value2

它是基本意思是，如果在流中添加新元素后发现消息数量超过了 1000000 个，那么就删除旧的消息，以便于元素总量重新回到 1000000 以内。它很像是在列表中使用的 RPUSH + LTRIM，但是，这次我们是使用了一个内置机制去完成的。然而，需要注意的是，上面的意思是每次我们增加一个新的消息时，我们还需要另外的工作去从流中删除旧的消息。这将消耗一些 CPU 资源，所以在计算 MAXLEN 之前，尽可能使用 ~ 符号，以表明我们不要求非常 精确 的 1000000 个消息，就是稍微多一些也不是大问题：

> XADD mystream MAXLEN ~ 1000000 * foo bar

这种方式的 XADD 仅当它可以删除整个节点的时候才会删除消息。相比普通的 XADD，这种方式几乎可以自由地对流进行封顶。

消费组（开发中）

这是***个 Redis 中尚未实现而在开发中的特性。灵感也是来自 Kafka，尽管在这里是以不同的方式实现的。重点是使用了 XREAD，客户端也可以增加一个 GROUP <name> 选项。相同组的所有客户端将自动得到 不同的 消息。当然，同一个流可以被多个组读取。在这种情况下，所有的组将收到流中到达的消息的相同副本。但是，在每个组内，消息是不会重复的。

当指定组时，能够指定一个 RETRY <milliseconds> 选项去扩展组：在这种情况下，如果消息没有通过 XACK 进行确认，它将在指定的毫秒数后进行再次投递。这将为消息投递提供更佳的可靠性，这种情况下，客户端没有私有的方法将消息标记为已处理。这一部分也正在开发中。

内存使用和节省加载时间

因为用来建模 Redis 流的设计，内存使用率是非常低的。这取决于它们的字段、值的数量和长度，对于简单的消息，每使用 100MB  内存可以有几百万条消息。此外，该格式设想为需要极少的序列化：listpack 块以 radix  树节点方式存储，在磁盘上和内存中都以相同方式表示的，因此它们可以很轻松地存储和读取。例如，Redis 可以在 0.3 秒内从 RDB 文件中读取  500 万个条目。这使流的复制和持久存储非常高效。

我还计划允许从条目中间进行部分删除。现在仅实现了一部分，策略是在条目在标记中标识条目为已删除，并且，当已删除条目占全部条目的比例达到指定值时，这个块将被回收重写，如果需要，它将被连到相邻的另一个块上，以避免碎片化。

Das obige ist der detaillierte Inhalt vonSo verwenden Sie Redis-Streams. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!