Comment utiliser les flux Redis

Origin

Depuis l'introduction des modules dans Redis 4.0, les utilisateurs ont commencé à réfléchir à la manière de résoudre ces problèmes. L'un des utilisateurs, Timothy Downs, m'a dit via IRC :

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

Ses idées m'ont inspiré. Après quelques jours de réflexion, j’ai réalisé que cela pourrait être notre chance de résoudre tous nos problèmes d’un seul coup. Je dois réimaginer ce qu'est le concept de « journal ». La journalisation est un élément de programmation de base que tout le monde a utilisé puisqu'il ouvre simplement un fichier en mode ajout et écrit les données dans un certain format. Cependant, les structures de données Redis doivent être abstraites. Ils sont en mémoire, et nous utilisons la mémoire non pas parce que nous sommes paresseux, mais parce qu'en utilisant certains pointeurs, nous pouvons conceptualiser les structures de données et les extraire des contraintes explicites. Par exemple, il y a plusieurs problèmes avec la journalisation en général : les décalages ne sont pas logiques, mais de vrais décalages d'octets, que se passe-t-il si vous voulez des décalages logiques liés à l'heure à laquelle l'entrée a été insérée ? Nous avons des requêtes de plage disponibles. De même, les journaux sont souvent difficiles à récupérer : comment supprimer les anciens éléments d'une structure de données qui ne peuvent être ajoutés ? Eh bien, dans notre journal idéal, nous dirions simplement, je veux l'entrée avec le numéro ***, et aucun des anciens éléments, etc.

Lorsque j'ai été inspiré par l'idée de Timothy et que j'ai essayé d'écrire une spécification, j'ai utilisé l'arbre de base du cluster Redis pour l'implémenter et j'en ai optimisé certaines parties en interne. Cela constitue la base de la mise en œuvre d'un journal qui utilise l'espace de manière efficace tout en permettant d'accéder à la plage en temps logarithmique. En même temps, j'ai commencé à lire sur le contenu lié aux flux de Kafka pour obtenir une inspiration supplémentaire. Il convient également très bien à mon design, *** empruntant à Kafka. Le concept de 🎜🎜# est à nouveau optimisé pour Redis pour s'adapter à la situation. lorsque Redis est utilisé en mémoire. Cependant, la spécification est restée sur papier et, au fil du temps, je l'ai presque réécrite du début à la fin afin d'incorporer dans la mise à niveau Redis de nombreuses suggestions que j'ai reçues lors de discussions avec d'autres. J'espère que le streaming Redis pourra devenir une fonctionnalité utile pour les séries chronologiques, et pas seulement une application courante de type événement et message. Écrivons du codeDe retour de la conférence Redis, j'ai passé l'été à implémenter une bibliothèque appelée listpack. Cette bibliothèque est le successeur de ziplist.c, qui était une structure de données qui représentait une liste d'éléments de chaîne dans une seule allocation. Il s'agit d'un format de sérialisation très spécial qui offre également la possibilité d'analyser dans l'ordre inverse (de droite à gauche) : pour remplacer les listes zip dans divers cas d'utilisation. En combinant les fonctionnalités des arbres de base et des listpacks, il est facile de créer un journal peu encombrant qui est également indexable, ce qui signifie qu'un accès aléatoire par ID et par heure est autorisé. Une fois cela en place, j'ai commencé à écrire du code pour implémenter la structure des données en streaming. Je suis encore en train de terminer l'implémentation, mais elle fonctionne désormais dans la branche streams de Redis sur Github. Je ne prétends pas que l'API est définitive à 100 %, mais il y a deux faits intéressants : premièrement, à cette époque, seul le groupe de consommateurs manquait, plus quelques commandes moins importantes pour manipuler le flux, mais, tous les grands aspects ont été atteint. Deuxièmement, une fois que tous les aspects sont relativement stables, j'ai décidé de prendre environ deux mois pour rétroporter toutes les fonctionnalités de streaming vers la branche 4.0. Cela signifie que les utilisateurs de Redis qui souhaitent utiliser des flux n'ont pas à attendre la sortie de Redis 4.2, ils seront immédiatement disponibles en production. Cela est possible car en tant que nouvelle structure de données, presque toutes les modifications du code apparaissent dans le nouveau code. À l'exception des opérations de blocage de liste : le code a été refactorisé et nous partageons le même code pour les opérations de blocage de flux et de liste, simplifiant grandement l'implémentation interne de Redis.

Tutoriel : Bienvenue dans les flux Redis

D'une certaine manière, vous pouvez considérer les flux comme une version améliorée des listes Redis. L'élément stream n'est plus une chaîne unique, mais un

field

field

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 et

value#🎜🎜#value#🎜🎜##🎜🎜 #Object composé de. Les requêtes par plage sont plus applicables et plus rapides. Dans un flux, chaque entrée possède un ID, qui est un décalage logique. Différents clients peuvent #🎜🎜#Blocking-wait#🎜🎜#blocking-wait#🎜🎜##🎜🎜# pour des éléments plus grands que l'ID spécifié. Une commande de base pour le streaming Redis est XADD. Oui, toutes les commandes de streaming Redis sont préfixées par un X. #🎜🎜#

> 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 万个条目。这使流的复制和持久存储非常高效。

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

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!