---
id: 1c72ae6efee059c2c727ea53b56bf6c5
author: snowmerak
title: 使用 Redis 客户端缓存提升响应性
description: 利用 Redis 服务器辅助的客户端缓存提升 API 响应性
language: zh
date: 2024-11-29T00:51:27.525566863Z
path: /blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502
category: database
---

## 什么是 Redis?

我想大多数人都对 Redis 并不陌生。不过，为了简要回顾，我们可以将其特性整理如下。

- 操作在**单线程**中执行，因此所有操作都具有**原子性**。
- 数据存储并计算于**In-Memory**之中，因此所有操作都很**快速**。
- Redis 可根据配置保存 **WAL**，从而能够快速地对最新状态进行**备份**和**恢复**。
- 支持 Set、Hash、Bit、List 等多种**数据类型**，因此具有较高的**生产力**。
- 拥有**庞大的社区**，可以**共享**各种经验、问题及解决方法。
- 经过**长期的开发与运营**，具有值得信赖的**稳定性**。

## 言归正传

### 试想一下？

假如您的服务缓存符合以下两个条件，情况会怎样呢？

1. 需要向用户提供频繁被查询的数据的最新状态，但由于更新不规律，必须频繁地更新缓存时
2. 数据不再更新，但需要频繁访问并查询同一缓存数据时

第一种情况可以考虑购物商城的实时热门排名。将购物商城的实时热门排名保存为 sorted set 时，若每当用户访问主页都要从 Redis 读取，则效率低下。
第二种情况是汇率数据：尽管汇率大约每 10 分钟公布一次，但实际查询却非常频繁。其中，对于韩元-美元、韩元-日元、韩元-人民币，缓存的查询尤为频繁。
在这些情况下，较为高效的做法是让 API 服务器在本地保存单独的缓存，并在数据发生变化时重新查询 Redis 以进行更新。

那么，如何才能在 数据库 - Redis - API 服务器 的架构中实现这样的操作呢？？

### 用 Redis PubSub 不行吗？

> 使用缓存时，订阅一个可以接收更新通知的频道吧！

- 那么就需要编写在更新时发送消息的逻辑。
- 由于 PubSub 会带来额外的操作，因此会影响性能。

![pubsub-write](/assets/images/redis-client-side-cache/01-pubsub-write.png)

![pubsub-read](/assets/images/redis-client-side-cache/01-pubsub-read.png)

### 那如果由 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 检测到变更，并向 ClientB 发布 Invalidation Message，通知 ClientB 清除缓存。
4. ClientB 接收该消息，并采取相应的措施。

![invalidation-message](/assets/images/redis-client-side-cache/02-invalidation-message.png)

### 该如何使用

#### 基本运行结构

连接到 Redis 的客户端执行 `CLIENT TRACKING ON REDIRECT <client-id>`，即可接收 invalidation message。而需要接收消息的客户端则通过 `SUBSCRIBE __redis__:invalidate` 订阅 invalidation message。

#### 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"
```

## 实现！实现！实现！

### Redigo + Ristretto

仅仅这样说明，在实际代码中该如何使用仍然有些含糊。因此，我们先用 `redigo` 和 `ristretto` 简单地搭建一个示例。

首先安装这两个依赖。

- `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,     // 用于跟踪键访问频率的键数量 (10M)。
		MaxCost:     1 << 30, // 缓存的最大成本 (1GB)。
		BufferItems: 64,      // 每个 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
}
```

首先，我们创建一个包含 ristretto 和 redigo 的简单 `RedisClient`。

```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))
		}
	}
}
```

代码稍显复杂。

- 为了进行 Tracking，我们再建立一个连接。这是考虑到 PubSub 可能会干扰其他操作而采取的措施。
- 查询新增连接的 ID，使查询数据的连接将 Tracking 重定向到该连接。
- 然后订阅 invalidation message。
- 处理订阅的代码略显复杂。由于 redigo 无法解析失效消息，因此需要在解析之前接收响应并进行处理。

```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
}
```

`Get` 方法如下所示，先查询 ristretto，如果没有，则从 Redis 中获取。

```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)
		}
	}
}
```

用于测试的代码如上所示。若您实际运行一下，就会发现每当 Redis 中的数据被更新时，值都会随之刷新。

但这种方式过于复杂。最重要的是，为了扩展到集群环境，必然需要对所有主节点或副本节点启用 Tracking。

### Rueidis

既然使用 Go 语言，我们便有了最现代且发展成熟的 `rueidis`。接下来，我们编写一段在使用 rueidis 的 Redis 集群环境中，使用 server assisted client side cache 的代码。

首先，安装依赖。

- `github.com/redis/rueidis`

然后编写从 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)
		}
	}
}
```

要使用 rueidis 的客户端缓存，只需调用 `DoCache`。然后，像指定本地缓存的保留时长那样将数据加入本地缓存；再次调用相同的 `DoCache` 时，便会从本地缓存中查询并获取数据。理所当然地，它也能正确处理失效消息。

### 为什么不用 redis-go？

遗憾的是，`redis-go` 并未通过官方 API 支持 server assisted client side cache。甚至在创建 PubSub 时新建连接，却没有直接访问该连接的 API，因此也无法获知 client id。基于此，我们判断 `redis-go` 无法完成此类配置，于是放弃了它。

## 很性感嘛

### 通过 client side cache 结构

- 如果是可以预先准备的数据，通过这种结构可以将对 Redis 的查询和流量降至最低，同时始终提供最新的数据。
- 借此可以构建一种 CQRS 结构，从而大幅提升读取性能。

![cqrs](/assets/images/redis-client-side-cache/003-cqrs.jpg)

### 变得有多性感？

实际上，现场正好在使用这种结构，因此我们对两个 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 协议也并非零成本。能够减少这部分开销是一个很大的机会。

但所有事物都是权衡取舍。为此，我们至少牺牲了以下两点。

- 需要实现、运维和维护客户端缓存管理要素
- 由此导致客户端的 CPU 和内存使用量增加

## 结论

就个人而言，这是一个令人满意的架构组成部分，延迟以及对 API 服务器的压力也都非常小。今后，如果可能，我仍希望以这种结构来构建架构。