--- id: 1c72ae6efee059c2c727ea53b56bf6c5 author: snowmerak title: Zvyšování odezvy pomocí client-side cache v Redisu description: Zlepšení odezvy API pomocí klientské mezipaměti podporované serverem Redis (Redis Server Assisted Client Side Cache) language: cs date: 2024-11-29T00:51:27.525566863Z path: /blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502 category: database --- ## Co je Redis? Předpokládám, že jen málokdo Redis nezná. Přesto, abych ho krátce připomněl prostřednictvím několika jeho vlastností, lze jej shrnout následovně. - Operace se provádějí v **jednom vlákně (single thread)**, a proto mají všechny operace **atomicitu**. - Data jsou uložena a zpracovávána **In-Memory**, a proto jsou všechny operace **rychlé**. - Redis může podle nastavení ukládat **WAL**, takže lze rychle **zálohovat** aktuální stav a **obnovit** jej. - Podporuje **různé typy** jako Set, Hash, Bit, List a další, což zajišťuje vysokou **produktivitu**. - Má **velkou komunitu**, takže lze **sdílet** a získávat různé zkušenosti, problémy a způsoby řešení. - Je **dlouhodobě vyvíjen a provozován**, což zajišťuje důvěryhodnou **stabilitu**. ## Nyní k věci ### Představte si? Co kdyby cache ve vaší službě splňovala následující dvě podmínky? 1. Je nutné uživateli poskytovat často dotazovaná data v aktuálním stavu, avšak jejich aktualizace probíhá nepravidelně, a proto je třeba cache obnovovat velmi často. 2. Data se neaktualizují, ale ke stejným cachovaným datům je nutné přistupovat a číst je velmi často. Prvním případem může být zohlednění přehledu aktuální oblíbenosti v internetovém obchodě. Pokud je přehled aktuální oblíbenosti v internetovém obchodě uložen jako sorted set, je neefektivní, aby jej Redis při každém přístupu uživatele na hlavní stránku znovu načítal. Druhým případem jsou data o směnných kurzech. I když jsou kurzy zveřejňovány zhruba v desetiminutových intervalech, jejich skutečné dotazy probíhají velmi často. Zejména u párů won-dolar, won-jen a won-juan dochází k velmi častému čtení z cache. V takových případech by bylo efektivní, aby API server měl vlastní lokální cache a při změně dat znovu načetl Redis a provedl aktualizaci. Jak tedy lze v architektuře databáze – Redis – API server tuto funkci implementovat?? ### Nedá se to řešit přes Redis PubSub? > Při použití cache bychom se měli přihlásit k odběru kanálu, ze kterého lze získat informaci o tom, zda došlo k aktualizaci! - Je pak nutné vytvořit logiku, která při aktualizaci odešle zprávu. - Dodatečné operace spojené s PubSub ovlivňují výkon. ![pubsub-write](/assets/images/redis-client-side-cache/01-pubsub-write.png) ![pubsub-read](/assets/images/redis-client-side-cache/01-pubsub-read.png) ### Co když Redis změny sám detekuje? > Co kdybychom pomocí Keyspace Notification dostávali oznámení o příkazech pro daný klíč? - Je nutné předem uložit a sdílet klíče a příkazy používané k aktualizaci, což je pracné. - Například u některých klíčů může být příkazem k aktualizaci jednoduchý Set, u jiných LPush, nebo RPush, SAdd a SRem, takže se situace stává složitou. - To výrazně zvyšuje pravděpodobnost komunikačních chyb při vývoji a lidských chyb při programování. > Co kdybychom pomocí Keyevent Notification dostávali oznámení po jednotlivých příkazech? - Je nutné přihlásit se k odběru všech příkazů používaných k aktualizaci. Přicházející klíče je třeba vhodně filtrovat. - Například u klíčů přicházejících příkazem Del klient pravděpodobně nemá žádnou lokální cache. - To může vést k zbytečnému plýtvání prostředky. ### Proto potřebujeme Invalidation Message! #### Co je Invalidation Message? Invalidation Messages je koncept poskytovaný v rámci Server Assisted Client-Side Cache, který byl přidán v Redis 6.0. Invalidation Message je doručována následujícím postupem. 1. Předpokládejme, že ClientB již jednou přečetl key. 2. ClientA tento key nově nastaví. 3. Redis změnu detekuje a vydá ClientB Invalidation Message, kterou ho informuje, aby cache vymazal. 4. ClientB zprávu přijme a provede příslušné kroky. ![invalidation-message](/assets/images/redis-client-side-cache/02-invalidation-message.png) ### Jak se to používá #### Základní struktura fungování Klient připojený k Redisu spustí `CLIENT TRACKING ON REDIRECT `, čímž zajistí, že bude přijímat invalidation message. Klient, který má tyto zprávy přijímat, se pak přihlásí k odběru pomocí `SUBSCRIBE __redis__:invalidate`. #### Výchozí tracking (default tracking) ```shell # client 1 > SET a 100 ``` ```shell # client 3 > CLIENT ID 12 > SUBSCRIBE __redis__:invalidate 1) "subscribe" 2) "__redis__:invalidate" 3) (integer) 1 ``` ```shell # client 2 > CLIENT TRACKING ON REDIRECT 12 > GET a # tracking ``` ```shell # client 1 > SET a 200 ``` ```shell # client 3 1) "message" 2) "__redis__:invalidate" 3) 1) "a" ``` #### Broadcasting tracking ```shell # client 3 > CLIENT ID 12 > SUBSCRIBE __redis__:invalidate 1) "subscribe" 2) "__redis__:invalidate" 3) (integer) 1 ``` ```shell # client 2 CLIENT TRACKING ON BCAST PREFIX cache: REDIRECT 12 ``` ```shell # client 1 > SET cache:name "Alice" > SET cache:age 26 ``` ```shell # client 3 1) "message" 2) "__redis__:invalidate" 3) 1) "cache:name" 1) "message" 2) "__redis__:invalidate" 3) 1) "cache:age" ``` ## Implementace! Implementace! Implementace! ### Redigo + Ristretto Pouhé takto abstraktní vysvětlení je nejasné, jak to skutečně použít v kódu. Proto nejprve sestavíme jednoduchou verzi pomocí `redigo` a `ristretto`. Nejprve nainstalujeme obě závislosti. - `github.com/gomodule/redigo` - `github.com/dgraph-io/ristretto` ```go package main import ( "context" "fmt" "log/slog" "time" "github.com/dgraph-io/ristretto" "github.com/gomodule/redigo/redis" ) type RedisClient struct { conn redis.Conn cache *ristretto.Cache[string, any] addr string } func NewRedisClient(addr string) (*RedisClient, error) { cache, err := ristretto.NewCache(&ristretto.Config[string, any]{ NumCounters: 1e7, // number of keys to track frequency of (10M). MaxCost: 1 << 30, // maximum cost of cache (1GB). BufferItems: 64, // number of keys per Get buffer. }) if err != nil { return nil, fmt.Errorf("failed to generate cache: %w", err) } conn, err := redis.Dial("tcp", addr) if err != nil { return nil, fmt.Errorf("failed to connect to redis: %w", err) } return &RedisClient{ conn: conn, cache: cache, addr: addr, }, nil } func (r *RedisClient) Close() error { err := r.conn.Close() if err != nil { return fmt.Errorf("failed to close redis connection: %w", err) } return nil } ``` Nejprve jednoduše vytvoříme `RedisClient`, který obsahuje ristretto a redigo. ```go func (r *RedisClient) Tracking(ctx context.Context) error { psc, err := redis.Dial("tcp", r.addr) if err != nil { return fmt.Errorf("failed to connect to redis: %w", err) } clientId, err := redis.Int64(psc.Do("CLIENT", "ID")) if err != nil { return fmt.Errorf("failed to get client id: %w", err) } slog.Info("client id", "id", clientId) subscriptionResult, err := redis.String(r.conn.Do("CLIENT", "TRACKING", "ON", "REDIRECT", clientId)) if err != nil { return fmt.Errorf("failed to enable tracking: %w", err) } slog.Info("subscription result", "result", subscriptionResult) if err := psc.Send("SUBSCRIBE", "__redis__:invalidate"); err != nil { return fmt.Errorf("failed to subscribe: %w", err) } psc.Flush() for { msg, err := psc.Receive() if err != nil { return fmt.Errorf("failed to receive message: %w", err) } switch msg := msg.(type) { case redis.Message: slog.Info("received message", "channel", msg.Channel, "data", msg.Data) key := string(msg.Data) r.cache.Del(key) case redis.Subscription: slog.Info("subscription", "kind", msg.Kind, "channel", msg.Channel, "count", msg.Count) case error: return fmt.Errorf("error: %w", msg) case []interface{}: if len(msg) != 3 || string(msg[0].([]byte)) != "message" || string(msg[1].([]byte)) != "__redis__:invalidate" { slog.Warn("unexpected message", "message", msg) continue } contents := msg[2].([]interface{}) keys := make([]string, len(contents)) for i, key := range contents { keys[i] = string(key.([]byte)) r.cache.Del(keys[i]) } slog.Info("received invalidation message", "keys", keys) default: slog.Warn("unexpected message", "type", fmt.Sprintf("%T", msg)) } } } ``` Kód je poněkud složitý. - Pro tracking navazujeme ještě jedno další připojení. Učiněno proto, aby PubSub nerušil ostatní operace. - Zjistíme ID přidaného připojení a v připojení, které čte data, nastavíme tracking tak, aby přesměrovával na toto připojení. - Poté se přihlásíme k odběru invalidation message. - Zpracování odběru je poněkud složité. Protože redigo neumí parsovat invalidation message, je nutné přijmout odpověď a zpracovat ji před parsováním. ```go func (r *RedisClient) Get(key string) (any, error) { val, found := r.cache.Get(key) if found { switch v := val.(type) { case int64: slog.Info("cache hit", "key", key) return v, nil default: slog.Warn("unexpected type", "type", fmt.Sprintf("%T", v)) } } slog.Info("cache miss", "key", key) val, err := redis.Int64(r.conn.Do("GET", key)) if err != nil { return nil, fmt.Errorf("failed to get key: %w", err) } r.cache.SetWithTTL(key, val, 1, 10*time.Second) return val, nil } ``` Metoda `Get` nejprve hledá v ristretto a pokud tam hodnota není, načte ji z Redisu, jak ukazuje následující kód. ```go package main import ( "context" "log/slog" "os" "os/signal" "time" ) func main() { ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) defer cancel() client, err := NewRedisClient("localhost:6379") if err != nil { panic(err) } defer client.Close() go func() { if err := client.Tracking(ctx); err != nil { slog.Error("failed to track invalidation message", "error", err) } }() ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() done := ctx.Done() for { select { case <-done: slog.Info("shutting down") return case <-ticker.C: v, err := client.Get("key") if err != nil { slog.Error("failed to get key", "error", err) return } slog.Info("got key", "value", v) } } } ``` Kód pro testování je výše uvedený. Pokud jej vyzkoušíte, uvidíte, že se hodnota pokaždé aktualizuje při každé změně dat v Redisu. Toto řešení je však příliš složité. Především je pro škálování na cluster nutné aktivovat tracking na všech masterech i replikách. ### Rueidis Pokud píšeme v jazyce Go, máme k dispozici nejmodernější a nejvyspělejší `rueidis`. Napíšeme kód, který používá server assisted client side cache v prostředí Redis clusteru pomocí rueidis. Nejprve nainstalujeme závislost. - `github.com/redis/rueidis` Poté napíšeme kód, který načítá data z Redisu. ```go package main import ( "context" "log/slog" "os" "os/signal" "time" "github.com/redis/rueidis" ) func main() { ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) defer cancel() client, err := rueidis.NewClient(rueidis.ClientOption{ InitAddress: []string{"localhost:6379"}, }) if err != nil { panic(err) } ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() done := ctx.Done() for { select { case <-done: slog.Info("shutting down") return case <-ticker.C: const key = "key" resp := client.DoCache(ctx, client.B().Get().Key(key).Cache(), 10*time.Second) if resp.Error() != nil { slog.Error("failed to get key", "error", resp.Error()) continue } i, err := resp.AsInt64() if err != nil { slog.Error("failed to convert response to int64", "error", err) continue } switch resp.IsCacheHit() { case true: slog.Info("cache hit", "key", key) case false: slog.Info("missed key", "key", key) } slog.Info("got key", "value", i) } } } ``` Pro použití client side cache v rueidis stačí jen zavolat `DoCache`. Poté se záznam přidá do lokální cache s dobou, po kterou má být v lokální cache uchován. Opakované volání `DoCache` pak načte data přímo z lokální cache. Samozřejmě zpracovává i invalidation message správně. ### Proč ne redis-go? `redis-go` bohužel neposkytuje oficiální API pro server assisted client side cache. Navíc při vytváření PubSub otevírá nové připojení a neexistuje API, které by umožnilo přímý přístup k tomuto připojení, takže nelze zjistit ani client ID. Proto jsme usoudili, že konfigurace pomocí `redis-go` je nemožná, a tuto variantu jsme vynechali. ## To je pěkné ### Prostřednictvím struktury client side cache - Pokud lze data připravit předem, můžeme díky této struktuře minimalizovat dotazy a provoz na Redis a vždy poskytovat aktuální data. - Tím lze vytvořit jakousi strukturu CQRS a výrazně zvýšit výkon čtení. ![cqrs](/assets/images/redis-client-side-cache/003-cqrs.jpg) ### Jak moc se to zlepšilo? Ve skutečnosti máme tuto strukturu právě v provozu, proto jsme u dvou API změřili přibližnou latenci. Omluvte prosím, že mohu uvést pouze velmi obecné údaje. 1. První API 1. Při prvním dotazu: průměrně 14,63 ms 2. Při dalších dotazech: průměrně 2,82 ms 3. Průměrný rozdíl: 10,98 ms 2. Druhé API 1. Při prvním dotazu: průměrně 14,05 ms 2. Při dalších dotazech: průměrně 1,60 ms 3. Průměrný rozdíl: 11,57 ms Maximálně šlo o přibližně 82% dodatečné zlepšení latence! Očekáváme, že došlo k následujícím zlepšením. - Vynechání síťové komunikace mezi klientem a Redisem a úspora provozu - Snížení počtu čtecích příkazů, které musí Redis sám provést - To má také efekt zvýšení výkonu zápisu. - Minimalizace parsování Redis protokolu - Parsování Redis protokolu také není zcela bez nákladů. Jeho snížení je velká příležitost. Každé řešení je ale kompromisem. Kvůli tomu jsme obětovali alespoň následující dvě věci. - Nutnost implementace, provozu a údržby prvků pro správu client side cache - Zvýšení využití CPU a paměti na straně klienta v důsledku tohoto řešení ## Závěr Osobně jsem považoval tento architektonický prvek za uspokojivý a latence i zátěž API serveru byla velmi nízká. Do budoucna bych rád, pokud to bude možné, navrhoval architekturu s touto strukturou.