--- id: 1c72ae6efee059c2c727ea53b56bf6c5 author: snowmerak title: Melhorando a responsividade com cache do lado do cliente no Redis description: Melhoria da Responsividade de APIs Utilizando Client Side Cache Assistido pelo Servidor Redis language: pt date: 2024-11-29T00:51:27.525566863Z path: /blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502 category: database --- ## O que é o Redis? Acredito que poucas pessoas desconhecem o Redis. Ainda assim, para mencionar brevemente algumas de suas características antes de prosseguir, elas podem ser resumidas da seguinte forma. - As operações são executadas em uma **thread única**, de modo que todas as operações possuem **atomicidade**. - Os dados são armazenados e processados **In-Memory**, de modo que todas as operações são **rápidas**. - O Redis pode armazenar o **WAL** conforme as opções configuradas, permitindo **realizar backup** e **recuperar** rapidamente o estado mais recente. - Por suportar **diversos tipos** de dados, como Set, Hash, Bit e List, possui alta **produtividade**. - Por possuir uma **grande comunidade**, é possível **compartilhar** e receber diversas experiências, problemas e soluções. - Por ter sido **desenvolvido e operado** por um longo período, possui **estabilidade** confiável. ## Agora, ao assunto principal ### Imagine? E se o cache do seu serviço atendesse às duas condições a seguir? 1. Dados consultados com frequência precisam ser fornecidos aos usuários em seu estado mais recente, mas, como as atualizações são irregulares, o cache precisa ser atualizado com frequência 2. Os dados não são atualizados, mas o mesmo dado de cache é acessado e consultado com muita frequência O primeiro caso pode ser exemplificado pelo ranking de popularidade em tempo real de um shopping online. Quando o ranking de popularidade em tempo real é armazenado como sorted set, ler no Redis sempre que um usuário acessa a página principal é ineficiente. O segundo caso se refere a dados de taxa de câmbio: mesmo que os dados sejam divulgados aproximadamente a cada 10 minutos, as consultas ocorrem com muita frequência. Em especial, para won-dólar, won-iene e won-yuan, o cache é consultado com grande frequência. Nesses casos, é mais eficiente que o servidor de API mantenha um cache local separado e, quando os dados forem alterados, consulte novamente o Redis para atualizá-lo. Então, como implementar esse comportamento na estrutura banco de dados - Redis - servidor de API? ### Não seria possível com o Redis PubSub? > Ao usar o cache, vamos assinar um canal que permita receber notificações de atualização! - Nesse caso, é necessário criar uma lógica que envie uma mensagem no momento da atualização. - Como há operações adicionais decorrentes do PubSub, o desempenho é afetado. ![pubsub-write](/assets/images/redis-client-side-cache/01-pubsub-write.png) ![pubsub-read](/assets/images/redis-client-side-cache/01-pubsub-read.png) ### E se o próprio Redis detectasse as alterações? > E se usássemos Keyspace Notification para receber notificações de comandos referentes a determinada chave? - É preciso definir previamente e compartilhar as chaves e os comandos usados nas atualizações, o que é trabalhoso. - Por exemplo, para algumas chaves o comando de atualização é um simples Set, enquanto para outras é LPush, RPush, SAdd ou SRem, o que torna tudo mais complexo. - Isso aumenta consideravelmente a possibilidade de falhas de comunicação durante o desenvolvimento e de erros humanos na codificação. > E se usássemos Keyevent Notification para receber notificações por comando? - É necessário assinar todos os comandos usados nas atualizações, além de filtrar adequadamente as chaves recebidas. - Por exemplo, entre todas as chaves recebidas por Del, algumas podem não ter cache local no respectivo cliente, o que é altamente provável. - Isso pode levar ao desperdício desnecessário de recursos. ### Por isso, o necessário é a Invalidation Message! #### O que é a Invalidation Message? As Invalidation Messages são um conceito fornecido como parte do Server Assisted Client-Side Cache, introduzido no Redis 6.0. A Invalidation Message é entregue pelo seguinte fluxo. 1. Suponha que o ClientB já tenha lido uma vez a key. 2. O ClientA define novamente essa key. 3. O Redis detecta a alteração e publica uma Invalidation Message para o ClientB, informando-o para limpar o cache. 4. O ClientB recebe a mensagem e toma as medidas apropriadas. ![invalidation-message](/assets/images/redis-client-side-cache/02-invalidation-message.png) ### Como utilizar #### Estrutura básica de funcionamento Um cliente conectado ao Redis executa `CLIENT TRACKING ON REDIRECT ` para receber as invalidation messages. Em seguida, o cliente que deve receber as mensagens assina `SUBSCRIBE __redis__:invalidate` para recebê-las. #### Default 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" ``` ## Implementação! Implementação! Implementação! ### Redigo + Ristretto Explicar apenas dessa forma torna difícil saber como usar isso de fato no código. Por isso, vamos primeiro montar uma estrutura simples com `redigo` e `ristretto`. Primeiro, instale as duas dependências. - `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, // número de chaves para rastrear a frequência (10M). MaxCost: 1 << 30, // custo máximo do cache (1GB). BufferItems: 64, // número de chaves por buffer de Get. }) 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 } ``` Primeiro, criamos de forma simples um `RedisClient` que contém ristretto e redigo. ```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)) } } } ``` O código está um pouco complexo. - Para realizar o Tracking, abrimos mais uma conexão. Isso foi feito considerando que o PubSub poderia interferir em outras operações. - Consultamos o ID da conexão adicionada e configuramos o Tracking da conexão que consulta os dados para redirecionar para essa conexão. - Em seguida, assinamos a invalidation message. - O código que processa a assinatura é um pouco complexo. Como o redigo não faz o parsing das invalidation messages, é preciso receber a resposta antes do parsing e tratá-la. ```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 } ``` O método `Get` primeiro consulta o ristretto e, caso não encontre o valor, o obtém do Redis, conforme o código abaixo. ```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) } } } ``` O código para teste é o acima. Se você testar, poderá verificar que o valor é atualizado sempre que os dados são alterados no Redis. No entanto, isso é muito complexo. Acima de tudo, para escalar em um cluster, é inevitável habilitar o Tracking em todos os masters ou réplicas. ### Rueidis Já que usamos Go, temos à disposição o `rueidis`, a opção mais moderna e avançada. Vamos escrever um código que utiliza o server assisted client side cache em um ambiente de cluster Redis com o rueidis. Primeiro, instale a dependência. - `github.com/redis/rueidis` Em seguida, escrevemos o código que consulta os dados no Redis. ```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) } } } ``` No rueidis, para usar o client side cache basta chamar `DoCache`. Assim, como ao adicionar ao cache local, é possível definir por quanto tempo os dados serão mantidos localmente, e ao chamar `DoCache` novamente, os dados são consultados e obtidos do cache local. Naturalmente, as invalidation messages também são tratadas corretamente. ### Por que não redis-go? Infelizmente, o `redis-go` não oferece suporte oficial ao server assisted client side cache. Além disso, ao criar um PubSub, ele abre uma nova conexão sem fornecer uma API para acessá-la diretamente, de modo que não é possível obter o client id. Por isso, consideramos que a configuração do `redis-go` seria inviável e o descartamos. ## Que elegante ### Por meio da estrutura de client side cache - Para dados que podem ser preparados antecipadamente, essa estrutura permite minimizar consultas e tráfego para o Redis, mantendo sempre os dados mais recentes. - Com isso, é possível criar uma espécie de estrutura CQRS e aumentar drasticamente o desempenho de leitura. ![cqrs](/assets/images/redis-client-side-cache/003-cqrs.jpg) ### Quanto mais elegante ficou? Como, por coincidência, estamos usando essa estrutura em produção, medimos a latência aproximada de duas APIs. Peço desculpas por apresentar apenas dados muito abstratos. 1. Primeira API 1. Na primeira consulta: média de 14,63 ms 2. Nas consultas seguintes: média de 2,82 ms 3. Diferença média: 10,98 ms 2. Segunda API 1. Na primeira consulta: média de 14,05 ms 2. Nas consultas seguintes: média de 1,60 ms 3. Diferença média: 11,57 ms Houve uma melhoria adicional de latência de até cerca de 82%! Esperamos que tenham ocorrido as seguintes melhorias: - Eliminação da comunicação de rede entre o cliente e o Redis e economia de tráfego - Redução do número de comandos de leitura que o próprio Redis precisa executar - Isso também tem o efeito de melhorar o desempenho de escrita. - Redução do parsing do protocolo Redis - O parsing do protocolo Redis também não tem custo zero. Poder reduzi-lo é uma grande oportunidade. No entanto, toda decisão envolve trade-offs. Para isso, sacrificamos pelo menos os dois pontos a seguir. - Necessidade de implementar, operar e manter os componentes de gerenciamento do client side cache - Aumento do uso de CPU e memória no cliente decorrente disso ## Conclusão Pessoalmente, foi um componente de arquitetura satisfatório, e a latência e a pressão sobre o servidor de API também foram bastante baixas. Penso em continuar estruturando a arquitetura dessa forma sempre que possível.