--- id: 1c72ae6efee059c2c727ea53b56bf6c5 author: snowmerak title: Forbedring af reaktionsevne med Redis-klientsidecache description: Forbedring af API-reaktionsevne ved hjælp af Redis Server Assisted Client Side Cache language: da date: 2024-11-29T00:51:27.525566863Z path: /blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502 category: database --- ## Hvad er Redis? Jeg går ud fra, at de fleste kender til Redis. Men for en god ordens skyld vil jeg kort nævne nogle af dets karakteristika, som kan opsummeres således. - Operationer udføres på **en enkelt tråd**, og derfor har alle operationer **atomicitet**. - Data lagres og behandles **In-Memory**, og derfor er alle operationer **hurtige**. - Afhængigt af konfigurationen kan Redis gemme en **WAL**, hvilket gør det muligt hurtigt at **sikkerhedskopiere** og **gendanne** den seneste tilstand. - Redis understøtter flere **datatyper**, såsom Set, Hash, Bit og List, og har derfor høj **produktivitet**. - Redis har et **stort fællesskab**, hvilket gør det muligt at **dele** og modtage forskellige erfaringer, problemer og løsninger. - Redis har været **udviklet og driftet i lang tid**, og derfor har det en pålidelig **stabilitet**. ## Så til hovedemnet ### Forestil dig? Hvad hvis cachen i din tjeneste opfylder følgende to betingelser? 1. Ofte forespurgte data skal leveres til brugeren i aktuel tilstand, men da opdateringerne sker uregelmæssigt, skal cachen opdateres hyppigt. 2. Data opdateres ikke, men den samme cachedata tilgås og forespørges gentagne gange. Det første tilfælde kan eksempelvis være realtidsrangeringen af populære produkter i en netbutik. Når realtidsrangeringen af populære produkter gemmes som et sorted set, er det ineffektivt at læse fra Redis, hver gang en bruger tilgår forsiden. Det andet tilfælde kan være valutakursdata. Selvom valutakurserne offentliggøres med ca. 10 minutters interval, sker de faktiske opslag meget hyppigt. Særligt for KRW–USD, KRW–JPY og KRW–CNY forespørges cachen meget hyppigt. I sådanne tilfælde vil det være en mere effektiv løsning, at API-serveren har en separat lokal cache og genindlæser fra Redis, når dataene ændres. Hvordan kan denne adfærd så implementeres i strukturen database - Redis - API-server? ### Kan Redis PubSub bruges? > Når cachen anvendes, bør man abonnere på en kanal, hvorfra man kan modtage besked om opdateringer! - Man skal dog udvikle logik, der sender en besked ved hver opdatering. - Den ekstra behandling, som PubSub medfører, påvirker ydeevnen. ![pubsub-write](/assets/images/redis-client-side-cache/01-pubsub-write.png) ![pubsub-read](/assets/images/redis-client-side-cache/01-pubsub-read.png) ### Hvad hvis Redis selv registrerer ændringerne? > Hvad hvis man bruger Keyspace Notification til at modtage kommandobaserede notifikationer for den pågældende nøgle? - Det er besværligt at skulle gemme og dele, hvilke nøgler og kommandoer der bruges til opdateringer, på forhånd. - For eksempel kan en nøgle have en simpel Set som opdateringskommando, mens en anden nøgle bruger LPush, eller RPush, SAdd og SRem som opdateringskommandoer, og det bliver dermed komplekst. - Dette øger markant risikoen for kommunikationsfejl under udviklingen og menneskelige fejl i kodningen. > Hvad hvis man bruger Keyevent Notification til at modtage notifikationer pr. kommando? - Det kræver abonnement på alle kommandoer, der bruges til opdateringer, og de indkommende nøgler skal filtreres korrekt. - For eksempel har klienten sandsynligvis ingen lokal cache for nogle af de nøgler, der kommer ind via Del. - Dette kan føre til unødvendigt spild af ressourcer. ### Derfor er Invalidation Message det, vi har brug for! #### Hvad er Invalidation Message? Invalidation Messages er et koncept, der er tilgængeligt som en del af Server Assisted Client-Side Cache, som blev tilføjet i Redis 6.0. En Invalidation Message leveres i følgende forløb. 1. Antag, at ClientB allerede har læst en key én gang. 2. ClientA sætter den pågældende key til en ny værdi. 3. Redis registrerer ændringen og udsender en Invalidation Message til ClientB for at fortælle ClientB, at cachen skal ryddes. 4. ClientB modtager beskeden og træffer de passende foranstaltninger. ![invalidation-message](/assets/images/redis-client-side-cache/02-invalidation-message.png) ### Hvordan bruges det? #### Grundlæggende driftsstruktur Den klient, der er forbundet til Redis, kører `CLIENT TRACKING ON REDIRECT `, så den modtager Invalidation Messages. Den klient, der skal modtage beskederne, abonnerer derefter med `SUBSCRIBE __redis__:invalidate` for at modtage 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 Hvis man kun forklarer det sådan, er det uklart, hvordan det skal bruges i faktisk kode. Derfor vil jeg først opbygge det enkelt med `redigo` og `ristretto`. Først installeres de to afhængigheder. - `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 } ``` Først oprettes en `RedisClient`, der indeholder ristretto og redigo på en enkel måde. ```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 lidt kompleks. - For at kunne udføre Tracking oprettes en ekstra forbindelse. Dette er en foranstaltning, der tager højde for, at PubSub kan forstyrre andre operationer. - Efter at have fundet forbindelsens ID omdirigeres Tracking for den forbindelse, der læser data, til denne forbindelse. - Derefter abonneres der på Invalidation Message. - Håndteringen af abonnementet er noget kompleks. Da redigo ikke parser invalidation-beskeder, skal svaret modtages og behandles, før det parses. ```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 op i ristretto som vist nedenfor, og henter derefter fra Redis, hvis værdien ikke findes. ```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 til test. Hvis du kører den, vil du se, at værdien opdateres, hver gang dataene ændres i Redis. Denne løsning er dog for kompleks. Frem for alt er det nødvendigt at aktivere Tracking på alle masters eller replicas for at kunne skalere til et cluster. ### Rueidis Når man skriver i Go, har vi `rueidis`, som er det mest moderne og veludviklede bibliotek. Lad os skrive kode, der bruger server assisted client side cache i et Redis-clustermiljø med rueidis. Først installeres afhængigheden. - `github.com/redis/rueidis` Dernæst skrives koden, der forespørger 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 at bruge client side cache i rueidis er det blot nødvendigt at kalde `DoCache`. Herefter tilføjes værdien til den lokale cache med angivelse af, hvor længe den skal opbevares lokalt, og et efterfølgende kald af `DoCache` henter værdien fra den lokale cache. Invalidation Messages håndteres selvfølgelig også korrekt. ### Hvorfor ikke redis-go? `redis-go` understøtter desværre ikke server assisted client side cache via sin officielle API. Desuden findes der ingen API, der giver direkte adgang til den forbindelse, der oprettes, når en PubSub-forbindelse etableres, og derfor kan man heller ikke få client ID'et. Jeg vurderede derfor, at en opsætning med `redis-go` ikke er mulig, og sprang den over. ## Det ser lækkert ud ### Gennem client side cache-strukturen - Hvis dataene kan forberedes på forhånd, kan denne struktur minimere forespørgsler og trafik mod Redis, samtidig med at der altid leveres de nyeste data. - Dermed kan man skabe en slags CQRS-struktur og markant forbedre læseydelsen. ![cqrs](/assets/images/redis-client-side-cache/003-cqrs.jpg) ### Hvor meget lækrere blev det? Vi bruger faktisk allerede denne struktur i produktion, så vi målte den simple latenstid for to API'er. Vi beder om forståelse for, at vi kun kan give meget abstrakte oplysninger. 1. Første API 1. Ved første opslag: gennemsnitligt 14,63 ms 2. Ved efterfølgende opslag: gennemsnitligt 2,82 ms 3. Gennemsnitlig forskel: 10,98 ms 2. Andet API 1. Ved første opslag: gennemsnitligt 14,05 ms 2. Ved efterfølgende opslag: gennemsnitligt 1,60 ms 3. Gennemsnitlig forskel: 11,57 ms Der var en yderligere latenstidsforbedring på op til ca. 82 %! Vi forventer, at følgende forbedringer har bidraget: - Udeladelse af netværkskommunikationen mellem klienten og Redis samt besparelse af trafik - Reduktion af antallet af læsekommandoer, som Redis selv skal udføre - Dette har også en effekt på skriveydelsen. - Minimering af parsing af Redis-protokollen - Parsing af Redis-protokollen har heller ikke nul omkostning. At kunne reducere dette er en stor mulighed. Alt er dog en afvejning. For at opnå dette har vi i det mindste ofret følgende to ting. - Implementering, drift og vedligeholdelse af elementer til styring af client side cache er nødvendig - Deraf følger en øget forbrug af CPU og hukommelse på klienten ## Konklusion Personligt synes jeg, at det var en tilfredsstillende arkitekturkomponent, og belastningen på latenstid og API-server var meget lav. Fremover håber jeg, at vi kan opbygge arkitekturen på denne måde, når det er muligt.