--- id: 1c72ae6efee059c2c727ea53b56bf6c5 author: snowmerak title: Forbedre responsiviteten med en klientside-cache for Redis description: Forbedring av API-responsivitet ved hjelp av Redis Server Assisted Client Side Cache language: "no" date: 2024-11-29T00:51:27.525566863Z path: /blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502 category: database --- ## Hva er Redis? Jeg antar at få mennesker ikke kjenner til Redis. Likevel vil jeg kort nevne noen av egenskapene, og oppsummere dem slik: - Operasjoner utføres i en **enkelt tråd**, og derfor har alle operasjoner **atomisitet**. - Data lagres og behandles **In-Memory**, og derfor er alle operasjoner **raske**. - Redis kan, avhengig av konfigurasjon, lagre **WAL**, slik at den raskt kan **sikkerhetskopiere** og **gjenopprette** den nyeste tilstanden. - Redis støtter flere **datatyper**, som Set, Hash, Bit og List, og har derfor høy **produktivitet**. - Redis har et **stort fellesskap**, slik at man kan **dele** og få tilgang til ulike erfaringer, problemer og løsninger. - Redis har blitt **utviklet og drevet** over lang tid, og har derfor pålitelig **stabilitet**. ## Så, til hovedsaken ### Tenk deg følgende Hva om hurtigbufferen i tjenesten din oppfyller følgende to betingelser? 1. Data som ofte forespørres må leveres til brukerne i oppdatert tilstand, men oppdateringene skjer uregelmessig, slik at hurtigbufferen må oppdateres hyppig 2. Data oppdateres ikke, men samme hurtigbufferdata må leses gjentatte ganger Det første tilfellet kan være en sanntidsrangering av populære produkter i en netthandel. Når rangeringen lagres som et sorted set, blir det ineffektivt om Redis leses hver gang en bruker åpner hovedsiden. Det andre tilfellet gjelder valutakursdata. Selv om valutakursene publiseres med cirka 10 minutters intervall, forekommer oppslagene svært hyppig. Særlig for KRW-USD, KRW-JPY og KRW-CNY blir hurtigbufferen spurt svært ofte. I slike tilfeller vil det være mer effektivt at API-serveren har en egen lokal hurtigbuffer, og at den henter data fra Redis på nytt når dataene endres. Hvordan kan man da implementere denne atferden i en struktur bestående av database, Redis og API-server? ### Fungerer ikke Redis PubSub? > Når man bruker hurtigbufferen, kan man abonnere på en kanal som gir beskjed om oppdateringer! - Da må det lages logikk som sender en melding ved hver oppdatering. - Den ekstra PubSub-behandlingen påvirker ytelsen. ![pubsub-write](/assets/images/redis-client-side-cache/01-pubsub-write.png) ![pubsub-read](/assets/images/redis-client-side-cache/01-pubsub-read.png) ### Hva om Redis selv oppdager endringene? > Hva om man mottar kommandovarsler for den aktuelle nøkkelen ved hjelp av Keyspace Notification? - Det er besværlig at nøklene og kommandoene som brukes til oppdatering må lagres på forhånd og deles. - For eksempel kan en nøkkel bruke vanlig Set som oppdateringskommando, mens en annen nøkkel bruker LPush, RPush, SAdd eller SRem. Dette gjør systemet komplisert. - Dette øker i betydelig grad risikoen for kommunikasjonsfeil under utviklingen og menneskelige feil i koden. > Hva om man mottar varsler per kommando ved hjelp av Keyevent Notification? - Det er nødvendig å abonnere på alle kommandoer som brukes til oppdatering, og de innkommende nøklene må filtreres på riktig måte. - For eksempel har klienten sannsynligvis ingen lokal hurtigbuffer for noen av nøklene som kommer inn via Del. - Dette kan føre til unødvendig sløsing med ressurser. ### Derfor trengs Invalidation Message! #### Hva er Invalidation Message? Invalidation Messages er et konsept som leveres som en del av Server Assisted Client-Side Cache, introdusert i Redis 6.0. Invalidation Message overføres etter følgende flyt. 1. Anta at ClientB allerede har lest en key én gang. 2. ClientA setter den aktuelle key på nytt. 3. Redis oppdager endringen og sender en Invalidation Message til ClientB, for å be ClientB om å slette hurtigbufferen. 4. ClientB mottar meldingen og iverksetter passende tiltak. ![invalidation-message](/assets/images/redis-client-side-cache/02-invalidation-message.png) ### Hvordan bruker man det? #### Grunnleggende struktur En klient som er koblet til Redis, kjører `CLIENT TRACKING ON REDIRECT ` for å motta invalidation messages. Deretter abonnerer klienten som skal motta meldingene på `SUBSCRIBE __redis__:invalidate` for å motta invalidation messages. #### 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" ``` ## Implementering! Implementering! Implementering! ### Redigo + Ristretto Det er vanskelig å forstå hvordan man faktisk bruker dette i kode, når man bare forklarer det slik. Derfor vil jeg først sette opp et enkelt eksempel med `redigo` og `ristretto`. Først installerer man de to avhengighetene. - `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, // antall nøkler det skal holdes oversikt over frekvensen til (10M). MaxCost: 1 << 30, // maksimal kostnad for hurtigbufferen (1GB). BufferItems: 64, // antall nøkler 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 } ``` Først oppretter vi en enkel `RedisClient` som inneholder ristretto og 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)) } } } ``` Koden er litt komplisert. - For å kunne spore endringene opprettes en ekstra tilkobling. Dette er gjort for at PubSub ikke skal forstyrre andre operasjoner. - Id-en til den ekstra tilkoblingen hentes, og tilkoblingen som leser data settes til å omdirigere sporingen til denne tilkoblingen. - Deretter abonneres det på invalidation messages. - Behandlingen av abonnementet er litt komplisert. Fordi redigo ikke parser invalidation messages, må svaret mottas før parsing og behandles manuelt. ```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 } ``` `Get`-metoden slår først opp i ristretto, og henter fra Redis hvis verdien ikke finnes der. ```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) } } } ``` Koden ovenfor er en test. Om du kjører den, vil du se at verdien oppdateres hver gang dataene endres i Redis. Dette er likevel for komplisert. Dessuten må Tracking aktiveres på alle masterne og replikaene for at løsningen skal kunne skaleres i en cluster. ### Rueidis Når man skriver Go, har vi `rueidis`, som er det mest moderne og best utviklede alternativet. La oss skrive kode som bruker server assisted client side cache i et Redis-cluster med rueidis. Først installerer man avhengigheten. - `github.com/redis/rueidis` Deretter skriver vi kode som henter data fra Redis. ```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) } } } ``` For å bruke client side cache i rueidis er det tilstrekkelig å kalle `DoCache`. Man angir selv hvor lenge verdien skal oppbevares i den lokale hurtigbufferen. Når `DoCache` kalles på nytt, hentes dataene fra den lokale hurtigbufferen. Invalidation messages håndteres også korrekt. ### Hvorfor ikke redis-go? `redis-go` støtter dessverre ikke server assisted client side cache gjennom sitt offisielle API. Når en PubSub opprettes, lages det dessuten en ny tilkobling, og det finnes ikke noe API som gir direkte tilgang til denne tilkoblingen, så client id kan ikke hentes. Derfor vurderte jeg at en løsning med `redis-go` ikke var mulig, og valgte å se bort fra den. ## Elegant! ### Gjennom client side cache-strukturen - For data som kan forberedes på forhånd, kan denne strukturen minimere spørringer og trafikk mot Redis, samtidig som man alltid leverer oppdaterte data. - Dette gir en slags CQRS-struktur, som kan øke lesytelsen betydelig. ![cqrs](/assets/images/redis-client-side-cache/003-cqrs.jpg) ### Hvor mye elegantere ble det? Vi bruker faktisk denne strukturen i produksjon, så jeg målte den enkle latensen for to API-er. Jeg ber om forståelse for at jeg bare kan oppgi resultatene på et svært abstrakt nivå. 1. Første API 1. Ved første oppslag: gjennomsnittlig 14,63 ms 2. Ved etterfølgende oppslag: gjennomsnittlig 2,82 ms 3. Gjennomsnittlig forskjell: 10,98 ms 2. Andre API 1. Ved første oppslag: gjennomsnittlig 14,05 ms 2. Ved etterfølgende oppslag: gjennomsnittlig 1,60 ms 3. Gjennomsnittlig forskjell: 11,57 ms I beste fall var forbedringen i ytterligere latens cirka 82 prosent! Jeg forventer at følgende forbedringer har bidratt: - Nettverkskommunikasjonen mellom klienten og Redis hoppes over, og trafikken spares - Antall leskommandoer som Redis selv må utføre, reduseres - Dette gir også bedre skriveytelse. - Parsing av Redis-protokollen minimeres - Parsing av Redis-protokollen har heller ikke null kostnad. Å redusere den er en stor mulighet. Alt er imidlertid en avveining. For å oppnå dette har vi minst ofret følgende to ting: - Implementering, drift og vedlikehold av komponenter for hurtigbufferhåndtering på klientsiden er nødvendig - Dette øker CPU- og minnebruken på klienten ## Konklusjon Personlig synes jeg dette var en tilfredsstillende arkitekturkomponent, og belastningen på API-serveren og latensen var svært lav. Fremover håper jeg at vi, når det er mulig, kan bygge arkitekturen på denne måten.