Redis Streams を使用する

Redis ストリームは、概念的にはエントリを追加することができるリストです。各エントリは一意の ID と値を持っています。ID はデフォルトで自動生成され、タイムスタンプが含まれます。値はハッシュ値です。範囲内にクエリを行ったり、ブロックコマンドを使って着信のたびにエントリを読み取ったりできます。Redis は一般的に、異なる要素を組み合せることで必要な結果が得られます。かつてニクラウス・ヴィルトが語ったように、プログラムとはアルゴリズムにデータ構造を追加したものですが、Redis はその両方をすでにユーザーに提供しています。

Redis ストリームは、履歴を保持するメッセージブローカー、メッセージキュー、統合ログ、チャットシステムの構築に適しています。いったん送信されると忘れ去られる Pub/Sub メッセージとは異なり、Redis ストリームはメッセージを永続的に保持します。Redis ストリームは、ストリームの要素を使用する際にクライアントのグループが協働することを可能にするコンシューマーグループという機能を実装しています。たとえば、グループ内のコンシューマーは、ID 別に項目を検索したり、項目の処理を確認したり、または保留中のメッセージの所有権を主張したりすることができます。コンシューマーグループについては今後のブログ記事で解説します。

以下は、IRC のチャネルにメッセージを送信する方法の 1 つです。

そして以下は、Redis でメッセージを発行する方法です。

非常によく似ています。動作もほとんど同じです。IRC に送信されたメッセージを読み取りたいときは、まずそこに参加する必要があります。

Redis に発行されたメッセージを読み取る場合も同じです。

こちらも非常によく似ています。両方とも、接続している限り、クライアントは着信したメッセージを読み取れます。メッセージの受信を停止したい場合は、そのためのコマンドも用意されています。IRC の場合は以下です。

Redis で同じ働きをするのは以下です。

#foo という名前は IRC では重要ですが Redis では重要ではありません。他の文字列を使うことも可能です。

それでは、同じ動作をストリームで実装してみましょう。メッセージの発行は以下のようになります。

ここでは何が行われているのでしょうか? 説明すべきことがたくさんありますので、さっそく始めます。

XADD は、ストリームの作成とエントリの発行に使用されるコマンドです。

#foo は、Redis キーの名前で、ストリームを表しています。この名前は任意です。ここでは、IRC のチャネル名を模した #foo を使用しています。

* は、ID の代わりを務める特別な値です。新しい ID をこれまでより大きな値で作成するよう Redis に伝えます。

msg は、フィールドです。エントリの値はハッシュ値であるため、1 つ以上のフィールドと 1 つの値を送信する必要があります。上記の例では、msg がフィールドに選択されています。

"hey!" は、msg の値です。

このコマンドは、今回の単純なユースケースには複雑過ぎるかもしれません。以下に見ていく通り、これは非常に柔軟性の高いコマンドだからです。ひとまず、メッセージの読み取りへ進みましょう。

このコマンドは、その要素まで分解してみないとあまり意味がわかりません。

XREAD は、エントリを取得するためのコマンドです。

BLOCK 1000 は、クライアントが、エントリが全くなければブロックを行い、着信が全くなければ 1000 ミリ秒後にタイムアウトすることを意味します。

STREAMS は、キーのリストとそれに続く ID のリストを受け入れるディレクティブです。ここでは、1 つのキーと 1 つの疑似 ID を送信します。以下をお読みください。

#foo は、Redis キーの名前で、ストリームを表しています。

$ は、疑似 ID で、コマンドがブロックされた後に作成された ID を表しています。つまり、ストリームに存在するそれ以前のエントリをすべて無視し、それ以降に着信したエントリにのみフォーカスすることを意味します。

エントリを取得してもコマンドがタイムアウトしても、さらにエントリを取得するには XREAD を再度呼び出す必要があります。つまり、XREAD は通常、ループ内で呼び出されます。エントリの読み取りを停止したいときは、ループから抜け出せば停止できます。

IRC、Pub/Sub、ストリームを使って同じ結果を得る方法についてみてきました。ストリームには他にも用途があります。ストリームをログとして使用する方法、つまり、エントリをプッシュし、その後 XREAD コマンドを使って着信のたびにエントリを読み取る方法についてもみてきました。当社が提供したバージョンは、ファイルで tail -f を使用する場合と同等です。(tail -f を使い慣れていない場合、詳細を入手するには man tail を実行します)。

過去に交わされた会話に追いつくことができる、最新のチャットアプリケーションをモデル化してみましょう。このバージョンでは、各クライアントは、接続されているか否かにかかわらず、メッセージがストリームに届いたときにすべてのメッセージを読み取ることができます。

過去に交わされた会話だけでなく、入室時点で着信した新規メッセージも閲覧可能なチャットを実装したいときは、クライアントが最後に閲覧されたメッセージを追跡する必要があります。その場合、クライアントは最後に読み取ったエントリの ID を覚えておかなければなりません。ストリーム ID について学ぶこれ以上の機会はありません。ID がどのように作成され使用されるのか、知っておきましょう。

まず、エントリを作成するためのコマンドをおさらいしましょう。

これを Redis 5 にフィードすると、次の応答が返ってきます。

1536495531827-0

これがエントリの新しい ID です。以下のフォーマットに従っています。

<timestamp>-<sequence>

前半はタイムスタンプ (単位はミリ秒) 、後半はシーケンス番号です。Redis が同一のミリ秒内に 2 つ以上の XADD コマンドを受信する場合、シーケンス番号の値が増加します。これにより、2 つのエントリに同じ ID が付くことはありません。また、新しい ID はそれ以前のどの ID よりも大きい値になります。

それでは、履歴を認識したクライアント側の視点で、XREAD コマンドについておさらいします。クライアントを接続するのは初めてで、現時点までのすべてのメッセージの読み取りを希望しているものと仮定します。コマンドは以下のようになります。

今回、クライアントはブロックしていません。#foo にある、ID の値が 0-0 よりも大きいすべてのエントリを呼び出します。確認ですが、0-0 は有効な ID ではありません。Redis に ID を作成させずに自分で ID を作成しても Redis はこの値を拒否します。0-0 は特別な値であるためです。また、頻繁に使用されるため、Redis は 0 を省略形として受け入れます。したがって、コマンドは次のように記述することも可能です。

Redis から得られる結果は以下の通りです。

1) 1) "#foo"
    2) 1) 1) 1536495531827-0
        2) 1) "msg"
            2) "hey!"

redis-cli を使用するとこのような表記になります。たとえば Python などから同じコマンドを呼び出すと、次のような構造になります。

結果を展開して、実際の形状を確認してみましょう。

[
    [
    "#foo",
        [
            [
            "1536495531827-0",
                [
                "msg",
                "hey!"
                ]
            ]
        ]
    ]
]

これはストリームの配列です (ここでは、要素は "#foo" のみです)。各ストリームには、ID およびフィールド/値のペアのリストから構成された、対応するエントリの配列があります。

次はチャットのメタファーです。保存すべき重要情報は最後に閲覧した ID です。したがって、自分のプログラムに "1536495531827-0" を保存する必要があります。新しいエントリをチェックする準備が整ったら、再度 XREAD を呼び出して、最後に閲覧した ID を渡します。

新しいエントリは存在しないので、呼び出しの応答には nil が返ります。新しいエントリを受信していた場合は、最後に閲覧したエントリの ID を更新します。すると、ユーザーに対応しているアプリケーションには、新たに到着した未読のメッセージが表示されます。