GoSuda

Îmbunătățirea reactivității prin cache pe partea de client Redis

By snowmerak views ...

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

pubsub-read

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

Cum se utilizează

Structura de funcționare de bază

Clientul conectat la Redis execută CLIENT TRACKING ON REDIRECT <client-id> 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

1# client 1
2> SET a 100

1# client 3
2> CLIENT ID
312
4> SUBSCRIBE __redis__:invalidate
51) "subscribe"
62) "__redis__:invalidate"
73) (integer) 1

1# client 2
2> CLIENT TRACKING ON REDIRECT 12
3> GET a # tracking

1# client 1
2> SET a 200

1# client 3
21) "message"
32) "__redis__:invalidate"
43) 1) "a"

Broadcasting tracking

1# client 3
2> CLIENT ID
312
4> SUBSCRIBE __redis__:invalidate
51) "subscribe"
62) "__redis__:invalidate"
73) (integer) 1

1# client 2
2CLIENT TRACKING ON BCAST PREFIX cache: REDIRECT 12

1# client 1
2> SET cache:name "Alice"
3> SET cache:age 26

1# client 3
21) "message"
32) "__redis__:invalidate"
43) 1) "cache:name"
51) "message"
62) "__redis__:invalidate"
73) 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
 1package main
 2
 3import (
 4	"context"
 5	"fmt"
 6	"log/slog"
 7	"time"
 8
 9	"github.com/dgraph-io/ristretto"
10	"github.com/gomodule/redigo/redis"
11)
12
13type RedisClient struct {
14	conn  redis.Conn
15	cache *ristretto.Cache[string, any]
16	addr  string
17}
18
19func NewRedisClient(addr string) (*RedisClient, error) {
20	cache, err := ristretto.NewCache(&ristretto.Config[string, any]{
21		NumCounters: 1e7,     // number of keys to track frequency of (10M).
22		MaxCost:     1 << 30, // maximum cost of cache (1GB).
23		BufferItems: 64,      // number of keys per Get buffer.
24	})
25	if err != nil {
26		return nil, fmt.Errorf("failed to generate cache: %w", err)
27	}
28
29	conn, err := redis.Dial("tcp", addr)
30	if err != nil {
31		return nil, fmt.Errorf("failed to connect to redis: %w", err)
32	}
33
34	return &RedisClient{
35		conn:  conn,
36		cache: cache,
37		addr:  addr,
38	}, nil
39}
40
41func (r *RedisClient) Close() error {
42	err := r.conn.Close()
43	if err != nil {
44		return fmt.Errorf("failed to close redis connection: %w", err)
45	}
46
47	return nil
48}

Mai întâi creăm un RedisClient simplu care conține ristretto și redigo.

 1func (r *RedisClient) Tracking(ctx context.Context) error {
 2	psc, err := redis.Dial("tcp", r.addr)
 3	if err != nil {
 4		return fmt.Errorf("failed to connect to redis: %w", err)
 5	}
 6
 7	clientId, err := redis.Int64(psc.Do("CLIENT", "ID"))
 8	if err != nil {
 9		return fmt.Errorf("failed to get client id: %w", err)
10	}
11	slog.Info("client id", "id", clientId)
12
13	subscriptionResult, err := redis.String(r.conn.Do("CLIENT", "TRACKING", "ON", "REDIRECT", clientId))
14	if err != nil {
15		return fmt.Errorf("failed to enable tracking: %w", err)
16	}
17	slog.Info("subscription result", "result", subscriptionResult)
18
19	if err := psc.Send("SUBSCRIBE", "__redis__:invalidate"); err != nil {
20		return fmt.Errorf("failed to subscribe: %w", err)
21	}
22	psc.Flush()
23
24	for {
25		msg, err := psc.Receive()
26		if err != nil {
27			return fmt.Errorf("failed to receive message: %w", err)
28		}
29
30		switch msg := msg.(type) {
31		case redis.Message:
32			slog.Info("received message", "channel", msg.Channel, "data", msg.Data)
33			key := string(msg.Data)
34			r.cache.Del(key)
35		case redis.Subscription:
36			slog.Info("subscription", "kind", msg.Kind, "channel", msg.Channel, "count", msg.Count)
37		case error:
38			return fmt.Errorf("error: %w", msg)
39		case []interface{}:
40			if len(msg) != 3 || string(msg[0].([]byte)) != "message" || string(msg[1].([]byte)) != "__redis__:invalidate" {
41				slog.Warn("unexpected message", "message", msg)
42				continue
43			}
44
45			contents := msg[2].([]interface{})
46			keys := make([]string, len(contents))
47			for i, key := range contents {
48				keys[i] = string(key.([]byte))
49				r.cache.Del(keys[i])
50			}
51			slog.Info("received invalidation message", "keys", keys)
52		default:
53			slog.Warn("unexpected message", "type", fmt.Sprintf("%T", msg))
54		}
55	}
56}

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.
 1func (r *RedisClient) Get(key string) (any, error) {
 2	val, found := r.cache.Get(key)
 3	if found {
 4		switch v := val.(type) {
 5		case int64:
 6			slog.Info("cache hit", "key", key)
 7			return v, nil
 8		default:
 9			slog.Warn("unexpected type", "type", fmt.Sprintf("%T", v))
10		}
11	}
12	slog.Info("cache miss", "key", key)
13
14	val, err := redis.Int64(r.conn.Do("GET", key))
15	if err != nil {
16		return nil, fmt.Errorf("failed to get key: %w", err)
17	}
18
19	r.cache.SetWithTTL(key, val, 1, 10*time.Second)
20	return val, nil
21}

El método Get consulta primero ristretto de la siguiente manera y, si no encuentra el valor, lo obtiene de Redis.

 1package main
 2
 3import (
 4	"context"
 5	"log/slog"
 6	"os"
 7	"os/signal"
 8	"time"
 9)
10
11func main() {
12	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
13	defer cancel()
14
15	client, err := NewRedisClient("localhost:6379")
16	if err != nil {
17		panic(err)
18	}
19	defer client.Close()
20
21	go func() {
22		if err := client.Tracking(ctx); err != nil {
23			slog.Error("failed to track invalidation message", "error", err)
24		}
25	}()
26
27	ticker := time.NewTicker(1 * time.Second)
28	defer ticker.Stop()
29	done := ctx.Done()
30
31	for {
32		select {
33		case <-done:
34			slog.Info("shutting down")
35			return
36		case <-ticker.C:
37			v, err := client.Get("key")
38			if err != nil {
39				slog.Error("failed to get key", "error", err)
40				return
41			}
42			slog.Info("got key", "value", v)
43		}
44	}
45}

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.

 1package main
 2
 3import (
 4	"context"
 5	"log/slog"
 6	"os"
 7	"os/signal"
 8	"time"
 9
10	"github.com/redis/rueidis"
11)
12
13func main() {
14	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
15	defer cancel()
16
17	client, err := rueidis.NewClient(rueidis.ClientOption{
18		InitAddress: []string{"localhost:6379"},
19	})
20	if err != nil {
21		panic(err)
22	}
23
24	ticker := time.NewTicker(1 * time.Second)
25	defer ticker.Stop()
26	done := ctx.Done()
27
28	for {
29		select {
30		case <-done:
31			slog.Info("shutting down")
32			return
33		case <-ticker.C:
34			const key = "key"
35			resp := client.DoCache(ctx, client.B().Get().Key(key).Cache(), 10*time.Second)
36			if resp.Error() != nil {
37				slog.Error("failed to get key", "error", resp.Error())
38				continue
39			}
40			i, err := resp.AsInt64()
41			if err != nil {
42				slog.Error("failed to convert response to int64", "error", err)
43				continue
44			}
45			switch resp.IsCacheHit() {
46			case true:
47				slog.Info("cache hit", "key", key)
48			case false:
49				slog.Info("missed key", "key", key)
50			}
51			slog.Info("got key", "value", i)
52		}
53	}
54}

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

¿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.