GoSuda

Повышение отзывчивости с помощью клиентского кэша Redis

By snowmerak views ...

Что такое Redis?

Полагаю, что мало кто не знаком с Redis. Тем не менее, кратко напомню его особенности, которые можно обобщить следующим образом.

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

Теперь к основной теме

Представьте себе?

Что, если кэш вашего сервиса соответствует следующим двум условиям?

  1. Часто запрашиваемые данные необходимо предоставлять пользователю в актуальном состоянии, но их обновления происходят нерегулярно, поэтому кэш приходится часто обновлять.
  2. Данные не обновляются, но к одним и тем же данным кэша требуется часто обращаться.

Первый случай может быть примером рейтинга популярных товаров интернет-магазина в реальном времени. Если хранить рейтинг популярных товаров интернет-магазина в виде 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 — это концепция, предоставляемая в рамках Server Assisted Client-Side Cache, добавленного в Redis 6.0. Invalidation Message передается по следующему процессу.

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

invalidation-message

Как это использовать

Базовая структура работы

Клиент, подключенный к Redis, выполняет CLIENT TRACKING ON REDIRECT <client-id>, чтобы получать invalidation message. Клиент, которому необходимо получать сообщения, подписывается на SUBSCRIBE __redis__:invalidate для получения invalidation message.

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 может помешать другим операциям.
  • Определяется ID добавленного соединения, и для соединения, которое будет запрашивать данные, 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 на всех мастерах и репликах.

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 данные будут получены из локального кэша. Разумеется, сообщения об инвалидации также обрабатываются корректно.

Почему не 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.63 мс
    2. При последующих запросах: в среднем 2.82 мс
    3. Средняя разница: 10.98 мс
  2. Второй API
    1. При первом запросе: в среднем 14.05 мс
    2. При последующих запросах: в среднем 1.60 мс
    3. Средняя разница: 11.57 мс

Максимальное дополнительное улучшение задержки составило около 82%!

Предполагаю, что были достигнуты следующие улучшения.

  • Исключение сетевого взаимодействия между клиентом и Redis и экономия трафика
  • Уменьшение количества команд чтения, которые должен выполнять сам Redis
    • Это также повышает производительность записи.
  • Минимизация парсинга протокола Redis
    • Парсинг протокола Redis тоже имеет ненулевую стоимость, поэтому его сокращение — большая возможность.

Однако все является компромиссом. Для этого мы пожертвовали как минимум двумя вещами.

  • Необходимость реализации, эксплуатации и сопровождения компонентов управления client side cache
  • Связанный с этим рост использования CPU и памяти на стороне клиента

Заключение

Лично я считаю, что это был удовлетворительный архитектурный компонент: задержка и нагрузка на API-сервер были очень низкими. Я надеюсь и дальше, по возможности, строить архитектуру с использованием такой структуры.