--- id: 1c72ae6efee059c2c727ea53b56bf6c5 author: snowmerak title: Îmbunătățirea reactivității prin cache pe partea de client Redis description: Îmbunătățirea reactivității API-urilor prin utilizarea Client Side Cache asistat de Redis Server language: ro date: 2024-11-29T00:51:27.525566863Z path: /blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502 category: database --- ## Ce este Redis? Presupun că nu sunt mulți cei care nu cunosc Redis. Totuși, pentru a le menționa pe scurt, câteva dintre caracteristicile sale pot fi rezumate după cum urmează. - Operațiile sunt executate pe **un singur fir de execuție (single thread)**, prin urmare toate operațiile au **atomicitate**. - Datele sunt stocate și procesate **In-Memory**, prin urmare toate operațiile sunt **rapide**. - Redis poate stoca **WAL** în funcție de opțiunile setate, ceea ce permite **copierea de siguranță** și **restaurarea** rapidă a celei mai recente stări. - Suportă **mai multe tipuri** de date, precum Set, Hash, Bit și List, ceea ce oferă o **productivitate** ridicată. - Dispune de o **comunitate largă**, ceea ce permite **schimbul** de experiențe, probleme și soluții. - Fiind **dezvoltat și întreținut de mult timp**, oferă o **stabilitate** în care se poate avea încredere. ## Acum, la subiect ### Imaginați-vă Ce s-ar întâmpla dacă memoria cache a serviciului dumneavoastră ar corespunde următoarelor două condiții? 1. Când datele accesate frecvent trebuie furnizate utilizatorului în stare actualizată, dar actualizarea are loc neregulat, fiind necesară reîmprospătarea frecventă a cache-ului 2. Când datele din cache nu se actualizează, dar aceleași date sunt accesate și interogate frecvent Primul caz poate fi ilustrat prin clasamentul în timp real al produselor populare dintr-un magazin online. Dacă acest clasament este stocat ca sorted set, citirea lui din Redis ori de câte ori un utilizator accesează pagina principală este ineficientă. Al doilea caz se referă la datele cursului de schimb valutar: deși cursurile sunt publicate aproximativ la fiecare 10 minute, interogarea efectivă are loc foarte frecvent. Mai ales pentru cursurile won-dolar, won-yen și won-yuan, cache-ul este interogat extrem de des. În astfel de cazuri, este mai eficient ca serverul API să dețină un cache local separat și să reîncarce datele din Redis atunci când acestea se modifică. Cum putem, așadar, să implementăm acest mod de funcționare în arhitectura bază de date - Redis - server API?? ### Nu ar fi suficient Redis PubSub? > Când folosim cache-ul, să ne abonăm la un canal prin care putem primi notificări despre actualizări! - Atunci trebuie creată logica de trimitere a unui mesaj la fiecare actualizare. - Operațiunile suplimentare generate de PubSub afectează performanța. ![pubsub-write](/assets/images/redis-client-side-cache/01-pubsub-write.png) ![pubsub-read](/assets/images/redis-client-side-cache/01-pubsub-read.png) ### Ce se întâmplă dacă Redis detectează singur modificările? > Dacă folosim Keyspace Notification pentru a primi notificări despre comenzile aplicate unei anumite chei? - Este necesar să stocăm și să comunicăm în prealabil cheile și comenzile utilizate pentru actualizare, ceea ce reprezintă o sarcină incomodă. - De exemplu, pentru unele chei comanda de actualizare este un simplu Set, pentru altele LPush, sau RPush, SAdd și SRem, ceea ce complică lucrurile. - Acest lucru crește considerabil probabilitatea apariției erorilor de comunicare în procesul de dezvoltare și a erorilor umane în scrierea codului. > Dacă folosim Keyevent Notification pentru a primi notificări pe unitate de comandă? - Este necesar să ne abonăm la toate comenzile care realizează actualizări. Pentru cheile primite trebuie aplicată o filtrare adecvată. - De exemplu, dintre toate cheile primite prin Del, pentru unele clientul probabil nu are cache local. - Acest lucru poate duce la consum inutil de resurse. ### De aceea este necesar Invalidation Message! #### Ce este Invalidation Message? Invalidation Messages reprezintă un concept oferit ca parte a Server Assisted Client-Side Cache, introdus în Redis 6.0. Invalidation Message este transmis conform următorului flux. 1. Presupunem că ClientB a citit deja o dată cheia (key). 2. ClientA setează din nou cheia respectivă. 3. Redis detectează modificarea și emite către ClientB un Invalidation Message, pentru a-i comunica să șteargă cache-ul. 4. ClientB primește mesajul și întreprinde acțiunile corespunzătoare. ![invalidation-message](/assets/images/redis-client-side-cache/02-invalidation-message.png) ### Cum se utilizează #### Structura de funcționare de bază Clientul conectat la Redis execută `CLIENT TRACKING ON REDIRECT ` pentru a primi Invalidation Message. Apoi, clientul care trebuie să primească mesajele se abonează la `SUBSCRIBE __redis__:invalidate` pentru a le recepționa. #### 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" ``` ## Implementare! Implementare! Implementare! ### Redigo + Ristretto Dacă explicăm doar în acest mod, nu este clar cum trebuie folosit efectiv în cod. Așadar, pentru început, vom construi o configurație simplă cu `redigo` și `ristretto`. Mai întâi instalăm cele două dependențe. - `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 } ``` Mai întâi creăm un `RedisClient` simplu care conține 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)) } } } ``` Il código es un poco complejo. - Para realizar el Tracking se establece una conexión adicional. Esto se hace considerando que PubSub podría interferir con otras operaciones. - Se obtiene el identificador de la conexión adicional y se configura el Tracking de la conexión que consulta los datos para que redirija a esa conexión. - Luego se suscribe al invalidation message. - El código que procesa la suscripción es algo complejo. Como redigo no analiza los mensajes de invalidación, es necesario recibir la respuesta antes del análisis y procesarla manualmente. ```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 } ``` El método `Get` consulta primero ristretto de la siguiente manera y, si no encuentra el valor, lo obtiene de Redis. ```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) } } } ``` El código anterior sirve para realizar pruebas. Si lo ejecuta, podrá comprobar que el valor se actualiza cada vez que los datos cambian en Redis. Sin embargo, esto resulta demasiado complejo. Sobre todo, para escalar en un clúster es necesario habilitar el Tracking en todos los maestros y réplicas. ### Rueidis Mientras se use el lenguaje Go, contamos con `rueidis`, la opción más moderna y avanzada. Escribiremos un código que utiliza server assisted client side cache en un entorno de clúster de Redis con rueidis. Primero, instalamos la dependencia. - `github.com/redis/rueidis` Luego escribimos el código que consulta datos en 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) } } } ``` Para utilizar client side cache con rueidis basta con llamar a `DoCache`. Al hacerlo, se añade el dato a la caché local, incluyendo el tiempo durante el que debe conservarse en ella. Si se vuelve a llamar a `DoCache`, los datos se obtienen de la caché local. Naturalmente, también se procesan correctamente los mensajes de invalidación. ### ¿Por qué no redis-go? Lamentablemente, `redis-go` no admite server assisted client side cache mediante su API oficial. Además, al crear PubSub genera una nueva conexión y no ofrece una API para acceder directamente a ella, por lo que tampoco es posible conocer el client id. Por ello, consideramos que la configuración con `redis-go` es imposible y la descartamos. ## Qué atractivo ### A través de la estructura de client side cache - Si los datos pueden prepararse de antemano, esta estructura permite minimizar las consultas y el tráfico hacia Redis, al tiempo que se ofrecen siempre datos actualizados. - De este modo, se puede crear una especie de estructura CQRS que mejora notablemente el rendimiento de lectura. ![cqrs](/assets/images/redis-client-side-cache/003-cqrs.jpg) ### ¿Cuánto más atractivo se volvió? En realidad, en el entorno de trabajo utilizamos justo esta estructura, así que medimos la latencia aproximada de dos API. Disculpen que solo pueda expresarlo de forma muy abstracta. 1. Primera API 1. Consulta inicial: promedio de 14.63 ms 2. Consultas posteriores: promedio de 2.82 ms 3. Diferencia promedio: 10.98 ms 2. Segunda API 1. Consulta inicial: promedio de 14.05 ms 2. Consultas posteriores: promedio de 1.60 ms 3. Diferencia promedio: 11.57 ms ¡Hubo una mejora adicional de latencia de hasta aproximadamente 82%! Esperamos que se hayan producido las siguientes mejoras. - Omisión de la comunicación de red entre el cliente y Redis y ahorro de tráfico - Reducción del número de comandos de lectura que Redis debe ejecutar - Esto también tiene el efecto de mejorar el rendimiento de escritura. - Minimización del análisis del protocolo de Redis - Analizar el protocolo de Redis tampoco tiene costo cero. Poder reducirlo es una gran oportunidad. Sin embargo, todo es un compromiso. Para ello, sacrificamos al menos las dos cosas siguientes. - Necesidad de implementar, operar y mantener los componentes de gestión del client side cache - Aumento del uso de CPU y memoria en el cliente como consecuencia ## Conclusión Personalmente, fue un componente de arquitectura satisfactorio, y la latencia y la carga sobre el servidor API también fueron muy bajas. Espero que, siempre que sea posible, se pueda seguir construyendo la arquitectura con esta estructura.