アプリボットPublicationへの投稿
😄

Redis Streamsをローカルで動かしてみた

のでぃ
に公開
Redis
stream
tech

この記事は Applibot Advent Calendar 2025 3日目の記事です。
前日の記事は、@siguma_sigさんのコーディングエージェントを使う上で意識したほうがいいLLMの特性です

本記事では、go-redisを用いた実装を通して、Redis Streamsの特性を理解していきます。

はじめに

みなさん、リアルタイム通信によるメッセージング機能を用いたサービスを実装したことはあるでしょうか？
例えば、誰かがメッセージを送信すると他の全てのユーザの元に届く、といったチャットサービスが挙げられます。これらの処理は、RedisのPub/Subを用いることで比較的容易に実装が可能です。

一方で、RedisのPub/Subには到達保証がありません。
そのため、確実に通知が届いていることを担保する必要があるようなサービスではそのまま利用することは難しいです。

自分の実装でも、シンプルな情報の送受信はスムーズに実装できましたが、再処理をするための周辺実装について考慮するべき点が多く苦労しました。
その中で、Redis Streamsが良い解決手段になる、との情報を目にしたため、実際に動かしてみることにしました。

Redis Streamsでできること

そもそもRedis Streamsでは何が実現できるのか、について整理します。

Redis Streams

https://redis.io/docs/latest/develop/data-types/streams/

Redis Streamsは、時系列にデータを格納する追記型のデータ構造です。
格納先のデータ構造はStream、格納するデータはEntryと呼ばれます。
また、通知を受信する対象はConsumerと呼ばれます。
同一のストリームを参照しているConsumerは、Consumer Groupに振り分けることで、異なるEntryを出し分けることができます。

コマンド一覧（抜粋）

  • XADD
    • EntryをStreamへ追加する
    • Pub/SubのPUBLISH相当
  • XREAD （グループ化している場合は、XREADGROUP）
    • StreamにあるEntryを読み取る
      • Optionでblock=0を指定することで、Entryが来るまで待機状態になる
    • Pub/SubのSUBSCRIBE相当
  • XACK
    • ConsumerからStreamへ、処理完了を明示的に通知する
  • XPENDING
    • 保留中のEntryがあれば取得する

これらを用いることで、リアルタイム通信による送受信を実現することが可能になります

Pub/Subと異なり、送受信にStreamを経由させることで、送受信データの永続化および状態管理を独立して実装できます。

Goで実装してみた

実際に、Redis Streamsでリアルタイム通信を行うアプリを実装してみます。
ライブラリはgo-redisを使用しています。
実践したコードは以下の通りです（文字列出力などは省略）

Producer

producer.go
func main () {
    // Redis接続
    rdb, err := client.NewClient(config.RedisAddr)
    if err != nil {
        panic(err)
    }
    defer rdb.Close()
    
    scanner := bufio.NewScanner(os.Stdin)
    for {
        content := strings.TrimSpace(scanner.Text())
    
        // XADD: メッセージを送信
        msgID, err := rdb.XAdd(ctx, &redis.XAddArgs{
            Stream: config.StreamName,
            Values: map[string]interface{}{
                "content":   content,
                "timestamp": time.Now().Unix(),
            },
        }).Result()
        if err != nil {
            fmt.Printf("Failed to send: %v\n", err)
        } else {
            fmt.Printf("Sent [%s] %s\n", msgID, content)
        }
    }
}

Consumer

再処理の観点で、今回は1つのConsumer Group内に1つのConsumerを配置し、新規メッセージはXREADGROUPで>を指定して、PENDINGメッセージはXREADGROUPで0を指定して定期的に再取得する実装としています。

consumer.go
func main() {
    consumerName := os.Args[1]
    ctx := context.Background()
    
    // Redis接続
    rdb, err := client.NewClient(config.RedisAddr)
    if err != nil {
        panic(err)
    }
    defer rdb.Close()
    
    // 各Consumerごとに異なるGroup名を使用（全員にメッセージが届く）
    groupName := fmt.Sprintf("group-%s", consumerName)
    
    // Consumer Group作成
    if err := client.CreateGroup(ctx, rdb, config.StreamName, groupName); err != nil {
        panic(err)
    }
    
    // PENDING再処理用のgoroutineを起動
    go checkPendingMessages(ctx, rdb, consumerName, groupName)
    
    // 新規メッセージ処理ループ
    for {
        processMessages(ctx, rdb, consumerName, groupName)
    }
}

func processMessages(ctx context.Context, rdb *redis.Client, consumerName, groupName string) {
	// XREADGROUP: 独自グループからメッセージを読み取る
	streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group:    groupName,
		Consumer: consumerName,
		Streams:  []string{config.StreamName, ">"}, // ">" = 未配達メッセージ
		Count:    1,
		Block:    2 * time.Second,
	}).Result()

	if err != nil {
		fmt.Printf("Read error: %v\n", err)
		return
	}

	// メッセージ処理
	for _, stream := range streams {
		for _, msg := range stream.Messages {
			handleMessage(ctx, rdb, consumerName, groupName, msg, false)
		}
	}
}

func handleMessage(ctx context.Context, rdb *redis.Client, consumerName, groupName string, msg redis.XMessage, isRetry bool) {
	content := msg.Values["content"].(string)

	// "FAIL"を含むメッセージは初回のみ処理失敗させる（再処理時は成功）
	if strings.Contains(content, "FAIL") && !isRetry {
		// XACKしない = PENDINGリストに残る
		return
	}

	// XACK: 処理完了を通知
	if err := rdb.XAck(ctx, config.StreamName, groupName, msg.ID).Err(); err != nil {
		return
	}
}

// 定期的に自分のPENDINGメッセージをチェックして再処理
func checkPendingMessages(ctx context.Context, rdb *redis.Client, consumerName, groupName string) {
	ticker := time.NewTicker(config.CheckInterval)
	defer ticker.Stop()

	for range ticker.C {
		// 自分のPENDINGメッセージを取得（"0"を指定）
		streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
			Group:    groupName,
			Consumer: consumerName,
			Streams:  []string{config.StreamName, "0"}, // "0" = 自分のPENDINGメッセージ
			Count:    10,
		}).Result()
		if err != nil {
			continue
		}

		if len(streams) == 0 || len(streams[0].Messages) == 0 {
			continue
		}

		// PENDING メッセージを再処理
		for _, msg := range streams[0].Messages {
			handleMessage(ctx, rdb, consumerName, groupName, msg, true)
		}
	}
}

出力例

producer
redis-streams-demo % go run producer.go
=== Producer Started ===
Stream: messages
Type message and press Enter to send

> こんにちは
Sent [1764655388322-0] こんにちは
> FAIL
Sent [1764655394997-0] FAIL
> さようなら
Sent [1764655406500-0] さようなら
consumer1
redis-streams-demo % go run consumer.go consumer-1
=== Consumer 'consumer-1' Started ===
Stream: messages, Group: group-consumer-1
PENDING check interval: 5s

[consumer-1] Received: こんにちは (ID: 1764655388322-0...)
[consumer-1] Processed successfully and ACKed

[consumer-1] Received: FAIL (ID: 1764655394997-0...)
[consumer-1] Processing FAILED (contains 'FAIL')
[consumer-1] Message not ACKed, will remain PENDING

[consumer-1] Found 1 PENDING messages, retrying...
[consumer-1] RETRY: FAIL (ID: 1764655394997-0...)
[consumer-1] Processed successfully and ACKed

[consumer-1] Received: さようなら (ID: 1764655406500-0...)
[consumer-1] Processed successfully and ACKed

PENDINGの間隔を5秒で設定しました。そのため、Consumer側が受け取り時の処理に失敗した場合でも、5秒以内に再処理を実行します。

（補足）同一のConsumer Groupの活用パターン

上記の例は、2つのConsumer Groupに登録した状態で行いました。
反対に、Consumer-1, 2をそれぞれ同一のConsumer Groupに登録することも可能です。
これは、同一Streamから負荷分散のために複数のConsumerを利用するケースで活用できます。
以下に、そのパターンの出力例も記載します。

producer
redis-streams-demo % go run producer.go
=== Producer Started ===
Stream: messages
Type message and press Enter to send

> こんにちは
Sent [1764654897192-0] こんにちは
> ハロー
Sent [1764654906331-0] ハロー
> やあ
Sent [1764654917623-0] やあ
consumer1
redis-streams-demo % go run consumer.go consumer-1
=== Consumer 'consumer-1' Started ===
Group: processing-group, Stream: messages
PENDING check interval: 5s

[consumer-1] Processed successfully and ACKed

[consumer-1] Received: ハロー (ID: 1764654906331-0...)
[consumer-1] Found 1 PENDING messages, retrying...
[consumer-1] RETRY: ハロー (ID: 1764654906331-0...)
[consumer-1] Processed successfully and ACKed

[consumer-1] Processed successfully and ACKed
consumer2
redis-streams-demo % go run consumer.go consumer-2
Consumer Group 'processing-group' ready

=== Consumer 'consumer-2' Started ===
Group: processing-group, Stream: messages
PENDING check interval: 5s

[consumer-2] Received: こんにちは (ID: 1764654897192-0...)
[consumer-2] Found 1 PENDING messages, retrying...
[consumer-2] RETRY: こんにちは (ID: 1764654897192-0...)
[consumer-2] Processed successfully and ACKed

[consumer-2] Processed successfully and ACKed

[consumer-2] Received: やあ (ID: 1764654917623-0...)
[consumer-2] Processed successfully and ACKed

Consumer1と2それぞれに、メッセージが振り分けられていることが確認できました。
振り分けはラウンドロビン方式により実施されています。

終わりに

実際に実装してみることで、両者の挙動の違いをより明確にすることができました。
Redis StreamsではRedis Pub/Subと比較して、Streams周辺の機能のみで再処理についての課題を解決可能でした。

日頃から選択肢を増やすことを意識しながら、今後も様々な技術にアンテナを伸ばしていければと考えています。

明日は@fss_bassさんです。

アプリボット
アプリボットPublication

株式会社アプリボット（applibot.co.jp/）のテックブログです！ 採用情報はこちら > applibot.co.jp/recruit/ 過去の技術ブログはこちら > blog.applibot.co.jp/

Discussion