--- id: 1c72ae6efee059c2c727ea53b56bf6c5 author: snowmerak title: Zwiększanie responsywności dzięki klientowi cache Redis po stronie klienta description: Poprawa responsywności API z wykorzystaniem Redis Server Assisted Client Side Cache language: pl date: 2024-11-29T00:51:27.525566863Z path: /blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502 category: database --- ## Czym jest Redis? Przypuszczam, że niewiele osób nie zna Redisa. Mimo to, pokrótce wspomnę o kilku jego cechach, które można podsumować w następujący sposób. - Operacje wykonywane są w **jednym wątku**, dzięki czemu wszystkie operacje mają **atomowość**. - Dane są przechowywane i przetwarzane **In-Memory**, dlatego wszystkie operacje są **szybkie**. - Redis może zapisywać **WAL** w zależności od opcji, dzięki czemu można szybko **tworzyć kopię zapasową** najnowszego stanu i **odzyskiwać** dane. - Obsługuje wiele **typów danych**, takich jak Set, Hash, Bit, List, co zapewnia wysoką **produktywność**. - Posiada **dużą społeczność**, dzięki czemu można **dzielić się** i otrzymywać różnorodne doświadczenia, problemy oraz rozwiązania. - Jest **rozwijany i eksploatowany przez długi czas**, dzięki czemu cechuje go godna zaufania **stabilność**. ## Przechodząc do sedna ### Wyobraźmy sobie? Co by było, gdyby pamięć podręczna Twojej usługi spełniała dwa poniższe warunki? 1. Często przeglądane dane muszą być udostępniane użytkownikom w aktualnej postaci, lecz ich aktualizacje zachodzą nieregularnie, przez co pamięć podręczną trzeba często odświeżać. 2. Dane nie ulegają aktualizacji, ale ta sama pamięć podręczna jest wielokrotnie odczytywana. W pierwszym przypadku można rozważyć ranking popularności produktów w sklepie internetowym w czasie rzeczywistym. Jeśli ranking popularności produktów zapisany jest jako sorted set, odczytywanie go z Redisa za każdym razem, gdy użytkownik wchodzi na stronę główną, jest nieefektywne. W drugim przypadku, dane kursów walut są publikowane mniej więcej co 10 minut, jednak rzeczywiste odczyty następują bardzo często. Szczególnie często odwołujemy się do pamięci podręcznej dla kursów USD/KRW, JPY/KRW oraz CNY/KRW. W takich przypadkach efektywniejszym rozwiązaniem byłoby, aby serwer API posiadał lokalną, osobną pamięć podręczną, a po zmianie danych ponownie odczytywał je z Redisa i aktualizował lokalną pamięć. Jak więc zrealizować takie działanie w strukturze baza danych - Redis - serwer API? ### Czy nie wystarczy Redis PubSub? > Podczas korzystania z pamięci podręcznej warto zasubskrybować kanał, na którym można otrzymywać informacje o aktualizacjach! - Wtedy trzeba stworzyć logikę, która przy aktualizacji wysyła wiadomość. - Dodatkowe operacje związane z PubSub wpływają na wydajność. ![pubsub-write](/assets/images/redis-client-side-cache/01-pubsub-write.png) ![pubsub-read](/assets/images/redis-client-side-cache/01-pubsub-read.png) ### A gdyby Redis sam wykrywał zmiany? > Co jeśli użyjemy Keyspace Notification, aby otrzymywać powiadomienia o poleceniach dla danego klucza? - Występuje kłopot, polegający na konieczności wcześniejszego zapisania i udostępnienia kluczy oraz poleceń używanych do aktualizacji. - Na przykład, dla jednych kluczy poleceniem aktualizującym jest zwykły Set, dla innych LPush, a dla jeszcze innych RPush, SAdd lub SRem, co sprawia, że sytuacja staje się złożona. - Znacznie zwiększa to prawdopodobieństwo błędów w komunikacji podczas rozwoju oraz błędów ludzkich w kodzie. > Co jeśli użyjemy Keyevent Notification, aby otrzymywać powiadomienia dla każdego polecenia? - Konieczna jest subskrypcja wszystkich poleceń używanych do aktualizacji. Należy również odpowiednio filtrować klucze, które napływają. - Na przykład, spośród wszystkich kluczy przychodzących z poleceniem Del, dla części z nich klient prawdopodobnie nie posiada lokalnej pamięci podręcznej. - Może to prowadzić do niepotrzebnego marnotrawstwa zasobów. ### Dlatego potrzebny jest Invalidation Message! #### Czym jest Invalidation Message? Invalidation Messages to koncepcja udostępniana w ramach Server Assisted Client-Side Cache, dodanego w Redis 6.0. Invalidation Message jest przekazywany zgodnie z następującym przebiegiem. 1. Zakładamy, że ClientB odczytał już raz klucz. 2. ClientA ustawia ten klucz na nowo. 3. Redis wykrywa zmianę i wysyła do ClientB Invalidation Message, informując go, aby usunął pamięć podręczną. 4. ClientB odbiera tę wiadomość i podejmuje odpowiednie działania. ![invalidation-message](/assets/images/redis-client-side-cache/02-invalidation-message.png) ### Jak to się używa? #### Podstawowa struktura działania Klient połączony z Redisem wykonuje `CLIENT TRACKING ON REDIRECT `, aby otrzymywać Invalidation Message. Następnie klient, który ma otrzymywać wiadomości, subskrybuje `SUBSCRIBE __redis__:invalidate`, aby odbierać Invalidation Message. #### 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" ``` ## Implementacja! Implementacja! Implementacja! ### Redigo + Ristretto Samo wyjaśnienie w ten sposób sprawia, że trudno wiedzieć, jak to faktycznie zastosować w kodzie. Dlatego najpierw zbudujemy prostą wersję z użyciem `redigo` i `ristretto`. Najpierw instalujemy dwie zależności. - `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 } ``` Najpierw w prosty sposób tworzymy `RedisClient`, który zawiera ristretto i 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)) } } } ``` Kod jest nieco skomplikowany. - Aby wykonać Tracking, nawiązujemy jeszcze jedno połączenie. Zrobiono to z uwagi na to, że PubSub mógłby przeszkadzać innym operacjom. - Pobieramy identyfikator dodatkowego połączenia i ustawiamy, aby Tracking w połączeniu, które odczytuje dane, przekierowywał się do tego połączenia. - Następnie subskrybujemy invalidation message. - Obsługa subskrypcji jest nieco skomplikowana. Ponieważ redigo nie parsuje wiadomości unieważniających, trzeba odebrać odpowiedź przed parsowaniem i dopiero ją przetworzyć. ```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` najpierw sprawdza ristretto w następujący sposób, a jeśli wartości tam nie ma, pobiera ją z Redisa. ```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) } } } ``` Powyższy kod służy do testów. Jeśli go uruchomisz, zobaczysz, że za każdym razem, gdy dane w Redisie zostaną zaktualizowane, wartość jest odświeżana. Jest to jednak zbyt skomplikowane. Co więcej, aby rozszerzyć rozwiązanie na klaster, konieczne jest włączenie Tracking dla wszystkich master oraz replik. ### Rueidis Skoro piszemy w Go, dysponujemy najnowocześniejszą i najbardziej rozwiniętą biblioteką, czyli `rueidis`. Napiszmy kod wykorzystujący server assisted client side cache w środowisku klastra Redis z użyciem rueidis. Najpierw instalujemy zależność. - `github.com/redis/rueidis` Następnie piszemy kod, który odpytuje Redisa o dane. ```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) } } } ``` Aby użyć client side cache w rueidis, wystarczy wywołać `DoCache`. Następnie można dodać wynik do lokalnej pamięci podręcznej, określając, jak długo ma być w niej przechowywany, a kolejne wywołanie `DoCache` pobierze dane z lokalnej pamięci. Oczywiście obsługuje on także invalidation message. ### Dlaczego nie redis-go? Niestety `redis-go` nie obsługuje server assisted client side cache w swoim oficjalnym API. Co więcej, przy tworzeniu PubSub otwiera nowe połączenie, a brak API umożliwiającego bezpośredni dostęp do tego połączenia sprawia, że nie da się poznać client id. Dlatego uznaliśmy, że konfiguracja z użyciem `redis-go` jest niemożliwa i ją pominęliśmy. ## Jak to seksowne ### Dzięki strukturze client side cache - Jeśli dane można przygotować z wyprzedzeniem, dzięki tej strukturze można zminimalizować zapytania i ruch do Redisa, a jednocześnie zawsze udostępniać aktualne dane. - Pozwala to stworzyć rodzaj struktury CQRS, a tym samym znacząco zwiększyć wydajność odczytu. ![cqrs](/assets/images/redis-client-side-cache/003-cqrs.jpg) ### Jak bardzo stało się to bardziej seksowne? Faktycznie, na miejscu używamy właśnie takiej struktury, dlatego sprawdziliśmy przybliżone opóźnienia dla dwóch API. Proszę o wyrozumiałość, że mogę przedstawić je jedynie w sposób bardzo ogólny. 1. Pierwsze API 1. Przy pierwszym zapytaniu: średnio 14,63 ms 2. Przy kolejnych zapytaniach: średnio 2,82 ms 3. Średnia różnica: 10,98 ms 2. Drugie API 1. Przy pierwszym zapytaniu: średnio 14,05 ms 2. Przy kolejnych zapytaniach: średnio 1,60 ms 3. Średnia różnica: 11,57 ms W najlepszym przypadku uzyskano dodatkową poprawę opóźnień rzędu 82%! Spodziewamy się, że przyczyniły się do tego następujące usprawnienia. - Pominięcie komunikacji sieciowej między klientem a Redisem oraz oszczędność ruchu - Zmniejszenie liczby poleceń odczytu, które musi wykonać sam Redis - Ma to również efekt w postaci zwiększenia wydajności zapisu. - Minimalizacja parsowania protokołu Redis - Parsowanie protokołu Redis także nie jest bezkosztowe. Ograniczenie tego jest dużą szansą. Jednak każde rozwiązanie wiąże się z kompromisami. Dlatego musieliśmy poświęcić przynajmniej dwie rzeczy. - Konieczność wdrożenia, utrzymania i obsługi elementów zarządzania client side cache - Wynikający z tego wzrost zużycia procesora i pamięci po stronie klienta ## Wnioski Osobiście uważam, że był to satysfakcjonujący element architektury, a obciążenie opóźnieniami oraz serwera API było bardzo niewielkie. Chciałbym w przyszłości, o ile to możliwe, budować architekturę właśnie w ten sposób.