Redis Stream は、Redis バージョン 5.0 で導入された新しい抽象データ型です。
概念的には、Redis Stream はエントリを追加できるリストです。 各エントリには一意の識別子があります。 デフォルトでは、ID は自動的に生成され、タイムスタンプが含まれます。 したがって、Unix の「tail -f」コマンドがログ ファイルを読み取り、新しいデータを待機している間にフリーズするのと同じように、時間の経過とともにレコードの範囲をクエリしたり、ストリームに到着した新しいデータを受信したりできます。 多数の「tail -f」プロセスが互いに競合することなく同時にファイルを読み取ることができるのと同様に、複数のクライアントが同時に XNUMX つのスレッドをリッスンできることに注意してください。

新しいデータ型の利点をすべて理解するために、Redis Stream の機能を部分的に複製する、古くから存在する Redis 構造を簡単に見てみましょう。

Redis PUB/SUB

Redis Pub/Sub は、Key-Value ストアにすでに組み込まれているシンプルなメッセージング システムです。 ただし、シンプルさには代償が伴います。

• 発行者が何らかの理由で失敗すると、すべての購読者を失います。
• 発行者は、すべての購読者の正確な住所を知る必要があります。
• データが処理されるよりも早くパブリッシュされると、パブリッシャーはサブスクライバーに仕事で過負荷をかける可能性があります。
• メッセージは、配信されたサブスクライバーの数や、このメッセージをどれだけ早く処理できたかに関係なく、発行直後にパブリッシャーのバッファーから削除されます。
• すべての購読者が同時にメッセージを受信します。 加入者自身は、同じメッセージを処理する順序について何らかの方法で加入者間で合意する必要があります。
• サブスクライバがメッセージを正常に処理したことを確認する組み込みメカニズムはありません。 サブスクライバーがメッセージを受信し、処理中にクラッシュしても、パブリッシャーはそれを知りません。

Redis リスト

Redis List は、読み取りコマンドのブロックをサポートするデータ構造です。 リストの先頭または末尾からメッセージを追加したり、読んだりできます。 この構造に基づいて、分散システム用に適切なスタックまたはキューを作成できます。ほとんどの場合、これで十分です。 Redis Pub/Sub との主な違い:

• メッセージは XNUMX 人のクライアントに配信されます。 最初に読み取りブロックされたクライアントが最初にデータを受信します。
• クリントは各メッセージの読み取り操作を自分で開始する必要があります。 リストはクライアントについて何も知りません。
• メッセージは、誰かが読むか明示的に削除するまで保存されます。 データをディスクにフラッシュするように Redis サーバーを構成すると、システムの信頼性が大幅に向上します。

ストリームの概要

ストリームへのエントリの追加

チーム XADD 新しいエントリをストリームに追加します。 レコードは単なる文字列ではなく、XNUMX つ以上のキーと値のペアで構成されます。 したがって、各エントリはすでに構造化されており、CSV ファイルの構造に似ています。

> XADD mystream * sensor-id 1234 temperature 19.8
1518951480106-0

上の例では、名前（キー）「mystream」を持つストリームに 1234 つのフィールドを追加します。「センサー ID」と「温度」の値は、それぞれ「19.8」と「XNUMX」です。 このコマンドは XNUMX 番目の引数として、エントリに割り当てられる識別子を受け取ります。この識別子は、ストリーム内の各エントリを一意に識別します。 ただし、この場合は、Redis に新しい ID を生成してもらいたいため、* を渡しました。 新しいIDが増えるたびに増えていきます。 したがって、新しいエントリはそれぞれ、以前のエントリと比較してより高い識別子を持ちます。

識別子の形式

コマンドによって返されるエントリ ID XADDは XNUMX つの部分で構成されます。

{millisecondsTime}-{sequenceNumber}

ミリ秒時間 — ミリ秒単位の Unix 時間 (Redis サーバー時間)。 ただし、現在の時刻が前の記録の時刻と同じかそれより小さい場合は、前の記録のタイムスタンプが使用されます。 したがって、サーバー時間が過去に戻った場合でも、新しい識別子は増分プロパティを保持したままになります。

シーケンス番号 同じミリ秒内に作成されたレコードに使用されます。 シーケンス番号 前のエントリに比べて 1 増加します。 なぜなら シーケンス番号 サイズが 64 ビットである場合、実際には、XNUMX ミリ秒以内に生成できるレコード数の制限に遭遇する必要はありません。

このような識別子の形式は、一見すると奇妙に見えるかもしれません。 不信感を抱いている読者は、なぜ時間が識別子の一部なのか疑問に思うかもしれません。 その理由は、Redis ストリームが ID による範囲クエリをサポートしているためです。 識別子はレコードの作成時刻に関連付けられているため、時間範囲をクエリすることが可能になります。 コマンドを見て具体的な例を見てみましょう。 エクスレンジ.

何らかの理由でユーザーが自分の識別子を指定する必要がある場合（たとえば、外部システムに関連付けられている場合）、それをコマンドに渡すことができます。 XADD 以下に示すように * の代わりに:

> XADD somestream 0-1 field value
0-1
> XADD somestream 0-2 foo bar
0-2

この場合、ID の増加を自分で監視する必要があることに注意してください。 この例では、最小の識別子は「0-1」であるため、コマンドは「0-1」以下の別の識別子を受け入れません。

> XADD somestream 0-1 foo bar
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item

ストリームごとのレコード数

コマンドを使用するだけでストリーム内のレコード数を取得できます。 XLEN。 この例では、このコマンドは次の値を返します。

> XLEN somestream
(integer) 2

範囲クエリ - XRANGE および XREVRANGE

範囲別にデータをリクエストするには、範囲の先頭と末尾の XNUMX つの識別子を指定する必要があります。 返される範囲には、境界を含むすべての要素が含まれます。 また、XNUMX つの特別な識別子「-」と「+」もあり、それぞれストリーム内の最小 (最初のレコード) と最大 (最後のレコード) の識別子を意味します。 以下の例では、すべてのストリーム エントリをリストします。

> 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"

返される各レコードは、識別子とキーと値のペアのリストという XNUMX つの要素の配列です。 レコード識別子は時間に関連しているとすでに述べました。 したがって、特定の期間の範囲をリクエストできます。 ただし、リクエストでは完全な識別子ではなく、Unix 時間のみを指定でき、それに関連する部分は省略されます。 シーケンス番号。 識別子の省略された部分は、範囲の先頭では自動的にゼロに設定され、範囲の末尾では可能な最大値に設定されます。 以下は、XNUMX ミリ秒の範囲をリクエストする方法の例です。

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

この範囲にはエントリが XNUMX つだけありますが、実際のデータ セットでは返される結果が膨大になる可能性があります。 このため エクスレンジ COUNT オプションをサポートします。 数量を指定すると、最初の N 個のレコードを簡単に取得できます。 次の N レコード (ページネーション) を取得する必要がある場合は、最後に受信した ID を使用して、ID を増やすことができます。 シーケンス番号 一つずつ言って、もう一度質問してください。 次の例でこれを見てみましょう。 10 個の要素の追加を開始します XADD (mystream にすでに 10 個の要素が入力されていると仮定します)。 コマンドごとに 2 つの要素を取得する反復を開始するには、全範囲から開始しますが、COUNT が 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"

次の 1519073279157 つの要素の反復を続けるには、最後に受信した ID (つまり 0-1) を選択し、それに XNUMX を加算する必要があります。 シーケンス番号.
結果の ID (この場合は 1519073279157-1) は、次の呼び出しの新しい範囲開始引数として使用できるようになりました。 エクスレンジ:

> 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"

等々。 複雑なため エクスレンジ 検索に O(log(N)) を使用し、次に M 要素を返すために O(M) を使用すると、各反復ステップが高速になります。 したがって、使用すると、 エクスレンジ ストリームを効率的に反復できます。

チーム XREVRANGE 同等です エクスレンジですが、要素は逆の順序で返されます。

> XREVRANGE mystream + - COUNT 1
1) 1) 1519073287312-0
   2) 1) "foo"
      2) "value_10"

コマンドに注意してください XREVRANGE 範囲引数の start と stop を逆の順序で受け取ります。

XREADを使用した新しいエントリの読み取り

多くの場合、ストリームをサブスクライブして新しいメッセージのみを受信するというタスクが発生します。 この概念は Redis Pub/Sub または Redis List のブロックに似ているように見えるかもしれませんが、Redis Stream の使用方法には根本的な違いがあります。

1. デフォルトでは、新しいメッセージはすべてすべての購読者に配信されます。 この動作は、新しいメッセージを XNUMX 人のサブスクライバーのみが読み取るブロッキング Redis リストとは異なります。
2. Redis Pub/Sub では、すべてのメッセージは忘れられ、永続化されませんが、Stream では、すべてのメッセージは (クライアントが明示的に削除しない限り) 無期限に保持されます。
3. Redis Stream を使用すると、XNUMX つのストリーム内のメッセージへのアクセスを区別できます。 特定の購読者は、自分の個人メッセージ履歴のみを表示できます。

コマンドを使用してスレッドを購読し、新しいメッセージを受信できます。 XREAD。 よりも少し複雑です エクスレンジ, そのため、最初は簡単な例から始めます。

> 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"

上の例は、ノンブロッキング フォームを示しています。 XREAD。 COUNT オプションはオプションであることに注意してください。 実際、必要なコマンド オプションは STREAMS オプションだけです。これは、ストリームのリストと対応する最大識別子を指定します。 「STREAMS mystream 0」と書きました。「0-0」より大きい識別子を持つ mystream ストリームのすべてのレコードを受信したいと考えています。 この例からわかるように、同時に複数のスレッドをサブスクライブできるため、このコマンドはスレッドの名前を返します。 たとえば、「STREAMS mystream otherstream 0 0」と書くことができます。 STREAMS オプションの後には、まず必要なすべてのストリームの名前を指定し、その後で識別子のリストを指定する必要があることに注意してください。

この単純な形式では、コマンドは次のコマンドと比べて特別なことは何も行いません。 エクスレンジ。 しかし、興味深いのは、簡単に向きを変えることができるということです。 XREAD BLOCK 引数を指定してブロッキング コマンドに追加します。

> XREAD BLOCK 0 STREAMS mystream $

上の例では、新しい BLOCK オプションがタイムアウト 0 ミリ秒で指定されています (これは、無期限に待機することを意味します)。 さらに、ストリーム mystream の通常の識別子を渡す代わりに、特別な識別子 $ が渡されました。 この特別な識別子は次のことを意味します XREAD mystream 内の最大の識別子を識別子として使用する必要があります。 したがって、聞き始めた瞬間から新しいメッセージのみを受信することになります。 ある意味、これは Unix の「tail -f」コマンドに似ています。

BLOCK オプションを使用する場合、必ずしも特別な識別子 $ を使用する必要はないことに注意してください。 ストリーム内に存在する任意の識別子を使用できます。 チームがブロックせずにリクエストをすぐに処理できる場合は処理し、そうでない場合はブロックします。

ブロッキング XREAD 名前を指定するだけで、複数のスレッドを同時にリッスンすることもできます。 この場合、コマンドはデータを受信した最初のストリームのレコードを返します。 特定のスレッドに対してブロックされた最初のサブスクライバが最初にデータを受信します。

消費者団体

特定のタスクでは、XNUMX つのスレッド内のメッセージへのサブスクライバーのアクセスを制限したいと考えています。 これが役立つ例としては、スレッドからさまざまなメッセージを受信するワーカーを備えたメッセージ キューが挙げられ、メッセージ処理の拡張が可能になります。

1 人のサブスクライバー C2、C3、C1 と、メッセージ 2、3、4、5、6、7、XNUMX を含むスレッドがあると想像すると、メッセージは次の図のように処理されます。

1 -> C1
2 -> C2
3 -> C3
4 -> C1
5 -> C2
6 -> C3
7 -> C1

この効果を実現するために、Redis Stream は Consumer Group と呼ばれる概念を使用します。 この概念は、ストリームからデータを受信する疑似サブスクライバーに似ていますが、実際にはグループ内の複数のサブスクライバーによってサービスが提供され、特定の保証が提供されます。

1. 各メッセージはグループ内の異なる加入者に配信されます。
2. グループ内では、サブスクライバは名前（大文字と小文字が区別される文字列）によって識別されます。 加入者が一時的にグループから脱落した場合、加入者は独自の固有の名前を使用してグループに復元できます。
3. すべての消費者グループは、「最初の未読メッセージ」の概念に従っています。 サブスクライバが新しいメッセージを要求すると、グループ内のどのサブスクライバにも以前に配信されたことのないメッセージのみを受信できます。
4. メッセージがサブスクライバーによって正常に処理されたことを明示的に確認するコマンドがあります。 このコマンドが呼び出されるまで、要求されたメッセージは「保留中」ステータスのままになります。
5. Consumer Group 内では、各加入者は自分に配信されたがまだ処理されていない (「保留中」ステータス) メッセージの履歴を要求できます。

ある意味、グループの状態は次のように表現できます。

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" with pending messages  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" with pending messages 
|       ... (and so forth)                             
+----------------------------------------+

ここで、Consumer Group の主なコマンドについて説明します。

• Xグループ グループの作成、破棄、管理に使用されます
• XREADGROUP グループを介してストリームを読み取るために使用されます
• ザック - このコマンドを使用すると、サブスクライバーはメッセージを正常に処理されたものとしてマークできます。

消費者団体の創設

mystream がすでに存在すると仮定します。 グループ作成コマンドは次のようになります。

> XGROUP CREATE mystream mygroup $
OK

グループを作成するときは、グループがメッセージを受信する識別子を渡す必要があります。 すべての新しいメッセージを受信したいだけの場合は、特別な識別子 $ を使用できます (上記の例のように)。 特別な識別子の代わりに 0 を指定すると、スレッド内のすべてのメッセージがグループで利用できるようになります。

グループが作成されたので、次のコマンドを使用してすぐにメッセージの読み取りを開始できます。 XREADGROUP。 このコマンドは次とよく似ています XREAD オプションの BLOCK オプションをサポートします。 ただし、必須の GROUP オプションがあり、常に XNUMX つの引数 (グループ名とサブスクライバ名) とともに指定する必要があります。 COUNT オプションもサポートされています。

スレッドを読む前に、メッセージをいくつか入れてみましょう。

> 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

次に、グループを通じてこのストリームを読んでみましょう。

> XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "message"
            2) "apple"

上記のコマンドは、そのままでは次のようになります。

「私、購読者アリス、mygroup のメンバーは、これまで誰にも配信されたことのない XNUMX つのメッセージを mystream から読みたいと思っています。」

サブスクライバは、グループに対して操作を実行するたびに、グループ内で自分自身を一意に識別する名前を提供する必要があります。 上記のコマンドには、特別な識別子「">」という非常に重要な詳細がもう XNUMX つあります。 この特別な識別子はメッセージをフィルタリングし、これまでに配信されたことのないメッセージのみを残します。

また、特殊な場合には、0 などの実際の識別子やその他の有効な識別子を指定できます。 この場合、コマンドは XREADGROUP 指定されたサブスクライバ (Alice) に配信されたものの、コマンドを使用してまだ確認されていない「保留中」ステータスのメッセージの履歴を返します。 ザック.

オプションを使用せずに ID 0 をすぐに指定することで、この動作をテストできます。 COUNT。 単一の保留中のメッセージ、つまり Apple メッセージが表示されるだけです。

> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "message"
            2) "apple"

ただし、メッセージが正常に処理されたことを確認すると、メッセージは表示されなくなります。

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (empty list or set)

今度はボブが何かを読む番です。

> 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"

私のグループのメンバーであるボブは、メッセージを XNUMX つまでにしてほしいと要求しました。 このコマンドは、特別な識別子「">」が原因で配信されなかったメッセージのみを報告します。 ご覧のとおり、「リンゴ」というメッセージはすでにアリスに届けられているため表示されず、ボブは「オレンジ」と「イチゴ」を受け取ります。

こうすることで、アリス、ボブ、およびグループの他の加入者は、同じストリームから異なるメッセージを読み取ることができます。 また、未処理のメッセージの履歴を読んだり、メッセージを処理済みとしてマークしたりすることもできます。

留意すべき点がいくつかあります。

• 加入者がメッセージをコマンドとみなすとすぐに、 XREADGROUP、このメッセージは「保留中」状態になり、その特定の加入者に割り当てられます。 他のグループ購読者はこのメッセージを読むことができません。
• 購読者は最初の言及時に自動的に作成されるため、明示的に作成する必要はありません。
• ととも​​に XREADGROUP 同時に複数の異なるスレッドからメッセージを読み取ることができますが、これを機能させるには、まず、次を使用して各スレッドに同じ名前のグループを作成する必要があります。 Xグループ

障害後の回復

加入者は障害から回復し、「保留中」ステータスのメッセージのリストを再度読み取ることができます。 ただし、現実の世界では、サブスクライバーが最終的に失敗する可能性があります。 サブスクライバが障害から回復できない場合、サブスクライバの滞留メッセージはどうなりますか?
Consumer Group は、まさにそのような場合、つまりメッセージの所有者を変更する必要がある場合に使用される機能を提供します。

最初に行う必要があるのは、コマンドを呼び出すことです 支出中、ステータスが「保留中」のグループ内のすべてのメッセージが表示されます。 最も単純な形式では、コマンドは XNUMX つの引数 (スレッド名とグループ名) だけを使用して呼び出されます。

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

チームは、グループ全体と各加入者の未処理メッセージの数を表示しました。 アリスが要求した唯一のメッセージが確認されたため、ボブには XNUMX つの未処理のメッセージしかありません。 ザック.

より多くの引数を使用して、より多くの情報を要求できます。

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} - 識別子の範囲 (「-」と「+」を使用できます)
{count} — 配信試行回数
{consumer-name} - グループ名

> 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

これで、各メッセージの詳細、つまり ID、加入者名、ミリ秒単位のアイドル時間、そして最後に配信試行回数が得られました。 ボブからのメッセージが 74170458 件あり、20 ミリ秒、つまり約 XNUMX 時間アイドル状態になっています。

を使用するだけでメッセージの内容を確認することを誰も止めていないことに注意してください。 エクスレンジ.

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

引数内で同じ識別子を 20 回繰り返すだけです。 ある程度のアイデアが得られたので、アリスは XNUMX 時間のダウンタイムの後、ボブはおそらく回復しないと判断し、これらのメッセージをクエリしてボブのために処理を再開する時期が来たと判断するかもしれません。 このために、次のコマンドを使用します XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}

このコマンドを使用すると、所有者を {consumer} に変更することで、まだ処理されていない「外部」メッセージを受信できます。 ただし、最小アイドル時間 {min-idle-time} を指定することもできます。 これは、XNUMX つのクライアントが同じメッセージの所有者を同時に変更しようとする状況を回避するのに役立ちます。

Client 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0
Clinet 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0

最初の顧客はダウンタイムをリセットし、配送カウンターを増やします。 したがって、XNUMX 番目のクライアントはそれをリクエストできません。

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

メッセージはアリスによって正常に要求され、アリスはメッセージを処理して確認応答できるようになりました。

上記の例から、リクエストが成功するとメッセージ自体の内容が返されることがわかります。 ただし、これは必須ではありません。 JUSTID オプションは、メッセージ ID のみを返すために使用できます。 これは、メッセージの詳細には興味がなく、システムのパフォーマンスを向上させたい場合に便利です。

宅配カウンター

出力に表示されるカウンター 支出中 各メッセージの配信数です。 このようなカウンタは XNUMX つの方法で増加します。メッセージが次の方法で正常に要求されたとき。 XCLAIM または通話使用時 XREADGROUP.

一部のメッセージが複数回配信されるのは通常のことです。 重要なことは、すべてのメッセージが最終的に処理されるということです。 メッセージ自体が破損していたり​​、メッセージ処理によりハンドラー コードでエラーが発生したりするため、メッセージの処理中に問題が発生することがあります。 この場合、誰もこのメッセージを処理できないことが判明する可能性があります。 配信試行カウンタがあるため、このカウンタを使用してそのような状況を検出できます。 したがって、配信数が指定した高い数に達したら、そのようなメッセージを別のスレッドに置き、システム管理者に通知を送信する方が賢明でしょう。

スレッドの状態

チーム XINFO スレッドとそのグループに関するさまざまな情報を要求するために使用されます。 たとえば、基本的なコマンドは次のようになります。

> 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"

上記のコマンドは、指定されたストリームに関する一般情報を表示します。 次に、もう少し複雑な例を示します。

> 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

上記のコマンドは、指定されたスレッドのすべてのグループの一般情報を表示します。

> 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

上記のコマンドは、指定されたストリームとグループのすべてのサブスクライバーに関する情報を表示します。
コマンド構文を忘れた場合は、コマンド自体にヘルプを求めてください。

> XINFO HELP
1) XINFO {subcommand} arg arg ... arg. Subcommands are:
2) CONSUMERS {key} {groupname}  -- Show consumer groups of group {groupname}.
3) GROUPS {key}                 -- Show the stream consumer groups.
4) STREAM {key}                 -- Show information about the stream.
5) HELP                         -- Print this help.

ストリームサイズの制限

多くのアプリケーションは、データをストリームに永久に収集することを望んでいません。 多くの場合、スレッドごとに許可されるメッセージの最大数を設定すると便利です。 また、指定したスレッド サイズに達したときに、すべてのメッセージをスレッドから別の永続ストアに移動すると便利な場合もあります。 コマンドの MAXLEN パラメータを使用して、ストリームのサイズを制限できます。 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"

MAXLEN を使用すると、古いレコードは指定された長さに達すると自動的に削除されるため、ストリームのサイズは一定になります。 ただし、この場合のプルーニングは、Redis メモリ内で最も効率的な方法では行われません。 次のように状況を改善できます。

XADD mystream MAXLEN ~ 1000 * ... entry fields here ...

上記の例の ~ 引数は、ストリームの長さを特定の値に制限する必要がないことを意味します。 この例では、これは 1000 以上の任意の数値 (1000、1010、または 1030 など) になります。 ストリームに少なくとも 1000 レコードを保存することを明示的に指定しました。 これにより、Redis 内でのメモリ管理がより効率的になります。

別チームもあります エクストリム、同じことを行います:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10


永続的なストレージとレプリケーション

Redis Stream はスレーブ ノードに非同期的にレプリケートされ、AOF (すべてのデータのスナップショット) や RDB (すべての書き込み操作のログ) などのファイルに保存されます。 コンシューマ グループ状態のレプリケーションもサポートされています。 したがって、メッセージがマスター ノードで「保留中」ステータスにある場合、スレーブ ノードでもこのメッセージは同じステータスになります。

ストリームからの個々の要素の削除

メッセージを削除するための特別なコマンドがあります XDEL。 このコマンドは、スレッドの名前と、その後に削除するメッセージ ID を取得します。

> 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"

このコマンドを使用するときは、実際のメモリがすぐに解放されないことを考慮する必要があります。

長さゼロのストリーム

ストリームと他の Redis データ構造の違いは、他のデータ構造に要素が含まれなくなると、副作用としてデータ構造自体がメモリから削除されることです。 したがって、たとえば、ZREM 呼び出しによって最後の要素が削除されると、ソートされたセットは完全に削除されます。 代わりに、スレッドは内部に要素がなくてもメモリ内に残ることができます。

まとめ

Redis Stream は、メッセージ ブローカー、メッセージ キュー、統合ログ、履歴保持チャット システムの作成に最適です。

かつて言ったように ニクラス・ヴィルト、プログラムはアルゴリズムとデータ構造であり、Redis はすでにその両方を提供します。

出所： habr.com