Redis Stream

Redis Stream は Redis 5.0 バージョンで新たに追加されたデータ構造です。

Redis Stream は主にメッセージキュー（MQ、Message Queue）に使用されます。Redis 自体には Redis のパブリッシュ/サブスクライブ (pub/sub) によるメッセージキューの機能がありますが、メッセージを永続化できないという欠点があり、ネットワーク切断や Redis のダウンなどが発生すると、メッセージは破棄されてしまいます。

簡単に言うと、pub/sub はメッセージを配信できますが、履歴メッセージを記録することはできません。

一方、Redis Stream はメッセージの永続化とマスター/スレーブレプリケーション機能を提供し、任意のクライアントがあらゆる時点のデータにアクセスでき、各クライアントのアクセス位置を記憶し、メッセージの喪失を防ぐこともできます。

Redis Stream の構造は次のとおりです。メッセージリンクリストがあり、追加されたすべてのメッセージを連結します。各メッセージには一意の ID と対応するコンテンツがあります。

各 Stream には一意の名前があり、それが Redis のキーです。xadd コマンドを最初に使用してメッセージを追加すると自動的に作成されます。

上の図の解説：

  • Consumer Group：コンシューマグループ。XGROUP CREATE コマンドで作成され、1つのコンシューマグループには複数のコンシューマ（Consumer）が存在します。
  • last_delivered_id：カーソル。各コンシューマグループにはカーソル last_delivered_id があり、いずれかのコンシューマがメッセージを読み取ると、カーソル last_delivered_id が前方に移動します。
  • pending_ids：コンシューマ（Consumer）の状態変数で、コンシューマの未確認 ID を維持する役割があります。pending_ids は、クライアントがすでに読み取ったがまだ ack（Acknowledge character：確認文字）されていないメッセージを記録します。

メッセージキュー関連コマンド：

  • XADD- メッセージを末尾に追加
  • XTRIM- ストリームをトリミングして長さを制限
  • XDEL- メッセージの削除
  • XLEN- ストリームに含まれる要素数、つまりメッセージ長を取得
  • XRANGE- メッセージリストを取得。削除済みメッセージは自動的にフィルタリングされます
  • XREVRANGE- メッセージリストを逆方向に取得。ID は大きいものから小さいものへ
  • XREAD- ブロッキングまたは非ブロッキング方式でメッセージリストを取得

コンシューマグループ関連コマンド：

  • XGROUP CREATE- コンシューマグループの作成
  • XREADGROUP GROUP- コンシューマグループ内のメッセージを読み取る
  • XACK- メッセージを「処理済み」としてマーク
  • XGROUP SETID- コンシューマグループに新しい最後の配信メッセージ ID を設定
  • XGROUP DELCONSUMER- コンシューマの削除
  • XGROUP DESTROY- コンシューマグループの削除
  • XPENDING- 未処理メッセージの関連情報を表示
  • XCLAIM- メッセージの所有権を移転
  • XINFO - ストリームとコンシューマグループの関連情報を表示。
  • XINFO GROUPS- コンシューマグループの情報を出力。
  • XINFO STREAM- ストリーム情報を出力

XADD

XADD を使用してキューにメッセージを追加します。指定したキューが存在しない場合はキューを作成します。XADD の構文形式：

XADD key ID field value [field value ...]
  • key：キュー名。存在しない場合は作成
  • ID：メッセージ ID。* を使用すると Redis が生成します。カスタムも可能ですが、単調増加を自分で保証する必要があります。
  • field value： レコード。

例

redis> XADD mystream * name Sara surname OConnor
"1601372323627-0"
redis> XADD mystream * field1 value1 field2 value2 field3 value3
"1601372323627-1"
redis> XLEN mystream
(integer) 2
redis> XRANGE mystream - +
1) 1) "1601372323627-0"
   2) 1) "name"
      2) "Sara"
      3) "surname"
      4) "OConnor"
2) 1) "1601372323627-1"
   2) 1) "field1"
      2) "value1"
      3) "field2"
      4) "value2"
      5) "field3"
      6) "value3"
redis>

XTRIM

XTRIM を使用してストリームをトリミングし、長さを制限します。構文形式：

XTRIM key MAXLEN [~] count
  • key：キュー名
  • MAXLEN：長さ
  • count：数量

例

127.0.0.1:6379> XADD mystream * field1 A field2 B field3 C field4 D
"1601372434568-0"
127.0.0.1:6379> XTRIM mystream MAXLEN 2
(integer) 0
127.0.0.1:6379> XRANGE mystream - +
1) 1) "1601372434568-0"
   2) 1) "field1"
      2) "A"
      3) "field2"
      4) "B"
      5) "field3"
      6) "C"
      7) "field4"
      8) "D"
127.0.0.1:6379>

redis>

XDEL

XDEL を使用してメッセージを削除します。構文形式：

XDEL key ID [ID ...]
  • key：キュー名
  • ID：メッセージ ID

例

> XADD mystream * a 1
1538561698944-0
> XADD mystream * b 2
1538561700640-0
> XADD mystream * c 3
1538561701744-0
> XDEL mystream 1538561700640-0
(integer) 1
127.0.0.1:6379> XRANGE mystream - +
1) 1) 1538561698944-0
   2) 1) "a"
      2) "1"
2) 1) 1538561701744-0
   2) 1) "c"
      2) "3"

XLEN

XLEN を使用してストリームに含まれる要素数、つまりメッセージ長を取得します。構文形式：

XLEN key
  • key：キュー名

例

redis> XADD mystream * item 1
"1601372563177-0"
redis> XADD mystream * item 2
"1601372563178-0"
redis> XADD mystream * item 3
"1601372563178-1"
redis> XLEN mystream
(integer) 3
redis>

XRANGE

XRANGE を使用してメッセージリストを取得します。削除済みメッセージは自動的にフィルタリングされます。構文形式：

XRANGE key start end [COUNT count]
  • key：キュー名
  • start：開始値、-最小値を表す
  • end：終了値、+最大値を表す
  • count：数量

例

redis> XADD writers * name Virginia surname Woolf
"1601372577811-0"
redis> XADD writers * name Jane surname Austen
"1601372577811-1"
redis> XADD writers * name Toni surname Morrison
"1601372577811-2"
redis> XADD writers * name Agatha surname Christie
"1601372577812-0"
redis> XADD writers * name Ngozi surname Adichie
"1601372577812-1"
redis> XLEN writers
(integer) 5
redis> XRANGE writers - + COUNT 2
1) 1) "1601372577811-0"
   2) 1) "name"
      2) "Virginia"
      3) "surname"
      4) "Woolf"
2) 1) "1601372577811-1"
   2) 1) "name"
      2) "Jane"
      3) "surname"
      4) "Austen"
redis>

XREVRANGE

XREVRANGE を使用してメッセージリストを取得します。削除済みメッセージは自動的にフィルタリングされます。構文形式：

XREVRANGE key end start [COUNT count]
  • key：キュー名
  • end：終了値、+最大値を表す
  • start：開始値、-最小値を表す
  • count：数量

例

redis> XADD writers * name Virginia surname Woolf
"1601372731458-0"
redis> XADD writers * name Jane surname Austen
"1601372731459-0"
redis> XADD writers * name Toni surname Morrison
"1601372731459-1"
redis> XADD writers * name Agatha surname Christie
"1601372731459-2"
redis> XADD writers * name Ngozi surname Adichie
"1601372731459-3"
redis> XLEN writers
(integer) 5
redis> XREVRANGE writers + - COUNT 1
1) 1) "1601372731459-3"
   2) 1) "name"
      2) "Ngozi"
      3) "surname"
      4) "Adichie"
redis>

XREAD

XREAD を使用して、ブロッキングまたは非ブロッキング方式でメッセージリストを取得します。構文形式：

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]
  • count：数量
  • milliseconds：オプション。ブロックするミリ秒数。設定がない場合は非ブロッキングモード
  • key：キュー名
  • id：メッセージ ID

例

# Stream の先頭から2つのメッセージを読み取る
> XREAD COUNT 2 STREAMS mystream writers 0-0 0-0
1) 1) "mystream"
   2) 1) 1) 1526984818136-0
         2) 1) "duration"
            2) "1532"
            3) "event-id"
            4) "5"
            5) "user-id"
            6) "7782813"
      2) 1) 1526999352406-0
         2) 1) "duration"
            2) "812"
            3) "event-id"
            4) "9"
            5) "user-id"
            6) "388234"
2) 1) "writers"
   2) 1) 1) 1526985676425-0
         2) 1) "name"
            2) "Virginia"
            3) "surname"
            4) "Woolf"
      2) 1) 1526985685298-0
         2) 1) "name"
            2) "Jane"
            3) "surname"
            4) "Austen"

XGROUP CREATE

XGROUP CREATE を使用してコンシューマグループを作成します。構文形式：

XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]
  • key：キュー名。存在しない場合は作成
  • groupname：グループ名。
  • $： 末尾から消費を開始することを示します。新しいメッセージのみを受け入れ、現在の Stream のメッセージはすべて無視されます。

先頭から消費開始:

XGROUP CREATE mystream consumer-group-name 0-0

末尾から消費開始:

XGROUP CREATE mystream consumer-group-name $

XREADGROUP GROUP

XREADGROUP GROUP を使用してコンシューマグループ内のメッセージを読み取ります。構文形式：

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
  • group：コンシューマグループ名
  • consumer：コンシューマ名。
  • count： 読み取り数。
  • milliseconds： ブロックするミリ秒数。
  • key： キュー名。
  • ID： メッセージ ID。
XREADGROUP GROUP consumer-group-name consumer-name COUNT 1 STREAMS mystream >
その他の拡張