--- id: 1c72ae6efee059c2c727ea53b56bf6c5 author: snowmerak title: Zvýšenie responzivity pomocou klientskeho cachovania v Redis description: Zlepšenie odozvy API pomocou Redis Server Assisted Client Side Cache language: sk date: 2024-11-29T00:51:27.525566863Z path: /blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502 category: database --- ## Čo je Redis? Predpokladám, že len málokto nepozná Redis. Napriek tomu ho na úvod stručne pripomeniem a zhrniem jeho niekoľko charakteristík takto: - Operácie sa vykonávajú v **jednom vlákne (single thread)**, a preto majú všetky operácie **atomicitu**. - Dáta sú uložené a spracovávané **v pamäti (In-Memory)**, a preto sú všetky operácie **rýchle**. - Redis môže podľa nastavenia ukladať **WAL**, vďaka čomu možno rýchlo **zálohovať** aktuálny stav a **obnoviť** ho. - Podporuje viacero **typov dát**, ako sú Set, Hash, Bit, List a ďalšie, čo prináša vysokú **produktivitu**. - Má **veľkú komunitu**, vďaka čomu možno získavať a **zdieľať** rôzne skúsenosti, problémy a riešenia. - Je **dlhodobo vyvíjaný a prevádzkovaný**, a preto poskytuje dôveryhodnú **stabilitu**. ## Teraz k podstate ### Predstavte si Čo ak by vaša služba mala cache, ktorá spĺňa tieto dve podmienky? 1. Často dotazované dáta je potrebné poskytovať používateľom v aktuálnom stave, no ich aktualizácie prichádzajú nepravidelne, takže cache treba obnovovať časté. 2. Dáta sa už neaktualizujú, ale k rovnakým cache dátam sa pristupuje a dopytuje veľmi často. Prvý prípad môže predstavovať reálne poradie obľúbenosti produktov v internetovom obchode. Ak je poradie obľúbenosti produktov v reálnom čase uložené v sorted set, je neefektívne čítať ho z Redisu zakaždým, keď používateľ otvorí hlavnú stránku. Druhý prípad sa týka dát o výmenných kurzoch. Hoci sa kurzy vyhlasujú približne v 10-minútových intervaloch, skutočné dopyty prichádzajú veľmi často. Najmä pre kurzy won-dolár, won-jen a won-juan sa cache dotazuje mimoriadne často. V takýchto prípadoch je efektívnejšie, aby API server mal vlastnú lokálnu cache a pri zmene dát znovu načítal údaje z Redisu a obnovil ju. Ako však túto funkcionalitu implementovať v štruktúre databáza - Redis - API server? ### Nedá sa to riešiť cez Redis PubSub? > Pri používaní cache by sme sa mali prihlásiť na odber kanála, ktorý nám oznámi, či sa dáta aktualizovali! - Potom musíme vytvoriť logiku, ktorá pri aktualizácii odošle správu. - Dodatočné operácie spojené s PubSub ovplyvň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) ### Čo ak Redis sám zistí zmenu? > Ak použijeme Keyspace Notification a dostaneme upozornenie na príkazy pre konkrétny kľúč? - Je potrebné vopred uložiť a zdieľať kľúče a príkazy používané na aktualizáciu, čo je zdĺhavé. - Napríklad pri niektorých kľúčoch je aktualizačným príkazom jednoduchý Set, pri iných LPush, prípadne RPush, SAdd a SRem. Situácia sa tak stáva zložitou. - To výrazne zvyšuje pravdepodobnosť komunikačných nedorozumení počas vývoja a ľudských chýb pri kódovaní. > Ak použijeme Keyevent Notification a dostaneme upozornenie na úrovni jednotlivých príkazov? - Je potrebný odber všetkých príkazov používaných na aktualizáciu. Prichádzajúce kľúče je následne nutné vhodne filtrovať. - Napríklad z množstva kľúčov prichádzajúcich pri príkaze Del môže byť pre daného klienta väčšina kľúčov bez lokálnej cache. - To môže viesť k zbytočnému plytvaniu zdrojmi. ### Preto potrebujeme Invalidation Message! #### Čo je Invalidation Message? Invalidation Messages sú koncept poskytovaný ako súčasť Server Assisted Client-Side Cache, ktorý bol pridaný v Redis 6.0. Invalidation Message sa doručuje v nasledujúcom postupe. 1. Predpokladajme, že ClientB už raz prečítal kľúč. 2. ClientA kľúč znovu nastaví. 3. Redis zmenu zistí a pošle ClientB Invalidation Message, ktorou ho informuje, aby vymazal cache. 4. ClientB správu prijme a vykoná primeraný krok. ![invalidation-message](/assets/images/redis-client-side-cache/02-invalidation-message.png) ### Ako sa to používa #### Základná štruktúra fungovania Klient pripojený k Redisu spustí `CLIENT TRACKING ON REDIRECT `, aby mohol prijímať invalidation messages. Klient, ktorý má tieto správy prijímať, sa potom prihlási na odber cez `SUBSCRIBE __redis__:invalidate`. #### 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" ``` ## Implementácia! Implementácia! Implementácia! ### Redigo + Ristretto Keď sa to vysvetlí iba týmto spôsobom, nie je jasné, ako to použiť v skutočnom kóde. Preto skúsime najprv jednoducho zostaviť riešenie pomocou `redigo` a `ristretto`. Najprv nainštalujeme dve 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 } ``` Najprv jednoducho vytvoríme `RedisClient`, ktorý 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 o niečo zložitejší. - Na sledovanie (Tracking) otvárame ešte jedno pripojenie. Týmto opatrením zohľadňujeme, že PubSub by mohol rušiť ostatné operácie. - Zistíme ID pridaného pripojenia a nastavíme, aby sa Tracking pre pripojenie, ktoré číta dáta, presmeroval na toto pripojenie. - Potom sa prihlásime na odber invalidation messages. - Spracovanie odberu je o niečo zložitejšie. Keďže redigo nevie parsovať invalidation messages, musíme prijať odpoveď pred parsovaním a spracovať ju sami. ```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 } ``` Metóda `Get` najprv vyhľadá hodnotu v ristretto a ak tam nie je, načíta ju z Redisu, ako je uvedené vyššie. ```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) } } } ``` Vyššie uvedený kód slúži na testovanie. Keď ho vyskúšate, uvidíte, že sa hodnota pri každej aktualizácii dát v Redisu znovu načíta. Toto riešenie je však príliš zložité. Najmä kvôli škálovaniu na cluster je nevyhnutné aktivovať Tracking na všetkých masteroch aj replikách. ### Rueidis Keďže programujeme v jazyku Go, máme k dispozícii najmodernejší a najvyspelejší `rueidis`. Napíšeme kód, ktorý používa server assisted client side cache v prostredí Redis clusteru pomocou rueidis. Najprv nainštalujeme závislosť: - `github.com/redis/rueidis` Potom napíšeme kód, ktorý dotazuje dáta 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) } } } ``` Na použitie client side cache v rueidis stačí zavolať iba `DoCache`. Potom sa dáta pridajú do lokálnej cache s určením, ako dlho v nej majú zostať. Ak `DoCache` zavoláme znova, dáta sa získajú z lokálnej cache. Invalidation messages sa samozrejme spracúvajú správne. ### Prečo nie redis-go? `redis-go` bohužiaľ neposkytuje server assisted client side cache v oficiálnom API. Navyše pri vytváraní PubSub otvára nové pripojenie a neexistuje API, ktoré by umožnilo priamy prístup k tomuto pripojeniu, takže nie je možné zistiť ani client ID. Preto sme usúdili, že konfigurácia pomocou `redis-go` nie je možná, a tento variant sme vynechali. ## Sexy ### Vďaka štruktúre client side cache - Ak ide o dáta, ktoré je možné pripraviť vopred, touto štruktúrou môžeme minimalizovať dotazy a prevádzku voči Redisu a zároveň poskytovať vždy aktuálne dáta. - Týmto spôsobom môžeme vytvoriť akúsi formu CQRS štruktúry, ktorá výrazne zvýši výkon čítania. ![cqrs](/assets/images/redis-client-side-cache/003-cqrs.jpg) ### O koľko sexy sa to stalo? Keďže táto štruktúra je práve používaná v produkcii, zmerali sme priemerné latencie pre dve API. Prosíme o pochopenie, že výsledky môžem uviesť len veľmi všeobecne. 1. Prvé API 1. Pri prvom dopyte: priemerne 14,63 ms 2. Pri ďalších dopytoch: priemerne 2,82 ms 3. Priemerný rozdiel: 10,98 ms 2. Druhé API 1. Pri prvom dopyte: priemerne 14,05 ms 2. Pri ďalších dopytoch: priemerne 1,60 ms 3. Priemerný rozdiel: 11,57 ms Latencia sa zlepšila až o 82 % a je to naozaj výrazné zlepšenie! Očakávame, že sme dosiahli najmä tieto zlepšenia: - Vynechanie sieťovej komunikácie medzi klientom a Redisom a úspora prenosu dát - Zníženie počtu čítacích príkazov, ktoré musí Redis sám vykonať - Toto má zároveň efekt zvýšenia výkonu zápisov. - Minimalizácia parsovania Redis protokolu - Parsovanie Redis protokolu tiež nie je bezplatné. Zníženie tejto záťaže je veľká príležitosť. Ale každé riešenie je kompromis. Kvôli tomu sme obetovali aspoň tieto dve veci: - Implementácia, prevádzka a údržba komponentov na správu client side cache - S tým súvisiace zvýšenie využitia CPU a pamäte na strane klienta ## Záver Osobne považujem túto architektonickú komponentu za uspokojivú a latencia aj záťaž API servera boli veľmi nízke. Do budúcna by som chcel, ak to bude možné, budovať architektúru práve touto štruktúrou.