GoSuda

Reagálóképesség javítása Redis client-side cache segítségével

By snowmerak views ...

Mi az a Redis?

Úgy gondolom, kevesen vannak, akik nem ismerik a Redist. Mindazonáltal néhány jellemzőjét röviden megemlítve a következőképpen foglalhatjuk össze.

  • A műveletek egyetlen szálon kerülnek végrehajtásra, ezért minden művelet atomi (atomicitás).
  • Az adatok In-Memory módon tárolódnak és kerülnek feldolgozásra, ezért minden művelet gyors.
  • A Redis a beállításoktól függően képes WAL tárolására, így gyorsan menthető (biztonsági mentés) a legfrissebb állapot, és gyorsan helyreállítható.
  • Támogatja a Set, Hash, Bit, List és más többféle típust, ami magas produktivitást biztosít.
  • Nagy közösséggel rendelkezik, így számos tapasztalatot, problémát és megoldást oszthatunk meg egymással.
  • Hosszú ideje fejlesztik és üzemeltetik, így megbízható stabilitással rendelkezik.

Térjünk rá a lényegre

Képzeljük el!

Mi lenne, ha a szolgáltatásuk gyorsítótára a következő két feltételnek felelne meg?

  1. Gyakran lekérdezett adatokat kell a felhasználóknak a legfrissebb állapotban biztosítani, de a frissítések szabálytalanok, ezért gyakran kell frissíteni a gyorsítótárat.
  2. Az adatok nem frissülnek, de ugyanazt a gyorsítótárazott adatot gyakran kell elérni és lekérdezni.

Az első esetre példa lehet egy webáruház valós idejű népszerűségi rangsora. Ha a valós idejű népszerűségi rangsort sorted set-ként tároljuk, akkor nem hatékony, ha a Redisből minden alkalommal beolvassuk, amikor a felhasználó eléri a főoldalt.
A második eset a devizaárfolyam-adatokra vonatkozik: bár az árfolyamadatokat nagyjából 10 percenként teszik közzé, a tényleges lekérdezések rendkívül gyakran történnek. Különösen a won–dollár, won–jen és won–jüan árfolyamok gyorsítótárát kérdezik le nagyon gyakran.
Ezekben az esetekben hatékonyabb működést eredményez, ha az API-szerver rendelkezik saját, helyi gyorsítótárral, és amikor az adatok megváltoznak, újra lekérdezi és frissíti azokat a Redisből.

Hogyan valósítható meg ez a működés egy adatbázis – Redis – API-szerver felépítésben??

Nem működik a Redis PubSub?

Gyorsítótár használatakor iratkozzunk fel egy csatornára, amelyen értesítést kaphatunk a frissítésekről!

  • Ekkor olyan logikát kell létrehozni, amely frissítéskor üzenetet küld.
  • A PubSub többletműveletet jelent, ezért hatással van a teljesítményre.

pubsub-write

pubsub-read

Mi lenne, ha a Redis maga érzékelné a változásokat?

Mi lenne, ha a Keyspace Notification segítségével a adott kulcsra vonatkozó parancsokról kapnánk értesítést?

  • Szükséges előre tárolni és megosztani a frissítéshez használt kulcsokat és parancsokat, ami bonyolult.
  • Például egyes kulcsoknál a sima Set a frissítő parancs, másoknál az LPush, vagy az RPush, SAdd és SRem lehet a frissítő parancs, így az összetettség nő.
  • Ez a fejlesztés során jelentősen megnöveli a kommunikációs hibák és a kódolási emberi hibák valószínűségét.

Mi lenne, ha a Keyevent Notification segítségével parancsonkénti értesítéseket kapnánk?

  • Minden frissítéshez használt parancsra fel kell iratkozni. A beérkező kulcsokat megfelelően szűrni kell.
  • Például a Del parancsnál a beérkező kulcsok egy részéhez az adott kliensnek valószínűleg nincs helyi gyorsítótára.
  • Ez szükségtelen erőforrás-pazarláshoz vezethet.

Ezért van szükség az Invalidation Message-re!

Mi az Invalidation Message?

Az Invalidation Messages a Redis 6.0-tól bevezetett, szerver által támogatott kliensoldali gyorsítótárazás (Server Assisted Client-Side Cache) részeként elérhető fogalom. Az Invalidation Message a következő folyamat szerint kerül továbbításra.

  1. Feltételezzük, hogy a ClientB már egyszer beolvasta a key értékét.
  2. A ClientA új értéket állít be a key-hez.
  3. A Redis érzékeli a változást, és Invalidation Message-t küld a ClientB-nek, amellyel utasítja, hogy törölje a gyorsítótárat.
  4. A ClientB fogadja az üzenetet, és megteszi a megfelelő intézkedéseket.

invalidation-message

Hogyan kell használni?

Alapvető működési szerkezet

A Redishez kapcsolódó kliens a CLIENT TRACKING ON REDIRECT <client-id> parancs végrehajtásával éri el, hogy invalidation message-eket kapjon. Az üzeneteket fogadó kliens pedig a SUBSCRIBE __redis__:invalidate paranccsal iratkozik fel az invalidation message-ekre.

Alapértelmezett 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"

Implementáció! Implementáció! Implementáció!

Redigo + Ristretto

Ha csak így magyarázzuk, nem egyértelmű, hogyan kell használni a kódban. Ezért egyszerűen, először a redigo és a ristretto segítségével építjük fel.

Először telepítsük a két függőséget.

  • 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,     // a kulcsok száma, amelyek gyakoriságát követjük (10M).
22		MaxCost:     1 << 30, // a gyorsítótár maximális költsége (1GB).
23		BufferItems: 64,      // kulcsok száma Get pufferenként.
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}

Először létrehozunk egy RedisClient-et, amely a ristrettót és a redigót tartalmazza.

 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}

A kód kissé bonyolult.

  • A Tracking érdekében egy további kapcsolatot hozunk létre. Ez azért szükséges, mert figyelembe vesszük, hogy a PubSub zavarhatja a többi műveletet.
  • A hozzáadott kapcsolat azonosítóját lekérdezzük, és az adatokat lekérdező kapcsolaton a Tracking-et erre a kapcsolatra irányítjuk (REDIRECT).
  • Ezt követően feliratkozunk az invalidation message-ekre.
  • A feliratkozás kezelésének kódja kissé bonyolult. Mivel a redigo nem értelmezi az érvénytelenítési üzeneteket, a feldolgozás előtt a választ nyers formában kell fogadni és kezelni.
 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}

A Get metódus a következőképpen először a ristrettót kérdezi le, és ha ott nincs adat, akkor a Redisből olvassa be.

 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}

A teszteléshez szükséges kód a fentiekben látható. Ha kipróbálja, láthatja, hogy minden alkalommal, amikor az adat frissül a Redisben, az érték is újra frissül.

Ez azonban túlságosan bonyolult. Mindenekelőtt a klaszterre történő skálázáshoz szükségszerűen minden master, illetve replika esetében engedélyezni kell a Tracking funkciót.

Rueidis

Mivel Go nyelven dolgozunk, rendelkezésünkre áll a legmodernebb és legfejlettebb rueidis. Írjunk kódot a server assisted client side cache használatára rueidis-szel egy Redis-klaszter környezetben.

Először telepítsük a függőséget.

  • github.com/redis/rueidis

Ezután írjunk kódot, amely adatot kérdez le a Redisből.

 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}

A rueidisben a kliensoldali gyorsítótár használatához elegendő a DoCache hívása. Ezután a helyi gyorsítótárban tárolandó időtartamot is megadhatjuk, és ha ugyanezt a DoCache hívást ismét végrehajtjuk, az adatot a helyi gyorsítótárból kapjuk vissza. Természetesen az invalidation message-eket is megfelelően kezeli.

Miért nem redis-go?

A redis-go sajnos nem támogatja a server assisted client side cache-t hivatalos API-n keresztül. Sőt, a PubSub létrehozásakor új kapcsolatot nyit, és nincs olyan API, amely közvetlenül elérné ezt a kapcsolatot, így a client id sem kérdezhető le. Ezért úgy ítéltük meg, hogy a redis-go esetében a felépítés egyszerűen nem lehetséges, és ezt a lehetőséget elvetettük.

Ez már szexi

A client side cache struktúrájának segítségével

  • Ha az adatok előre elkészíthetők, ezzel a struktúrával minimalizálható a Redisre irányuló lekérdezések és forgalom száma, miközben mindig a legfrissebb adatokat tudjuk biztosítani.
  • Ezáltal egyfajta CQRS struktúra hozható létre, amellyel jelentősen növelhető az olvasási teljesítmény.

cqrs

Mennyivel lett szexibb?

Valójában a helyszínen éppen ilyen struktúrában használjuk, ezért két API egyszerű késleltetését vizsgáltuk meg. Kérjük, nézzék el, hogy csak nagyon elvont módon tudok erről írni.

  1. Első API
    1. Első lekérdezéskor: átlagosan 14,63 ms
    2. Későbbi lekérdezéseknél: átlagosan 2,82 ms
    3. Átlagos különbség: 10,98 ms
  2. Második API
    1. Első lekérdezéskor: átlagosan 14,05 ms
    2. Későbbi lekérdezéseknél: átlagosan 1,60 ms
    3. Átlagos különbség: 11,57 ms

Legfeljebb mintegy 82%-os további késleltetés-csökkenést értünk el!

Feltételezésünk szerint a következő javulások történhettek:

  • A kliens és a Redis közötti hálózati kommunikáció kihagyása és a forgalom megtakarítása
  • A Redisnek végrehajtandó olvasási parancsok számának csökkenése
    • Ez a írási teljesítményt is növeli.
  • A Redis-protokoll feldolgozásának (parsing) minimalizálása
    • A Redis-protokoll feldolgozása sem ingyenes. Ennek csökkentése nagy lehetőség.

Minden azonban kompromisszum. Ehhez legalább a következő két dolgot áldoztuk fel:

  • A kliensoldali gyorsítótár-kezelő elemek megvalósítását, üzemeltetését és karbantartását
  • Ebből adódóan a kliens CPU- és memóriahasználatának növekedését

Következtetés

Személy szerint elégedett vagyok ezzel az architektúra-elemmel, és a késleltetés, valamint az API-szerverre nehezedő terhelés is nagyon kicsi volt. Remélem, hogy a jövőben is lehetőség szerint ilyen struktúrával építjük fel az architektúrát.