Redisストリームの使用方法

Origin

Redis 4.0 にモジュールが導入されて以来、ユーザーはこれらの問題を解決する方法を考え始めました。ユーザーの 1 人、Timothy Downs は IRC 経由で次のように語ってくれました:

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

彼のアイデアは私にインスピレーションを与えました。数日考えた後、これはすべての問題を一度に解決するチャンスかもしれないと気づきました。 「ログ」という概念が何であるかを再考する必要があります。ロギングは、ファイルを追加モードで開き、特定の形式でデータを書き込むだけなので、誰もが使用したことのある基本的なプログラミング要素です。ただし、Redis データ構造は抽象的である必要があります。これらはメモリ内にあり、私たちがメモリを使用するのは怠惰だからではなく、いくつかのポインタを使用することでデータ構造を概念化し、明示的な制約から抽象化できるからです。たとえば、一般にロギングにはいくつかの問題があります。オフセットは論理的なものではなく、実際のバイト オフセットです。エントリが挿入された時間に関連する論理的なオフセットが必要な場合はどうすればよいでしょうか。範囲クエリを利用できます。同様に、ログはガベージ コレクションが難しいことがよくあります。追加することしかできないデータ構造内の古い要素をどうやって削除するのでしょうか?理想的なログでは、「*** という番号のエントリが必要で、古い要素は一切不要です」とだけ言うことになります。

Timothy のアイデアに触発されて仕様を書こうとしたとき、Redis クラスターの基数ツリーを使用して実装し、その一部を最適化しました。これにより、スペース効率の高いログを実装するための基礎が提供され、同時に 対数時間 で範囲にアクセスすることが可能になります。同時に、追加のインスピレーションを得るために、Kafka のストリーム関連コンテンツについて読み始めました。これは、私のデザインにも非常に適しており、*** Kafka を基にしています消費者グループ消費者グループ コンセプトを採用し、メモリ内で使用されている Redis と連携できるように Redis 用に再度最適化しました。しかし、仕様は紙の上に残されており、他の人との議論から得た提案の多くを Redis のアップグレードに組み込むために、時間をかけて最初から最後までほとんど書き直しました。 Redis ストリーミングが、一般的なイベントやメッセージ型のアプリケーションだけでなく、時系列でも役立つ機能になることを願っています。

コードを書いてみましょう

Redis カンファレンスから戻った後、夏の間、listpack というライブラリを実装しました。このライブラリは、単一の割り当てで文字列要素のリストを表すデータ構造である

ziplist.c の後継です。これは非常に特殊なシリアル化形式であり、逆順序 (右から左) で解析する機能も備えており、さまざまな使用例でジップリストを置き換えることができます。

基数ツリーとリストパックの機能を組み合わせると、スペース効率の高いログを簡単に構築できます。また、インデックス付けも可能です。つまり、ID と時間によるランダム アクセスが許可されます。これを準備した後、ストリーミング データ構造を実装するためのコードを書き始めました。まだ実装を終えているところですが、現在は Github 上の Redis の streams ブランチで実行されています。 API が 100% 最終版であるとは主張しませんが、興味深い事実が 2 つあります。1 つは、当時はコンシューマ グループのみが欠落していた点と、フローを操作するためのそれほど重要ではないコマンドがいくつかありましたが、大きな側面はすべて変更されました。達成。次に、すべての面が比較的安定したら、すべてのストリーミング機能

backport を約 2 か月以内に 4.0 ブランチにバックポートすることにしました。つまり、ストリームを使用したい Redis ユーザーは、Redis 4.2 がリリースされるまで待つ必要はなく、すぐに実稼働環境で使用できるようになります。新しいデータ構造として、ほとんどすべてのコード変更が新しいコードに現れるため、これが可能になります。ブロッキング リスト操作を除く: コードがリファクタリングされ、ストリームとリストのブロッキング操作に同じコードを共有することで、Redis の内部実装が大幅に簡素化されました。 チュートリアル: Redis ストリームへようこそ

ある意味では、ストリームは Redis リストの拡張バージョンと考えることができます。ストリーム要素は単一の文字列ではなくなり、

field

field と valuevalue で構成されるオブジェクトになりました。範囲クエリはより適用可能で高速です。ストリームでは、各エントリには論理オフセットである ID があります。さまざまなクライアントは、指定された ID より大きい要素に対して block-waitblocking-wait できます。 Redis ストリーミングの基本的なコマンドは XADD です。はい、すべての Redis ストリーミング コマンドにはプレフィックス #XX が付けられます。

> 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></name> 选项。相同组的所有客户端将自动得到 不同的 消息。当然，同一个流可以被多个组读取。在这种情况下，所有的组将收到流中到达的消息的相同副本。但是，在每个组内，消息是不会重复的。

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

内存使用和节省加载时间

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

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

以上がRedisストリームの使用方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。