クライアントサイドキャッシュ(Redis)によるリアクティビティの向上

Redisとは何か

Redisをご存じない方はあまりいないと思います。しかし、いくつかの特性について簡単に触れてから先へ進むとすれば、次のように整理できるでしょう。

• シングルスレッドで演算が実行されるため、すべての演算が原子性を持ちます。
• In-Memoryでデータが保存され演算されるため、すべての演算が高速です。
• Redisはオプションに応じてWALを保存することができ、最新の状態を迅速にバックアップし、復旧することができます。
• Set、Hash、Bit、Listなどの様々な型をサポートしており、高い生産性を持ちます。
• 大規模なコミュニティを持っているため、多様な経験、課題、解決方法を共有してもらうことができます。
• 長期間にわたり開発および運用されてきたため、信頼できる安定性があります。

では本題へ

想像してみてください

もし皆さんのサービスのキャッシュが次の二つの条件に当てはまるとしたらどうでしょうか。

1. 頻繁に照会されるデータを最新の状態でユーザーに提供しなければならないが、更新が不規則であるため、キャッシュの更新を頻繁に行う必要がある場合
2. 更新は行われないが、同一のキャッシュデータに頻繁にアクセスして照会しなければならない場合

最初のケースとしては、ショッピングモールのリアルタイム人気ランキングが考えられます。ショッピングモールのリアルタイム人気ランキングをsorted setで保存した場合、ユーザーがメインページにアクセスするたびにRedisから読み取るのは非効率です。
二番目のケースは為替データに関するものです。おおよそ10分周期で為替データが告示されるにもかかわらず、実際の照会は非常に頻繁に発生します。特にウォン-ドル、ウォン-円、ウォン-元については、非常に頻繁にキャッシュを照会することになります。
このようなケースでは、APIサーバーがローカルに別途キャッシュを保持しておき、データが変更された場合にRedisを再度照会して更新する方が効率的な動作となるでしょう。

では、データベース - Redis - APIサーバーの構造において、このような動作をどのように実装できるのでしょうか??

Redis PubSubでは実現できないか？

キャッシュを使用する際、更新の有無を受け取れるチャネルを購読しよう！

• そのためには、更新時にメッセージを送信するロジックを作成しなければなりません。
• PubSubによる追加の動作が加わるため、性能に影響を与えます。

では、Redisが変更を検知するとしたら？

Keyspace Notificationを使用して、該当キーに対するコマンドの通知を受け取るとしたら？

• 更新に使用されるキーとコマンドを事前に保存し、共有しなければならないという煩わしさがあります。
• 例えば、あるキーについては単純なSetが更新コマンドであり、あるキーではLPush、あるいはRPushやSAddおよびSRemが更新コマンドになるなど、複雑になります。
• これは開発過程でのコミュニケーションミスやコーディング上のヒューマンエラーを発生させる可能性を大幅に高めます。

Keyevent Notificationを使用して、コマンド単位で通知を受け取るとしたら？

• 更新に使用されるすべてのコマンドに対する購読が必要です。そこから入ってくるキーについて適切なフィルタリングが必要になります。
• 例えば、Delで入ってくるすべてのキーのうち一部については、そのクライアントがローカルキャッシュを保持していない可能性が高いです。
• これは不要なリソースの浪費につながる可能性があります。

そこで必要になるのがInvalidation Messageです！

Invalidation Messageとは何か？

Invalidation MessagesはRedis 6.0から追加されたServer Assisted Client-Side Cacheの一環として提供される概念です。Invalidation Messageは次の流れで伝達されます。

1. ClientBがすでにkeyを一度読み取ったと仮定します。
2. ClientAが該当keyを新たに設定します。
3. Redisは変更を検知し、ClientBにInvalidation Messageを発行して、ClientBにキャッシュを削除するよう通知します。
4. ClientBは該当メッセージを受け取り、適切な措置を講じます。

どのように使うのか

基本動作の構造

Redisに接続されたクライアントがCLIENT TRACKING ON REDIRECT <client-id>を実行することで、invalidation messageを受け取るようにします。そして、メッセージを受け取る必要のあるクライアントはSUBSCRIBE __redis__:invalidateを実行して、invalidation messageを購読します。

default tracking

1# client 1
2> SET a 100

1# client 3
2> CLIENT ID
312
4> SUBSCRIBE __redis__:invalidate
51) "subscribe"
62) "__redis__:invalidate"
73) (integer) 1

1# client 2
2> CLIENT TRACKING ON REDIRECT 12
3> GET a # tracking

1# client 1
2> SET a 200

1# client 3
21) "message"
32) "__redis__:invalidate"
43) 1) "a"

broadcasting tracking

1# client 3
2> CLIENT ID
312
4> SUBSCRIBE __redis__:invalidate
51) "subscribe"
62) "__redis__:invalidate"
73) (integer) 1

1# client 2
2CLIENT TRACKING ON BCAST PREFIX cache: REDIRECT 12

1# client 1
2> SET cache:name "Alice"
3> SET cache:age 26

1# client 3
21) "message"
32) "__redis__:invalidate"
43) 1) "cache:name"
51) "message"
62) "__redis__:invalidate"
73) 1) "cache:age"

実装！実装！実装！

Redigo + Ristretto

このように説明するだけでは、実際にコード上でどのように使えばよいか曖昧です。そこで、簡単にredigoとristrettoで先に構成してみましょう。

まず二つの依存関係をインストールします。

• github.com/gomodule/redigo
• github.com/dgraph-io/ristretto

 1package main
 2
 3import (
 4	"context"
 5	"fmt"
 6	"log/slog"
 7	"time"
 8
 9	"github.com/dgraph-io/ristretto"
10	"github.com/gomodule/redigo/redis"
11)
12
13type RedisClient struct {
14	conn  redis.Conn
15	cache *ristretto.Cache[string, any]
16	addr  string
17}
18
19func NewRedisClient(addr string) (*RedisClient, error) {
20	cache, err := ristretto.NewCache(&ristretto.Config[string, any]{
21		NumCounters: 1e7,     // キーの頻度を追跡する数 (10M)。
22		MaxCost:     1 << 30, // キャッシュの最大コスト (1GB)。
23		BufferItems: 64,      // Getバッファごとのキー数。
24	})
25	if err != nil {
26		return nil, fmt.Errorf("failed to generate cache: %w", err)
27	}
28
29	conn, err := redis.Dial("tcp", addr)
30	if err != nil {
31		return nil, fmt.Errorf("failed to connect to redis: %w", err)
32	}
33
34	return &RedisClient{
35		conn:  conn,
36		cache: cache,
37		addr:  addr,
38	}, nil
39}
40
41func (r *RedisClient) Close() error {
42	err := r.conn.Close()
43	if err != nil {
44		return fmt.Errorf("failed to close redis connection: %w", err)
45	}
46
47	return nil
48}

まず簡単に、ristrettoとredigoを含むRedisClientを作成します。

 1func (r *RedisClient) Tracking(ctx context.Context) error {
 2	psc, err := redis.Dial("tcp", r.addr)
 3	if err != nil {
 4		return fmt.Errorf("failed to connect to redis: %w", err)
 5	}
 6
 7	clientId, err := redis.Int64(psc.Do("CLIENT", "ID"))
 8	if err != nil {
 9		return fmt.Errorf("failed to get client id: %w", err)
10	}
11	slog.Info("client id", "id", clientId)
12
13	subscriptionResult, err := redis.String(r.conn.Do("CLIENT", "TRACKING", "ON", "REDIRECT", clientId))
14	if err != nil {
15		return fmt.Errorf("failed to enable tracking: %w", err)
16	}
17	slog.Info("subscription result", "result", subscriptionResult)
18
19	if err := psc.Send("SUBSCRIBE", "__redis__:invalidate"); err != nil {
20		return fmt.Errorf("failed to subscribe: %w", err)
21	}
22	psc.Flush()
23
24	for {
25		msg, err := psc.Receive()
26		if err != nil {
27			return fmt.Errorf("failed to receive message: %w", err)
28		}
29
30		switch msg := msg.(type) {
31		case redis.Message:
32			slog.Info("received message", "channel", msg.Channel, "data", msg.Data)
33			key := string(msg.Data)
34			r.cache.Del(key)
35		case redis.Subscription:
36			slog.Info("subscription", "kind", msg.Kind, "channel", msg.Channel, "count", msg.Count)
37		case error:
38			return fmt.Errorf("error: %w", msg)
39		case []interface{}:
40			if len(msg) != 3 || string(msg[0].([]byte)) != "message" || string(msg[1].([]byte)) != "__redis__:invalidate" {
41				slog.Warn("unexpected message", "message", msg)
42				continue
43			}
44
45			contents := msg[2].([]interface{})
46			keys := make([]string, len(contents))
47			for i, key := range contents {
48				keys[i] = string(key.([]byte))
49				r.cache.Del(keys[i])
50			}
51			slog.Info("received invalidation message", "keys", keys)
52		default:
53			slog.Warn("unexpected message", "type", fmt.Sprintf("%T", msg))
54		}
55	}
56}

コードが少し複雑です。

• Trackingを行うためにコネクションをもう一つ確立します。これはPubSubが他の動作の妨げになることを考慮した措置です。
• 追加したコネクションのIDを照会し、データを照会するコネクションでTrackingをそのコネクションへリダイレクトするようにします。
• そして、invalidation messageを購読します。
• 購読を処理するコードが少し複雑です。redigoがinvalidation messageのパースに対応していないため、パース前の応答を受け取って処理しなければなりません。

 1func (r *RedisClient) Get(key string) (any, error) {
 2	val, found := r.cache.Get(key)
 3	if found {
 4		switch v := val.(type) {
 5		case int64:
 6			slog.Info("cache hit", "key", key)
 7			return v, nil
 8		default:
 9			slog.Warn("unexpected type", "type", fmt.Sprintf("%T", v))
10		}
11	}
12	slog.Info("cache miss", "key", key)
13
14	val, err := redis.Int64(r.conn.Do("GET", key))
15	if err != nil {
16		return nil, fmt.Errorf("failed to get key: %w", err)
17	}
18
19	r.cache.SetWithTTL(key, val, 1, 10*time.Second)
20	return val, nil
21}

Getメソッドは次のように、まずristrettoを照会し、なければRedisから取得するようにします。

 1package main
 2
 3import (
 4	"context"
 5	"log/slog"
 6	"os"
 7	"os/signal"
 8	"time"
 9)
10
11func main() {
12	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
13	defer cancel()
14
15	client, err := NewRedisClient("localhost:6379")
16	if err != nil {
17		panic(err)
18	}
19	defer client.Close()
20
21	go func() {
22		if err := client.Tracking(ctx); err != nil {
23			slog.Error("failed to track invalidation message", "error", err)
24		}
25	}()
26
27	ticker := time.NewTicker(1 * time.Second)
28	defer ticker.Stop()
29	done := ctx.Done()
30
31	for {
32		select {
33		case <-done:
34			slog.Info("shutting down")
35			return
36		case <-ticker.C:
37			v, err := client.Get("key")
38			if err != nil {
39				slog.Error("failed to get key", "error", err)
40				return
41			}
42			slog.Info("got key", "value", v)
43		}
44	}
45}

テストのためのコードは上記のとおりです。一度テストしてみると、Redisでデータが更新されるたびに新しい値へ更新される様子を確認できるでしょう。

しかし、これは非常に複雑です。何よりも、クラスタに対して拡張するためには、必然的にすべてのマスターまたはレプリカに対してTrackingを有効化する必要があります。

Rueidis

Go言語を使う以上、私たちには最も現代的で発展したrueidisがあります。rueidisを用いたRedisクラスタ環境でのserver assisted client side cacheを使用するコードを書いてみましょう。

まず、依存関係をインストールします。

• github.com/redis/rueidis

次に、Redisからデータを照会するコードを書きます。

 1package main
 2
 3import (
 4	"context"
 5	"log/slog"
 6	"os"
 7	"os/signal"
 8	"time"
 9
10	"github.com/redis/rueidis"
11)
12
13func main() {
14	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
15	defer cancel()
16
17	client, err := rueidis.NewClient(rueidis.ClientOption{
18		InitAddress: []string{"localhost:6379"},
19	})
20	if err != nil {
21		panic(err)
22	}
23
24	ticker := time.NewTicker(1 * time.Second)
25	defer ticker.Stop()
26	done := ctx.Done()
27
28	for {
29		select {
30		case <-done:
31			slog.Info("shutting down")
32			return
33		case <-ticker.C:
34			const key = "key"
35			resp := client.DoCache(ctx, client.B().Get().Key(key).Cache(), 10*time.Second)
36			if resp.Error() != nil {
37				slog.Error("failed to get key", "error", resp.Error())
38				continue
39			}
40			i, err := resp.AsInt64()
41			if err != nil {
42				slog.Error("failed to convert response to int64", "error", err)
43				continue
44			}
45			switch resp.IsCacheHit() {
46			case true:
47				slog.Info("cache hit", "key", key)
48			case false:
49				slog.Info("missed key", "key", key)
50			}
51			slog.Info("got key", "value", i)
52		}
53	}
54}

rueidisでclient side cacheを使用するには、単にDoCacheを呼び出すだけで十分です。すると、ローカルキャッシュにどれだけ保持するかといった設定とともにローカルキャッシュへ追加され、同じくDoCacheを呼び出せば、ローカルキャッシュ内からデータが取得されます。当然ながら、無効化メッセージも正しく処理されます。

なぜredis-goではないのか？

redis-goは残念ながら、公式APIとしてserver assisted client side cacheをサポートしていません。さらに、PubSubを生成する際に新しいコネクションを作成しながら、そのコネクションに直接アクセスするAPIがないため、client idを知ることすらできません。そのため、redis-goでは構成自体が不可能と判断し、見送りました。

セクシーになりましたね

client side cache構造を通じて

• 事前に準備できるデータであれば、この構造によってRedisへのクエリおよびトラフィックを最小化し、常に最新のデータを提供できるでしょう。
• これにより、一種のCQRS構造を作ることができ、読み取り性能を飛躍的に向上させられます。

どれほどセクシーになったのか？

実際に現場では、ちょうどこのような構造で運用しているため、二つのAPIについて簡単なレイテンシを調べてみました。非常に抽象的な形でしか示せないことをご容赦ください。

1. 一つ目のAPI
   1. 初回照会時: 平均14.63ms
   2. 以降の照会時: 平均2.82ms
   3. 平均差: 10.98ms
2. 二つ目のAPI
   1. 初回照会時: 平均14.05ms
   2. 以降の照会時: 平均1.60ms
   3. 平均差: 11.57ms

最大で約82%の追加のレイテンシ改善がありました！

おそらく次のような改善点があったものと期待しています。

• クライアントとRedis間のネットワーク通信過程の省略およびトラフィックの節約
• Redis自体が実行すべき読み取りコマンド数の減少
  • これは書き込み性能の向上にも効果があります。
• Redisプロトコルのパース最小化
  • Redisプロトコルをパースすることにもコストはゼロではありません。これを減らせるのは大きな機会です。

しかし、すべてはトレードオフです。このために、私たちは少なくとも以下の二つを犠牲にしました。

• クライアントサイドキャッシュ管理要素の実装、運用および保守が必要であること
• これによるクライアントのCPUおよびメモリ使用量の増加

結論

個人的には満足のいくアーキテクチャの構成要素であり、レイテンシおよびAPIサーバーへのストレスも非常に少なかったです。今後も可能であれば、このような構造でアーキテクチャを構成できればと考えています。