大家好，我是渔夫子。今天聊聊在项目中通过优化 redis 写入而降低 cpu 使用率的一次经历。
01 背景
本文是项目中基于 redis 记录实时请求量的一个功能，因流量上涨造成 redis 服务器的 CPU 高于 80%而触发了自动报警机制，经分析将实时写入 redis 的方式变更成批量写入的方式，从而将 CPU 使用率降低了 30%左右的经历。
具体业务需求是这样的：我们会将接受到的请求按地域属性进行划分。目标是针对具体的国家请求进行总数的控制。当达到预设的最大请求数时，就不再处理该流量，直接给客户端返回 204 响应。如果没有达到最大请求数，则需要对实时请求数+1。如下图所示：
02 实现版本一
第一个版本很简单，就是将最大值存放在 redis 中，然后按天的维度记录每个国家流量的实时请求数量。每次流量来了之后，先查询出该国家流量的最大值，以及当天的实时请求数，然后做比较，如果实时数已经超过了最大值，就直接返回，否则就对实时数进行+1 操作即可。下面我们以来自中国（用 CN 表示）流量为例进行说明。
首先，我们存在 redis 中的 key 的规则如下：
第一个版本的实现代码如下：
func HasExceedLimitReq() bool { key := "CN:max:req"
maxReq := redis.Get(key)
day := time.Now().Format("20060102") dailyKey := "CN:"+day+":req" dailyReq := redis.Get(dailyKey)
if dailyReq > maxReq { return true }
redis.Incr(dailyKey, dailyReq) redis.Expire(dailyKey, 7*24*time.Hour) return false}
复制代码
在上面的实现中，对于 dailyKey 我们不需要长期保留，实际上只要过了当天，该 key 的值就没用了，出于查询历史数据的原因，我们就设置了 7 天的有效期。 但 redis 的 Incr 操作不带过期时间，所以就在 Incr 操作后增加了一个 Expire 的操作。
好了，我们看下这个实现会有什么问题。首先逻辑上没什么问题。当一个请求进来之后，在没有超量的情况下，我们会对 redis 有 4 次操作：两次查询操作和两次写操作（incr 和 expire）。也就是说，redis 扛的 QPS 是流量本身的 4 倍。 如果当流量 QPS 不断增长的时候，比如达到了 10 万，那么 redis 收到的请求量就是 40 万。redis 的 CPU 消耗自然也就上来了。
那么我们看看哪些地方是可以优化的呢？首先就是 Expire 操作看起来不是每次都需要，理论上只要设置一次过期时间就可以了，不需要每次都设置，这样就可以减少一次写操作。如下实现版本二
03 实现版本二：减少 Expire 的执行次数
我们通过使用一个 hasUpdateExpire 的 map 类型，来记录某个 key 是否已经被设置了有效期的标识。如下：
var hasUpdateExpire = make(map[string]struct{}) //全局变量
func HasExceedLimitReq() bool { key := "CN:max:req"
maxReq := redis.Get(key)
day := time.Now().Format("20060102") dailyKey := "CN:"+day+":req" dailyReq := redis.Get(dailyKey)
if dailyReq > maxReq { return true }
redis.Incr(dailyKey, dailyReq) if hasUpdateExpire[dailyKey]; !ok { redis.Expire(dailyKey, 7*24*time.Hour) hasUpdateExpire[dailyKey] = struct{}{} } return false}
复制代码
我们知道在 Go 中，map 是非并发安全的。那么下面这段代码是存在并发安全的：
if hasUpdateExpire[dailyKey]; !ok { redis.Expire(dailyKey, 7*24*time.Hour) hasUpdateExpire[dailyKey] = struct{}{} }
复制代码
也就是说有可能有多个协程同时执行到了if hasUpdateExpire[dailyKey]这里，并且都获取到了 ok 为 false 的值，那么这时就会有多个协程都会执行如下两行代码：
redis.Expire(dailyKey, 7*24*time.Hour)hasUpdateExpire[dailyKey] = struct{}{}
复制代码
但这里根据我们业务的场景，即使多执行几次 Expire 操作也没关系，在 QPS 高的情况下，比起总的请求次数来说多设置 expire 几次可以忽略。
那如果 qps 再继续增加怎么办？那就是异步批量写入。这种写入方式适合于那种对计数不要求准确的场景。我们来看看版本三。
04 实现版本三：异步批量写入
在该版本中，我们的技术不直接写入 redis，而是写在内存缓存中，即一个全局变量中，同时启动一个定时器，每个一段时间就将内存中的数据批量写入到 redis 中。如下图所示：
所以 我们定义了如下数据结构：
import ( "sync" "time"
"github.com/go-redis/redis")
const ( DefaultExpiration = 86400 * time.Second * 7)
type CounterCache struct { rwMu sync.RWMutex redisClient redis.Cmdable
countCache map[string]int64 hasUpdateExpire map[string]struct{}}
func NewCounterCache(redisClient redis.Cmdable) *CounterCache { c := &CounterCache{ redisClient: redisClient, countCache: make(map[string]int64), } go c.startFlushTicker() return c}
func (c *CounterCache) IncrBy(key string, value int64) int64 { val := c.incrCacheBy(key, value) redisCount, _ := c.redisClient.Get(key).Int64() return val + redisCount}
func (c *CounterCache) incrCacheBy(key string, value int64) int64 { c.rwMu.Lock() defer c.rwMu.Unlock() count := c.countCache[key] count += value c.countCache[key] = count return count}
func (c *CounterCache) Get(key string) (int64, error) { cacheVal := c.get(key) redisValue, err := c.redisClient.Get(key).Int64() if err != nil && err != redis.Nil { return cacheVal, err }
return redisValue + cacheVal, nil}
func (c *CounterCache) get(key string) int64 { c.rwMu.RLock() defer c.rwMu.RUnlock() return c.countCache[key]}
func (c *CounterCache) startFlushTicker() { ticker := time.NewTicker(time.Second * 5) for { select { case <-ticker.C: c.flush() } }}
func (c *CounterCache) flush() { var oldCountCache map[string]int64 c.rwMu.Lock() oldCountCache = c.countCache c.countCache = make(map[string]int64) c.rwMu.Unlock()
for key, value := range oldCountCache { c.redisClient.IncrBy(key, value) if _, ok := c.hasUpdateExpire[key]; !ok { err := c.redisClient.Expire(key, DefaultExpiration) if err == nil { c.hasUpdateExpire[key] = struct{}{} } } }}
复制代码
这里主要的思想就是在写入数据的时候先暂存在结构体的 countCache 中。然后每个 CounterCache 实例都会启动一个定时器 ticker，该定时器每隔一段时间就将 countCache 中的数据更新到 redis 中。我们看下这的使用方式：
package main
import ( "net/http" "sync" "time"
"github.com/go-redis/redis")
var counterCache *CounterCache
func main() { redisClient := redis.NewClient(&redis.Options{ Addr: "127.0.0.1:6379", Password: "", }) counterCache = NewCounterCache(redisClient)
http.HandleFunc("/", IndexHandler) http.ListenAndServe(":8080", nil)}
func IndexHandler(w http.ResponseWriter, r *http.Request) { if HasExceedLimitReq() { return } //处理正常逻辑}
func HasExceedLimitReq() bool { maxKey := "CN:max:req" maxCount, _ := counterCache.Get(maxKey)
dailyKey := "CN:" + time.Now().Format("20060102") + ":req" dailyCount, _ := counterCache.Get(dailyKey)
if dailyCount > maxCount { return true }
counterCache.IncrBy(dailyKey, 1) return false}
复制代码
这里的使用场景就是在对计数不要求准确的情况下使用的。比如说如果服务器异常退出了，那么暂存在 countCache 中还没来得及刷新到 redis 中的数据就会造成丢失。
另外一点需要注意的是就是 countCache 变量是一个 map，我们知道，在 Go 中 map 是非并发安全的操作，所以要注意加读写锁。
05 总结
随着服务 qps 的增长，我们在不限制 qps 的前提下，各种资源的使用率都会增长。我们的优化思路就是减少不必要的写次数、由实时写更改成批量写的思想，从而达到减少对 redis 操作的目的。这种计数方式使用的场景是在对计数要求不那么准确的情况，例如视频的播放量、微博大 V 的阅读量等等。
---特别推荐---
特别推荐：一个专注 go 项目实战、项目中踩坑经验及避坑指南、各种好玩的 go 工具的公众号。「Go 学堂」，专注实用性，非常值得大家关注。点击下方公众号卡片，直接关注。关注送《100 个 go 常见的错误》pdf 文档。
评论