はじめに

Redis Streamsを勉強する機会があり、
その動きの流れを見ていって、
よくあるアーティストやイベントとかのチケット一般販売（先着購入順）の
システムっぽい作りが出来そうと思ったので
勉強として簡単な流れのものを作ってみました。

Redis streamsとは

まず、Redis streamsはRedis 5.0で導入されたデータ型で、
以下のような特徴があります。

・データを時系列で持つ追記型のデータ構造
・各データは任意の複数のフィールドを持つことができる
・時間の範囲を指定して複数のデータを読むことが出来る
・データを取得しても過去のデータを残せる（揮発しない）

＜Redis公式＞
https://redis.io/docs/latest/develop/data-types/streams/

ここから先は、Redis Streamsの機能を、
チケットを購入する流れと交えながら説明していきます。

ざっくりシステム概要

まず、チケット購入のシステムとしてやりたいことは以下の通りです。

・販売するチケットは３枚
・チケット番号はno-001, no-002, no-003
・先着順で1人１枚ずつ購入可能
・3枚売れたらそれ以降は購入不可
・購入画面、決済画面、完了画面で表現する

今回はRedis7.2.5でやっていきます。
最初にRedis Streamsの準備をします。

XGROUP CREATE

まずはチケット管理をするStreamsを作ります。
コマンドは「XGROUP CREATE」を実行します。
このコマンドは最初の1回だけ実行します。
正常に作成されると「OK」が返ってきます。

以下の例はストリーム名を「ticket_stream」、グループ名を「mygroup」とし、
ストリーム全体を取得対象として作成した場合です。

1【コマンド】
2XGROUP CREATE ticket_stream mygroup 0
3【結果】
4OK

XREAD

XREADコマンドでstreamのデータの情報が見れます。
まだ何も入れていないので結果は(nil)が返ってきます。

1【コマンド】
2XREAD STREAMS ticket_stream 0
3【結果】
4(nil)

XADD

次にチケットの情報を３枚分投入します。
こちらはXADDコマンドで入れていきます。
正常に入れることができると、結果としてメッセージのIDが返却されます。

まずは1件分入れてみます。

1【コマンド】
2XADD ticket_stream * ticket_id 10001 ticket_value no-001
3【結果】
4"1730825420059-0"

「ticket_stream」にメッセージID="1730825420059-0"、
それのValueとして
{ticket_id=10001、ticket_value=no-001}
という配列が入ったものが設定されます。

メッセージIDは「*」で指定した場合、Unix時間（ミリ秒）で自動的に採番されます。
「-」より右側の数字は、ミリ秒単位で同じタイミングに作られた場合、
「0」、「1」とカウントアップして作られます。

「*」ではなく任意の値も設定できます。
ただし、一度設定した値より小さい数値には設定できません。

ここでもう一度XREADでストリームの中身見てみます。

1【コマンド】
2XREAD STREAMS ticket_stream 0
3【結果】
4127.0.0.1:6379> XREAD STREAMS ticket_stream 0
51) 1) "ticket_stream"
6   2) 1) 1) "1730825420059-0"
7         2) 1) "ticket_id"
8            2) "10001"
9            3) "ticket_value"
10            4) "no-001"

ticket_streamの中に"1730825420059-0"が出来ており、その中に設定した値が入っているのが分かります。
同じような形であと２件入れた結果は以下の通りです。

1127.0.0.1:6379> XADD ticket_stream * ticket_id 10002 ticket_value no-002
2"1730825771052-0"
3127.0.0.1:6379> XADD ticket_stream * ticket_id 10003 ticket_value no-003
4"1730825779936-0"
5127.0.0.1:6379> XREAD STREAMS ticket_stream 0
61) 1) "ticket_stream"
7   2) 1) 1) "1730825420059-0"
8         2) 1) "ticket_id"
9            2) "10001"
10            3) "ticket_value"
11            4) "no-001"
12      2) 1) "1730825771052-0"
13         2) 1) "ticket_id"
14            2) "10002"
15            3) "ticket_value"
16            4) "no-002"
17      3) 1) "1730825779936-0"
18         2) 1) "ticket_id"
19            2) "10003"
20            3) "ticket_value"
21            4) "no-003"

購入画面と購入の処理

購入画面の画像

チケット購入の画面をこんな感じで用意しました。
購入ボタンを押せばチケットを買えるような感じです。

この購入ボタンを押した時に
チケット情報を取得する時には以下のコマンドを実行します。

1# 【コマンド】
2XREADGROUP GROUP mygroup userA COUNT 1 STREAMS ticket_stream >

まだ誰も取得をしていないものを、
IDの若い順から取得します。
今回は１枚ずつ取っていきたいので「count」で制限してますが、
countを指定しなければ未取得のもの全てを取得することも出来ます。

それではまず、Aさんが購入ボタンを1回押した場合です。
画面の処理をするためにRuby on Railsでのメソッドで記載します。

Aさん購入後の画像

1【Ruby on Railsでのメソッド】
2messages = redis.xreadgroup("mygroup", userA, "ticket_stream", ">", count: 1)
3【messageの中身】
4{"ticket_stream"=>[["1730825420059-0", {"ticket_id"=>"10001", "ticket_value"=>"no-001"}]]}

ちなみに取得後もう一度同じコマンドを実行すると、
次のものが取得出来ます。
この場合、「no-002」の情報が取れる感じです。
取得できるものが無くなった場合、空が返却されます。
同じものは２度取得は出来ません。

＜Aさんの次にBさんが購入ボタンを押した場合＞

1【Ruby on Railsでのメソッド】
2messages = redis.xreadgroup("mygroup", userB, "ticket_stream", ">", count: 1)
3【messageの中身】
4{"ticket_stream"=>[["1730825771052-0", {"ticket_id"=>"10002", "ticket_value"=>"no-001"}]]}

Pendingリスト

取得したものをもう一度確認するにはどうすれば良いのか？
そういう場合は以下のコマンドを実行します。

1# 【コマンド】
2XREADGROUP GROUP mygroup userA STREAMS ticket_stream 0
3【Ruby on Railsでのメソッド】
4 messages = redis.xreadgroup("mygroup", "userA", "ticket_stream", "0")
5【messageの中身】
6{"ticket_stream"=>[["1730825420059-0", {"ticket_id"=>"10001", "ticket_value"=>"A001"}]]}

先ほどのコマンドと見比べると「>」の代わりに「0」を指定している感じです。

Redis側にPendingリストというものがあり、
一度取得したものは基本的に処理がPending(保留)扱いとなり、
そちらのリストに格納されます。
こちらは保留が解消されない限り、残り続けます。
（XREADGROUPのオプションで保留にしないことも出来ます）

Pendingの情報に関しては以下のコマンドで確認できます。

1【コマンド】
2XPENDING ticket_stream mygroup START END COUNT
3【Ruby on Railsでのメソッド】
4puts redis.xpending("ticket_stream", "mygroup", "-", "+", 100, "userA")
5
6【実行結果】（まだ何もしていない時の状態）
7127.0.0.1:6379> XPENDING ticket_stream mygroup
81) (integer) 0
92) (nil)
103) (nil)
114) (nil)

1【実行結果】（Aさんが購入ボタンを押した後の状態）
2127.0.0.1:6379> XPENDING ticket_stream mygroup
31) 1730825420059-0
42) userA
53) (integer) 869132
64) (integer) 1
7
8【Ruby on Railsでの出力結果】
9{"entry_id"=>"1730825420059-0", "consumer"=>"userA", "elapsed"=>869132, "count"=>1}

Pending情報が入ると上記のように表示されます。

1. メッセージID
2. コンシューマー名 （Aさん＝userA）
3. メッセージがコンシューマに配信されてから経過したミリ秒数
4. このメッセージが配信された回数

例えば購入ボタンを押して、決済画面に行った時、
このPending情報を使えば今処理しているものとして対応出来そうです。

また、決済中になんらかのエラーでもう一度決済画面に戻りたい、
といった場合にPending情報を利用すれば対応できそうです。

XACK

保留のものを完了にしてPendingリストから無くすためには
以下のコマンドを実行します。

1【コマンド】
2XACK ticket_stream mygroup "1730825420059-0"
3【Ruby on Railsでのメソッド】
4puts redis.xack("ticket_stream", "mygroup", "1730825420059-0")
5【実行結果】
6(integer) 1

ストリーム名、グループ、メッセージのIDを指定する感じです。
Pendingのものが正常に完了にした件数を
実行結果として返します。
これ以降、改めてPending情報を取得するコマンドを実行しても、
完了にしたIDのものは出なくなっているはずです。

チケット購入システムであれば、
決済画面から情報を入力して完了画面に遷移する時に対応する感じです。

流れとして

ここまでの流れでチケットの状況は

Aさん：購入完了
Bさん：決済手続き中
チケット：残り１枚

といった感じになるかなと。

この後にCさんが購入ボタンを押してから
Dさんが購入画面に来ても、すでに３枚購入（購入手続き）されているので
「予定枚数終了」という感じに。

Pending情報と実データの結びつき

ちなみにPending情報が残っているまま、
対象のIDの情報を削除するとどうなるでしょうか？
以下は、Pending中の１件をXDELで削除して、
改めてXREADGROUPでPending情報を指定した時の結果です。

1【削除コマンド】
2127.0.0.1:6379> XDEL ticket_stream 1730825420059-0
3【結果】
4(integer) 1
5
6【もう一度Pending情報取得】
7127.0.0.1:6379> XREADGROUP GROUP mygroup user1 STREAMS ticket_stream 0
81) 1) "ticket_stream"
9   2) 1) 1) "1730825420059-0"
10         2) (nil)

実際の値は削除されているのですが、
Pending情報のキーは残ったままで、
その値は(nil)といった形で出力されます。
こうなるとPending情報の状態がよくわからなくなってしまうので、
完了させてから削除して、
Pending情報に残さないようにしましょう。

まとめ

今回はRedis Streamsの動きを知るために最低限の処理で記載しました。
もし、システムとしてさらに作り込むとなると、
購入手続き中のエラー対応や、購入後のDB管理、
あとは同一人物の枚数制限方法や決済のセキュリティ対策など、
いろいろとやるべきことは出てくるかなと。

また、Redis Streams自体ももっといろんな使い方できると思うので、
使えるコマンドや制約を調べていって、
今後関わるシステムの対応の一つの選択肢として
使っていけたら良いなと思います。