GoSuda

Verbeteren van de responsiviteit met een client-side cache in Redis

By snowmerak views ...

Wat is Redis?

Ik denk dat er weinig mensen zijn die Redis niet kennen. Toch wil ik er kort enkele kenmerken van noemen, die als volgt samengevat kunnen worden.

  • Bewerkingen worden uitgevoerd in een enkele thread, waardoor alle bewerkingen atomair zijn.
  • Gegevens worden In-Memory opgeslagen en verwerkt, waardoor alle bewerkingen snel zijn.
  • Redis kan, afhankelijk van de instellingen, een WAL opslaan, waardoor de meest recente status snel kan worden gebackupt en hersteld.
  • Redis ondersteunt meerdere datatypen, zoals Set, Hash, Bit en List, waardoor het een hoge productiviteit biedt.
  • Redis heeft een grote community, waardoor men ervaringen, problemen en oplossingen kan delen en ontvangen.
  • Redis wordt al lange tijd ontwikkeld en onderhouden, waardoor het een betrouwbare stabiliteit heeft.

Nu ter zake

Stel je voor

Hoe zou het zijn als de cache van uw service aan de volgende twee voorwaarden voldoet?

  1. Vaak opgevraagde gegevens moeten actueel aan gebruikers worden aangeboden, maar omdat de updates onregelmatig plaatsvinden, moet de cache vaak worden bijgewerkt.
  2. De gegevens worden niet bijgewerkt, maar dezelfde cachegegevens worden vaak opgevraagd.

Het eerste geval kan bijvoorbeeld de real-time populariteitsranglijst van een webwinkel zijn. Wanneer de real-time populariteitsranglijst van een webwinkel is opgeslagen als sorted set, is het inefficiënt om deze bij elke toegang van een gebruiker tot de hoofdpagina vanuit Redis uit te lezen.
Het tweede geval betreft wisselkoersgegevens. Hoewel wisselkoersen ongeveer elke 10 minuten worden gepubliceerd, vinden de daadwerkelijke opvragingen zeer frequent plaats. Met name voor won-dollar, won-yen en won-yuan wordt de cache zeer frequent opgevraagd.
In dergelijke gevallen is het efficiënter dat de API-server een eigen lokale cache bijhoudt en, wanneer de gegevens veranderen, Redis opnieuw raadpleegt om de cache bij te werken.

Hoe kan men dit gedrag realiseren in een structuur van database, Redis en API-server?

Lukt dit niet met Redis PubSub?

Abonneer je op een kanaal waarmee je kunt ontvangen of er een update is wanneer je de cache gebruikt!

  • Er moet dan logica worden gebouwd die bij een update een bericht verstuurt.
  • Doordat er extra bewerkingen door PubSub worden toegevoegd, heeft dit invloed op de prestaties.

pubsub-write

pubsub-read

Wat als Redis de wijzigingen zelf detecteert?

Wat als je via Keyspace Notification meldingen ontvangt van de commando's voor specifieke sleutels?

  • Er bestaat de last dat de sleutels en commando's die voor updates worden gebruikt van tevoren moeten worden opgeslagen en gedeeld.
  • Zo kan voor sommige sleutels een eenvoudige Set het updatecommando zijn, terwijl voor andere sleutels LPush, RPush, SAdd of SRem het updatecommando is, wat het complex maakt.
  • Hierdoor neemt de kans op communicatiefouten tijdens de ontwikkeling en menselijke fouten in de code sterk toe.

Wat als je via Keyevent Notification meldingen per commando ontvangt?

  • Er is een abonnement nodig op alle commando's die voor updates worden gebruikt. Voor de binnenkomende sleutels is een passende filtering nodig.
  • Zo is het bij Del waarschijnlijk dat de client voor sommige binnenkomende sleutels geen lokale cache heeft.
  • Dit kan leiden tot onnodige verspilling van resources.

Daarom is Invalidation Message nodig!

Wat is een Invalidation Message?

Invalidation Messages is een concept dat wordt aangeboden als onderdeel van de Server Assisted Client-Side Cache die sinds Redis 6.0 beschikbaar is. Een Invalidation Message wordt via de volgende stroom verzonden.

  1. Veronderstel dat ClientB een key al eens heeft gelezen.
  2. ClientA stelt die key opnieuw in.
  3. Redis detecteert de wijziging en stuurt een Invalidation Message naar ClientB om hem te laten weten dat de cache moet worden gewist.
  4. ClientB ontvangt het bericht en neemt de passende maatregelen.

invalidation-message

Hoe gebruik je het?

Basisstructuur

Een client die verbonden is met Redis voert CLIENT TRACKING ON REDIRECT <client-id> uit, zodat deze invalidation messages ontvangt. De client die de berichten moet ontvangen, abonneert zich vervolgens met SUBSCRIBE __redis__:invalidate op de 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"

Implementatie! Implementatie! Implementatie!

Redigo + Ristretto

Als alleen zo wordt uitgelegd, is het onduidelijk hoe je dit in de praktijk in code moet gebruiken. Daarom ga ik eerst met redigo en ristretto een eenvoudige opzet maken.

Installeer eerst de twee dependencies.

  • 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). // het aantal sleutels waarvan de frequentie wordt bijgehouden (10M).
22		MaxCost:     1 << 30, // maximum cost of cache (1GB). // maximale kosten van de cache (1GB).
23		BufferItems: 64,      // number of keys per Get buffer. // aantal sleutels 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}

Eerst maak ik een RedisClient aan die ristretto en redigo bevat.

 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}

De code is wat ingewikkeld.

  • Voor Tracking wordt nog een verbinding gemaakt. Dit is een maatregel omdat PubSub de overige bewerkingen zou kunnen verstoren.
  • Het ID van de toegevoegde verbinding wordt opgevraagd, waarna de Tracking van de verbinding waarmee de gegevens worden opgevraagd naar die verbinding wordt omgeleid.
  • Vervolgens abonneert men zich op de invalidation messages.
  • De verwerking van de abonnementen is wat ingewikkeld. Omdat redigo de invalidation messages niet kan parsen, moet de respons vóór het parsen worden ontvangen en verwerkt.
 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}

De Get-methode zoekt eerst in ristretto, en haalt de waarde alleen uit Redis wanneer deze daar niet staat, zoals hieronder wordt weergegeven.

 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}

De code voor de test is hierboven weergegeven. Als u deze test uitvoert, zult u zien dat de waarde telkens opnieuw wordt bijgewerkt wanneer de gegevens in Redis veranderen.

Dit is echter te ingewikkeld. Bovendien is het voor uitbreiding naar een cluster noodzakelijk om Tracking op alle masters en replica's in te schakelen.

Rueidis

Als je Go gebruikt, hebben we de meest moderne en geavanceerde optie: rueidis. Laten we code schrijven die server assisted client side cache gebruikt in een Redis-clusteromgeving met rueidis.

Installeer eerst de dependency.

  • github.com/redis/rueidis

Schrijf vervolgens code die gegevens uit Redis opvraagt.

 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}

Om de client side cache in rueidis te gebruiken, hoeft men alleen DoCache aan te roepen. Daarbij geeft men op hoe lang de gegevens in de lokale cache moeten blijven. Wanneer DoCache daarna opnieuw wordt aangeroepen, worden de gegevens uit de lokale cache opgehaald. Uiteraard worden invalidation messages ook correct verwerkt.

Waarom niet redis-go?

Helaas ondersteunt redis-go de server assisted client side cache niet via de officiële API. Bovendien is er bij het aanmaken van een PubSub een nieuwe verbinding, en is er geen API om rechtstreeks toegang tot die verbinding te krijgen, waardoor ook het client-ID niet kan worden opgevraagd. Daarom heb ik geoordeeld dat de opzet met redis-go niet mogelijk is, en heb ik deze optie laten vallen.

Wat ziet dat er goed uit

Via de client side cache-structuur

  • Voor gegevens die vooraf kunnen worden voorbereid, kunnen de queries en het verkeer naar Redis via deze structuur tot een minimum worden beperkt, terwijl altijd actuele gegevens worden aangeboden.
  • Hiermee kan een soort CQRS-structuur worden gecreëerd, waardoor de leesprestaties sterk kunnen worden verbeterd.

cqrs

Hoeveel beter is het geworden?

Deze structuur wordt momenteel daadwerkelijk in productie gebruikt, dus heb ik de latentie van twee API's kort onderzocht. Ik vraag uw begrip dat ik dit slechts zeer abstract kan weergeven.

  1. Eerste API
    1. Bij de eerste opvraging: gemiddeld 14,63 ms
    2. Bij latere opvragingen: gemiddeld 2,82 ms
    3. Gemiddeld verschil: 10,98 ms
  2. Tweede API
    1. Bij de eerste opvraging: gemiddeld 14,05 ms
    2. Bij latere opvragingen: gemiddeld 1,60 ms
    3. Gemiddeld verschil: 11,57 ms

Er was een extra latentieverbetering van maximaal ongeveer 82%!

Ik verwacht dat de volgende verbeteringen zijn opgetreden.

  • Overslaan van de netwerkcommunicatie tussen client en Redis en besparing op verkeer
  • Vermindering van het aantal leescommando's dat Redis zelf moet uitvoeren
    • Dit heeft ook het effect dat de schrijfprestaties verbeteren.
  • Minimalisatie van het parsen van het Redis-protocol
    • Het parsen van het Redis-protocol kost ook niet niets. Het verminderen daarvan is een grote kans.

Echter, alles is een afweging. Hiervoor hebben we minimaal de volgende twee zaken opgeofferd.

  • Implementatie, beheer en onderhoud van de client side cache-componenten zijn nodig
  • Daardoor neemt het gebruik van CPU en geheugen van de client toe

Conclusie

Persoonlijk was dit een bevredigend architectuurcomponent, en de latentie en de belasting van de API-server waren ook zeer laag. Ik denk dat ik in de toekomst, waar mogelijk, weer zo'n structuur voor de architectuur zal kiezen.