GoSuda

Forbedring af reaktionsevne med Redis-klientsidecache

By snowmerak views ...

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

pubsub-read

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

Hvordan bruges det?

Grundlæggende driftsstruktur

Den klient, der er forbundet til Redis, kører CLIENT TRACKING ON REDIRECT <client-id>, så den modtager Invalidation Messages. Den klient, der skal modtage beskederne, abonnerer derefter med SUBSCRIBE __redis__:invalidate for at modtage 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

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

Først oprettes en RedisClient, der indeholder ristretto og redigo på en enkel måde.

 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 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.
 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 op i ristretto som vist nedenfor, og henter derefter fra Redis, hvis værdien ikke findes.

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

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

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.