GoSuda

Forbedre responsiviteten med en klientside-cache for Redis

By snowmerak views ...

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

pubsub-read

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

Hvordan bruker man det?

Grunnleggende struktur

En klient som er koblet til Redis, kjører CLIENT TRACKING ON REDIRECT <client-id> for å motta invalidation messages. Deretter abonnerer klienten som skal motta meldingene på SUBSCRIBE __redis__:invalidate for å motta invalidation messages.

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"

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
 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,     // antall nøkler det skal holdes oversikt over frekvensen til (10M).
22		MaxCost:     1 << 30, // maksimal kostnad for hurtigbufferen (1GB).
23		BufferItems: 64,      // antall nøkler 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}

Først oppretter vi en enkel RedisClient som inneholder ristretto og 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}

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

Get-metoden slår først opp i ristretto, og henter fra Redis hvis verdien ikke finnes der.

 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}

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.

 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}

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

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.