GoSuda

Förbättra responsiviteten med client-side caching i Redis

By snowmerak views ...

Vad är Redis?

Jag antar att få människor inte känner till Redis. Men för att ändå kort nämna några av dess egenskaper kan de sammanfattas enligt följande.

  • Operationer utförs i en enda tråd, så alla operationer har atomicitet.
  • Data lagras och bearbetas In-Memory, så alla operationer är snabba.
  • Redis kan, beroende på alternativ, spara en WAL, vilket gör det möjligt att snabbt säkerhetskopiera och återställa det senaste tillståndet.
  • Redis stöder flera typer såsom Set, Hash, Bit och List, vilket ger hög produktivitet.
  • Redis har en stor community, vilket gör att man kan dela och få tillgång till olika erfarenheter, problem och lösningar.
  • Redis har utvecklats och drivits under lång tid, vilket ger en pålitlig stabilitet.

Så till huvudämnet

Föreställ dig

Tänk om cacheminnet i er tjänst uppfyller följande två villkor?

  1. Data som ofta efterfrågas måste tillhandahållas till användaren i aktuellt skick, men eftersom uppdateringarna sker oregelbundet måste cacheminnet uppdateras ofta.
  2. Data uppdateras inte, men samma cachedata efterfrågas ofta.

Det första fallet kan exemplifieras med en realtidsrankning av populära produkter i en nätbutik. Om den realtidsrankningen lagras som en sorted set blir det ineffektivt att läsa den från Redis varje gång en användare besöker startsidan.
Det andra fallet gäller valutakursdata. Även om valutakurserna publiceras med ungefär tio minuters intervall, sker själva uppslagningen mycket ofta. Särskilt för won-dollar, won-yen och won-yuan sker cacheuppslagningar mycket frekvent.
I sådana fall är det troligen mer effektivt att API-servern har en separat lokal cache och, när data ändras, läser om från Redis och uppdaterar den.

Hur kan man då implementera detta beteende i en arkitektur med databas, Redis och API-server??

Fungerar inte Redis PubSub?

När vi använder en cache borde vi prenumerera på en kanal där vi kan få besked om uppdateringar!

  • Då måste man skapa logik som skickar ett meddelande vid uppdatering.
  • Eftersom PubSub medför ytterligare operationer påverkar det prestandan.

pubsub-write

pubsub-read

Tänk om Redis själv upptäcker ändringar?

Om vi använder Keyspace Notification och får kommandonotifieringar för respektive nyckel?

  • Det krävs besvär med att i förväg lagra och dela vilka nycklar och kommandon som används för uppdateringar.
  • Till exempel kan en nyckel uppdateras med ett enkelt Set, medan en annan nyckel uppdateras med LPush, RPush, SAdd eller SRem, vilket gör det komplicerat.
  • Detta ökar avsevärt risken för kommunikationsmissar under utvecklingen och för mänskliga fel i kodningen.

Om vi använder Keyevent Notification och får notifieringar per kommando?

  • Det krävs prenumeration på alla kommandon som används för uppdateringar, och inkommande nycklar måste filtreras på lämpligt sätt.
  • Till exempel har klienten troligen ingen lokal cache för vissa av de nycklar som kommer via Del.
  • Detta kan leda till onödigt resursslöseri.

Därför behövs Invalidation Message!

Vad är Invalidation Message?

Invalidation Messages är ett koncept som tillhandahålls som en del av Server Assisted Client-Side Cache, införd i Redis 6.0. Invalidation Message förmedlas enligt följande flöde.

  1. Anta att ClientB redan har läst en key en gång.
  2. ClientA sätter den key på nytt.
  3. Redis upptäcker ändringen och publicerar ett Invalidation Message till ClientB för att informera ClientB om att rensa cacheminnet.
  4. ClientB tar emot meddelandet och vidtar lämpliga åtgärder.

invalidation-message

Hur använder man det?

Grundläggande struktur

Genom att en klient ansluten till Redis kör CLIENT TRACKING ON REDIRECT <client-id> får den ta emot invalidation messages. Den klient som ska ta emot meddelandena prenumererar sedan på dem med SUBSCRIBE __redis__:invalidate.

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"

Implementation! Implementation! Implementation!

Redigo + Ristretto

Om man bara förklarar det så här är det oklart hur man faktiskt ska använda det i kod. Därför ska vi först bygga upp något enkelt med redigo och ristretto.

Installera först de två beroendena.

  • 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,     // antal nycklar för att spåra frekvens (10M).
22		MaxCost:     1 << 30, // maximal kostnad för cachen (1GB).
23		BufferItems: 64,      // antal nycklar per Get-buffert.
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 skapar vi enkelt en RedisClient som innehåller ristretto och 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 är något komplicerad.

  • För att kunna göra Tracking öppnar vi ytterligare en anslutning. Detta görs för att PubSub inte ska störa annan verksamhet.
  • Vi hämtar ID:t för den tillagda anslutningen och låter Tracking omdirigeras till den anslutningen från den anslutning som läser data.
  • Därefter prenumererar vi på invalidation messages.
  • Koden för att hantera prenumerationen är något komplicerad. Eftersom redigo inte tolkar invalidation messages måste vi ta emot svaret före tolkning och hantera det därefter.
 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}

Metoden Get slår först upp värdet i ristretto och hämtar det från Redis om det inte finns där, enligt koden ovan.

 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}

Ovanstående är koden för testning. Om du kör den kommer du att se att värdet uppdateras varje gång data ändras i Redis.

Men detta är för komplicerat. Framför allt måste Tracking aktiveras för alla masters eller repliker för att kunna skalas upp i ett kluster.

Rueidis

Så länge man skriver i Go har vi rueidis, den mest moderna och välutvecklade lösningen. Låt oss skriva kod som använder server assisted client side cache i en Redis-klusermiljö med rueidis.

Installera först beroendet.

  • github.com/redis/rueidis

Skriv sedan kod som läser data från 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}

För att använda client side cache i rueidis behöver man bara anropa DoCache. Därefter lägger den till data i den lokala cachen, inklusive hur länge den ska behållas lokalt, och när DoCache anropas på nytt hämtas data från den lokala cachen. Givetvis hanteras invalidation messages också korrekt.

Varför inte redis-go?

redis-go stöder tyvärr inte server assisted client side cache via sitt officiella API. Dessutom saknas ett API som direkt kan komma åt anslutningen när en PubSub skapar en ny anslutning, så client id kan inte heller hämtas. Därför bedömde vi att konfigurationen inte var möjlig med redis-go och valde att inte använda det.

Vad snyggt

Genom client side cache-strukturen

  • Om data kan förberedas i förväg kan vi med denna struktur minimera frågor och trafik mot Redis och ändå alltid tillhandahålla det senaste datat.
  • På så sätt kan vi skapa en slags CQRS-struktur och kraftigt förbättra läsprestandan.

cqrs

Hur mycket snyggare blev det?

Vi använder faktiskt denna struktur i produktion just nu, så vi har mätt den enkla latensen för två API:er. Jag ber om överseende med att det endast kan beskrivas mycket översiktligt.

  1. Första API:et
    1. Vid första anropet: i genomsnitt 14,63 ms
    2. Vid efterföljande anrop: i genomsnitt 2,82 ms
    3. Genomsnittlig skillnad: 10,98 ms
  2. Andra API:et
    1. Vid första anropet: i genomsnitt 14,05 ms
    2. Vid efterföljande anrop: i genomsnitt 1,60 ms
    3. Genomsnittlig skillnad: 11,57 ms

Den extra latensförbättringen var som mest cirka 82 procent!

Vi förväntar oss att följande förbättringar har bidragit:

  • Nätverkskommunikation mellan klient och Redis hoppas över, vilket sparar trafik.
  • Antalet läskommandon som Redis själv behöver utföra minskar.
    • Detta har också effekten att skrivprestandan ökar.
  • Parsning av Redis-protokollet minimeras.
    • Att parsa Redis-protokollet har inte noll kostnad. Att kunna minska detta är en stor möjlighet.

Men all förbättring är en avvägning. För detta har vi åtminstone offrat följande två saker.

  • Implementering, drift och underhåll av komponenter för client side cache krävs.
  • Detta leder till ökad CPU- och minnesanvändning hos klienten.

Slutsats

Personligen är detta en tillfredsställande arkitekturkomponent, och både latensen och belastningen på API-servern har varit mycket låg. Jag tycker att vi framöver, om möjligt, bör bygga arkitekturen med en sådan struktur.