GoSuda

Подобряване на отзивчивостта чрез клиентски кеш в Redis

By snowmerak views ...

Какво е Redis?

Предполагам, че малко хора не познават Redis. Въпреки това, за да го обобщя накратко, чрез няколко негови характеристики, може да ги представим по следния начин.

  • Операциите се изпълняват в една нишка, поради което всички операции притежават атомарност.
  • Данните се съхраняват и обработват In-Memory, поради което всички операции са бързи.
  • Redis може, в зависимост от опциите, да записва WAL, което позволява бързо архивиране и възстановяване на най-новото състояние.
  • Поддържа множество типове данни, като Set, Hash, Bit, List и други, което осигурява висока продуктивност.
  • Разполага с голяма общност, в която можете да споделяте и получавате различен опит, проблеми и решения.
  • Redis се разработва и поддържа от дълго време, поради което притежава надеждна стабилност.

Към същинската част

Представете си?

Какво би станало, ако кешът на вашата услуга отговаря на следните две условия?

  1. Често заявяваните данни трябва да се предоставят на потребителя в актуално състояние, но тъй като обновяванията са нередовни, кешът трябва да се обновява често.
  2. Данните не се обновяват, но към един и същ кеш често се осъществява достъп и заявки.

Първият случай може да бъде класclassa за събиране на данни за актуалната популярност на продуктите в онлайн магазин. Ако класацията за актуална популярност на магазина се съхранява като sorted set, четенето ѝ от Redis при всяко посещение на потребителя на началната страница е неефективно.
Вторият случай са данни за валутни курсове. Въпреки че курсовете се публикуват официално приблизително на всеки 10 минути, реалните заявки към тях са много чести. Особено за двойките вон-долар, вон-йена и вон-юан кешът се заявява изключително често.
В такива случаи е по-ефективно API сървърът да поддържа собствен локален кеш, който при промяна на данните да се обновява чрез повторно четене от Redis.

Как бихме могли да реализираме това поведение в структурата база данни - Redis - API сървър?

Не може ли през Redis PubSub?

Когато използваме кеш, нека се абонираме за канал, от който можем да получаваме информация за обновявания!

  • Тогава трябва да създадем логика, която изпраща съобщение при всяко обновяване.
  • Допълнителните операции, свързани с PubSub, оказват влияние върху производителността.

pubsub-write

pubsub-read

Какво, ако Redis сам засича промените?

Ами ако използваме Keyspace Notification и получаваме известия за командите, отнасящи се до съответния ключ?

  • Съществува неудобството, че трябва предварително да съхраняваме и споделяме ключовете и командите, използвани за обновяване.
  • Например, за някои ключове командата за обновяване е просто Set, за други е LPush, а може да бъдат и RPush, SAdd или SRem, което усложнява картината.
  • Това значително увеличава вероятността от комуникационни грешки в процеса на разработка и човешки грешки при писане на код.

Ами ако използваме Keyevent Notification и получаваме известия по команда?

  • Необходимо е абониране за всички команди, използвани за обновяване. Входящите ключове трябва да се филтрират подходящо.
  • Например, сред всички ключове, получени чрез Del, за някои клиентът вероятно няма локален кеш.
  • Това може да доведе до ненужно разхищение на ресурси.

Затова ни е необходимо Invalidation Message!

Какво е Invalidation Message?

Invalidation Messages е концепция, предоставяна от Redis 6.0 нататък като част от Server Assisted Client-Side Cache. Invalidation Message се предава по следния поток.

  1. Приемаме, че ClientB вече е прочел даден key веднъж.
  2. ClientA задава наново този key.
  3. Redis засича промяната и изпраща Invalidation Message до ClientB, за да му укаже да изчисти кеша.
  4. ClientB получава съобщението и предприема подходящи действия.

invalidation-message

Как се използва

Основна структура на работа

Клиент, свързан към Redis, изпълнява CLIENT TRACKING ON REDIRECT <client-id>, за да получава Invalidation Message. Клиентът, който трябва да получава съобщенията, се абонира за 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"

Реализация! Реализация! Реализация!

Redigo + Ristretto

Ако обясним само така, е неясно как всъщност да го използваме в код. Затова първо ще изградим прост пример с redigo и ristretto.

Първо инсталирайте двете зависимости.

  • 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,     // брой ключове, за които се следи честотата (10M).
22		MaxCost:     1 << 30, // максимална цена на кеша (1GB).
23		BufferItems: 64,      // брой ключове на всеки буфер за Get.
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}

Първо създаваме RedisClient, който съдържа ristretto и 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}

Кодът е малко сложен.

  • За да се извърши Tracking, се създава още една връзка. Това е мярка, предприета с оглед на възможността PubSub да пречи на други операции.
  • Идентификаторът на добавената връзка се получава, за да се пренасочи Tracking на връзката, която чете данните, към нея.
  • След това се абонираме за Invalidation Message.
  • Обработката на абонаментите е малко сложна. Тъй като redigo не парсва Invalidation Message, трябва да получим отговора преди парсване и да го обработим.
 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 първо търси стойността в ristretto, а ако я няма, я взема от Redis, както е показано по-горе.

 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}

Кодът за тестване е показан по-горе. Ако го изпробвате, ще видите, че стойността се обновява при всяка промяна на данните в Redis.

Въпреки това този подход е твърде сложен. Освен това, за да се мащабира в клъстер, е неизбежно да се активира Tracking за всички master и replica възли.

Rueidis

Щом работим на Go, разполагаме с най-модерната и развита библиотека rueidis. Ще напишем код, който използва server assisted client side cache в клъстерна среда на Redis чрез rueidis.

Първо инсталирайте зависимостта.

  • github.com/redis/rueidis

След това напишете код, който чете данни от 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}

За да използвате client side cache в rueidis, е достатъчно просто да извикате DoCache. След това, като зададете например колко дълго да се пази данните в локалния кеш, при повторно извикване на DoCache данните се извличат от локалния кеш. Разбира се, Invalidation Message също се обработва коректно.

Защо не redis-go?

За съжаление redis-go не поддържа server assisted client side cache чрез официалния API. Освен това при създаване на PubSub се открива нова връзка, до която няма директен API достъп, поради което не можем да получим client id. Затова приехме, че конфигурацията с redis-go е невъзможна, и я пропуснахме.

Вълнуващо

Чрез структурата на client side cache

  • Ако данните могат да бъдат подготвени предварително, чрез тази структура можем да минимизираме заявките и трафика към Redis, като същевременно винаги предоставяме актуални данни.
  • По този начин можем да създадем вид CQRS структура и значително да повишим производителността на четене.

cqrs

Колко по-добре стана?

Към момента в реална среда използваме точно такава структура, затова измерихме сравнително латентността на два API. Моля, имайте предвид, че мога да ги опиша само доста абстрактно.

  1. Първи API
    1. При първоначално извикване: средно 14.63ms
    2. При последващи извиквания: средно 2.82ms
    3. Средна разлика: 10.98ms
  2. Втори API
    1. При първоначално извикване: средно 14.05ms
    2. При последващи извиквания: средно 1.60ms
    3. Средна разлика: 11.57ms

Подобрението на латентността достигна до около 82% допълнително!

Очакваме да са настъпили следните подобрения:

  • Пропускане на мрежовата комуникация между клиента и Redis и спестяване на трафик
  • Намаляване на броя команди за четене, които самият Redis трябва да изпълни
    • Това има и ефект на повишаване на производителността при запис.
  • Минимизиране на парсването на Redis протокола
    • Парсването на Redis протокола също не е без разход. Намаляването му е голяма възможност.

Но всяко решение е компромис. За да постигнем това, пожертвахме поне следните две неща:

  • Необходимост от реализиране, поддръжка и текуща експлоатация на компонентите за управление на client side cache
  • Увеличено използване на CPU и памет от клиента в резултат на това

Заключение

Лично за мен това беше задоволителен архитектурен компонент, а натоварването на API сървъра и латентността бяха много ниски. Занапред бих искал, когато е възможно, да изграждаме архитектурата именно по този начин.