--- id: 1c72ae6efee059c2c727ea53b56bf6c5 author: snowmerak title: Reagálóképesség javítása Redis client-side cache segítségével description: Redis Server Assisted Client Side Cache segítségével történő API válaszidő-javulás language: hu date: 2024-11-29T00:51:27.525566863Z path: /blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502 category: database --- ## 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](/assets/images/redis-client-side-cache/01-pubsub-write.png) ![pubsub-read](/assets/images/redis-client-side-cache/01-pubsub-read.png) ### 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](/assets/images/redis-client-side-cache/02-invalidation-message.png) ### Hogyan kell használni? #### Alapvető működési szerkezet A Redishez kapcsolódó kliens a `CLIENT TRACKING ON REDIRECT ` 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 ```shell # client 1 > SET a 100 ``` ```shell # client 3 > CLIENT ID 12 > SUBSCRIBE __redis__:invalidate 1) "subscribe" 2) "__redis__:invalidate" 3) (integer) 1 ``` ```shell # client 2 > CLIENT TRACKING ON REDIRECT 12 > GET a # tracking ``` ```shell # client 1 > SET a 200 ``` ```shell # client 3 1) "message" 2) "__redis__:invalidate" 3) 1) "a" ``` #### Broadcasting tracking ```shell # client 3 > CLIENT ID 12 > SUBSCRIBE __redis__:invalidate 1) "subscribe" 2) "__redis__:invalidate" 3) (integer) 1 ``` ```shell # client 2 CLIENT TRACKING ON BCAST PREFIX cache: REDIRECT 12 ``` ```shell # client 1 > SET cache:name "Alice" > SET cache:age 26 ``` ```shell # client 3 1) "message" 2) "__redis__:invalidate" 3) 1) "cache:name" 1) "message" 2) "__redis__:invalidate" 3) 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` ```go package main import ( "context" "fmt" "log/slog" "time" "github.com/dgraph-io/ristretto" "github.com/gomodule/redigo/redis" ) type RedisClient struct { conn redis.Conn cache *ristretto.Cache[string, any] addr string } func NewRedisClient(addr string) (*RedisClient, error) { cache, err := ristretto.NewCache(&ristretto.Config[string, any]{ NumCounters: 1e7, // a kulcsok száma, amelyek gyakoriságát követjük (10M). MaxCost: 1 << 30, // a gyorsítótár maximális költsége (1GB). BufferItems: 64, // kulcsok száma Get pufferenként. }) if err != nil { return nil, fmt.Errorf("failed to generate cache: %w", err) } conn, err := redis.Dial("tcp", addr) if err != nil { return nil, fmt.Errorf("failed to connect to redis: %w", err) } return &RedisClient{ conn: conn, cache: cache, addr: addr, }, nil } func (r *RedisClient) Close() error { err := r.conn.Close() if err != nil { return fmt.Errorf("failed to close redis connection: %w", err) } return nil } ``` Először létrehozunk egy `RedisClient`-et, amely a ristrettót és a redigót tartalmazza. ```go func (r *RedisClient) Tracking(ctx context.Context) error { psc, err := redis.Dial("tcp", r.addr) if err != nil { return fmt.Errorf("failed to connect to redis: %w", err) } clientId, err := redis.Int64(psc.Do("CLIENT", "ID")) if err != nil { return fmt.Errorf("failed to get client id: %w", err) } slog.Info("client id", "id", clientId) subscriptionResult, err := redis.String(r.conn.Do("CLIENT", "TRACKING", "ON", "REDIRECT", clientId)) if err != nil { return fmt.Errorf("failed to enable tracking: %w", err) } slog.Info("subscription result", "result", subscriptionResult) if err := psc.Send("SUBSCRIBE", "__redis__:invalidate"); err != nil { return fmt.Errorf("failed to subscribe: %w", err) } psc.Flush() for { msg, err := psc.Receive() if err != nil { return fmt.Errorf("failed to receive message: %w", err) } switch msg := msg.(type) { case redis.Message: slog.Info("received message", "channel", msg.Channel, "data", msg.Data) key := string(msg.Data) r.cache.Del(key) case redis.Subscription: slog.Info("subscription", "kind", msg.Kind, "channel", msg.Channel, "count", msg.Count) case error: return fmt.Errorf("error: %w", msg) case []interface{}: if len(msg) != 3 || string(msg[0].([]byte)) != "message" || string(msg[1].([]byte)) != "__redis__:invalidate" { slog.Warn("unexpected message", "message", msg) continue } contents := msg[2].([]interface{}) keys := make([]string, len(contents)) for i, key := range contents { keys[i] = string(key.([]byte)) r.cache.Del(keys[i]) } slog.Info("received invalidation message", "keys", keys) default: slog.Warn("unexpected message", "type", fmt.Sprintf("%T", msg)) } } } ``` 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. ```go func (r *RedisClient) Get(key string) (any, error) { val, found := r.cache.Get(key) if found { switch v := val.(type) { case int64: slog.Info("cache hit", "key", key) return v, nil default: slog.Warn("unexpected type", "type", fmt.Sprintf("%T", v)) } } slog.Info("cache miss", "key", key) val, err := redis.Int64(r.conn.Do("GET", key)) if err != nil { return nil, fmt.Errorf("failed to get key: %w", err) } r.cache.SetWithTTL(key, val, 1, 10*time.Second) return val, nil } ``` 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. ```go package main import ( "context" "log/slog" "os" "os/signal" "time" ) func main() { ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) defer cancel() client, err := NewRedisClient("localhost:6379") if err != nil { panic(err) } defer client.Close() go func() { if err := client.Tracking(ctx); err != nil { slog.Error("failed to track invalidation message", "error", err) } }() ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() done := ctx.Done() for { select { case <-done: slog.Info("shutting down") return case <-ticker.C: v, err := client.Get("key") if err != nil { slog.Error("failed to get key", "error", err) return } slog.Info("got key", "value", v) } } } ``` 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. ```go package main import ( "context" "log/slog" "os" "os/signal" "time" "github.com/redis/rueidis" ) func main() { ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) defer cancel() client, err := rueidis.NewClient(rueidis.ClientOption{ InitAddress: []string{"localhost:6379"}, }) if err != nil { panic(err) } ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() done := ctx.Done() for { select { case <-done: slog.Info("shutting down") return case <-ticker.C: const key = "key" resp := client.DoCache(ctx, client.B().Get().Key(key).Cache(), 10*time.Second) if resp.Error() != nil { slog.Error("failed to get key", "error", resp.Error()) continue } i, err := resp.AsInt64() if err != nil { slog.Error("failed to convert response to int64", "error", err) continue } switch resp.IsCacheHit() { case true: slog.Info("cache hit", "key", key) case false: slog.Info("missed key", "key", key) } slog.Info("got key", "value", i) } } } ``` 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](/assets/images/redis-client-side-cache/003-cqrs.jpg) ### 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.