一、簡(jiǎn)介
在日常工作中，經(jīng)常會(huì)遇到對(duì)某種操作進(jìn)行頻次控制或者統(tǒng)計(jì)次數(shù)的需求，此時(shí)常用的做法是采用redis的incr來(lái)遞增,記錄訪問次數(shù), 以及 expire 來(lái)設(shè)置失效時(shí)間。本文將以一個(gè)實(shí)際的例子來(lái)說(shuō)明incr存在的一個(gè)"坑"，以及給出解決方案。
如：
26.redis實(shí)現(xiàn)日限流、周限流（含黑名單、白名單）
27.Go實(shí)現(xiàn)一月（30天）內(nèi)不發(fā)送重復(fù)內(nèi)容的站內(nèi)信給用戶
有這么一個(gè)場(chǎng)景，用戶需要進(jìn)行ocr識(shí)別，為了防止接口被刷，這里面做了一個(gè)限制(每分鐘調(diào)用次數(shù)不能超過xx次)。 經(jīng)過調(diào)研后，決定使用redis的incr和expire來(lái)實(shí)現(xiàn)這個(gè)功能
二、代碼演進(jìn)
第一版代碼（存在bug隱患）
// 執(zhí)行ocr調(diào)用
func (o *ocrSvc)doOcr(ctx context.Context,uid int)(interface,err){
// 如果調(diào)用次數(shù)超過了指定限制，就直接拒絕此次請(qǐng)求
ok,err := o.checkMinute(uid)
if err != nil {
return nil,err
}
if !ok {
return nil,errors.News("frequently called")
}
// 執(zhí)行第三方ocr調(diào)用(偽代碼：模擬一個(gè)rpc接口)
ocrRes,err := doOcrByThird()
if err != nil {
return nil,err
}
// 調(diào)用成功則執(zhí)行 incr操作
if err := o.redis.Incr(ctx,buildUserOcrCountKey(uid));err!=nil{
return nil,err
}
return ocrRes,nil
}
// 校驗(yàn)每分鐘調(diào)用次數(shù)是否超過限制
func (o *ocrSvc)checkMinute (ctx context.Context,uid int) (bool, error) {
minuteCount, err := o.redis.Get(ctx, buildUserOcrCountKey(uid))
if err != nil && !errors.Is(err, eredis.Nil) {
log.Error("checkMinute: redis.Get failed", zap.Error(err))
return false, constx.ErrServer
}
if errors.Is(err, eredis.Nil) {
// 過期了，或者沒有該用戶的調(diào)用次數(shù)記錄(設(shè)置初始值為0，過期時(shí)間為1分鐘)
o.redis.Set(ctx, buildUserOcrCountKey(uid),0,time.Minute)
return true, nil
}
// 已經(jīng)超過每分鐘的調(diào)用次數(shù)
if cast.ToInt(minuteCount) >= config.UserOcrMinuteCount() {
log.Warn("checkMinute: user FrequentlyCalled", zap.Int64("uid", uid), zap.String("minuteCount", minuteCount))
return false, nil
}
return true, nil
}
詳解
這一版代碼存在什么問題呢？問題出在了 開始判斷沒有超出限制，然后執(zhí)行第三方rpc接口調(diào)用也成功了，接下來(lái)直接進(jìn)行計(jì)數(shù)加一（incr）操作有問題，如下圖所示
說(shuō)明：
- 假設(shè)當(dāng)前用戶在進(jìn)行
ocr識(shí)別時(shí)，未超過調(diào)用次數(shù)。但是在redis中的ttl還剩1秒鐘 - 然后調(diào)用第三方
ocr進(jìn)行識(shí)別,加入耗時(shí)超過了1秒 - 識(shí)別成功后，調(diào)用次數(shù)
+1。這里就很有可能出問題，比如：在incr的時(shí)候剛好該key在1s前過期了，那么redis是怎么做的呢，它會(huì)將該key的值設(shè)置為1，ttl設(shè)置為-1，ttl設(shè)置為-1，ttl設(shè)置為-1（重要的事情說(shuō)三遍）,-1表示沒有過期時(shí)間 - 這時(shí)候
bug就出現(xiàn)了，用戶的調(diào)用次數(shù)一直在漲，并且也不會(huì)過期，達(dá)到臨界值時(shí)用戶的請(qǐng)求就會(huì)被拒掉，相當(dāng)于該用戶之后都不能訪問這個(gè)接口了，并且這種key變多后，由于沒有過期時(shí)間，還會(huì)一直占用redis的內(nèi)存。
總結(jié)
以上代碼說(shuō)明了一個(gè)問題，也就是incr和expire必須具備原子性。而我們第一版代碼顯然在邊界條件下是不滿足要求的，極有可能造成bug，影響用戶體驗(yàn)，強(qiáng)烈不推薦使用，接下來(lái)一步一步引入修正后的代碼
第二版代碼（幾乎無(wú)隱患）
從對(duì)第一版代碼的分析可知，是由于查詢次數(shù)還沒有達(dá)到限制后，又進(jìn)行了一些rpc調(diào)用，或者處理了一些其他業(yè)務(wù)邏輯，這個(gè)時(shí)間內(nèi)，可能key過期了，然后我們直接使用incr進(jìn)行計(jì)數(shù)加一，導(dǎo)致了永不過期的key產(chǎn)生。那么我們是不是可以在incr前先保證key還沒有過期就行呢？答案是可以的，代碼如下：
// 執(zhí)行ocr調(diào)用
func (o *ocrSvc)doOcr(ctx context.Context,uid int)(interface,err){
// 如果調(diào)用次數(shù)超過了指定限制，就直接拒絕此次請(qǐng)求
ok,err := o.checkMinute(uid)
if err != nil {
return nil,err
}
if !ok {
return nil,errors.News("frequently called")
}
// 執(zhí)行第三方ocr調(diào)用(偽代碼：模擬一個(gè)rpc接口)
ocrRes,err := doOcrByThird()
if err != nil {
return nil,err
}
// 調(diào)用成功則執(zhí)行 incr操作
exists, err := o.redis.Exists(ctx, buildUserOcrCountKey(uid)).Result()
if err != nil {
log.Error("doOcr: redis.Exists failed", zap.Error(err))
return nil, err
}
if exists == 1 { // key存在，計(jì)數(shù)加1
if err := o.redis.Incr(ctx,buildUserOcrCountKey(uid));err!=nil{
return nil,err
}
} else { // key不存在，設(shè)置key與過期時(shí)間
if err := o.redis.Set(ctx,buildUserOcrCountKey(uid),1,expireTime);err!=nil{
return nil,err
}
}
return ocrRes,nil
}
// 校驗(yàn)每分鐘調(diào)用次數(shù)是否超過限制
func (o *ocrSvc)checkMinute (ctx context.Context,uid int) (bool, error) {
minuteCount, err := o.redis.Get(ctx, buildUserOcrCountKey(uid))
if err != nil && !errors.Is(err, eredis.Nil) {
log.Error("checkMinute: redis.Get failed", zap.Error(err))
return false, constx.ErrServer
}
if errors.Is(err, eredis.Nil) {
// 過期了，或者沒有該用戶的調(diào)用次數(shù)記錄(設(shè)置初始值為0，過期時(shí)間為1分鐘)
o.redis.Set(ctx, buildUserOcrCountKey(uid),0,time.Minute)
return true, nil
}
// 已經(jīng)超過每分鐘的調(diào)用次數(shù)
if cast.ToInt(minuteCount) >= config.UserOcrMinuteCount() {
log.Warn("checkMinute: user FrequentlyCalled", zap.Int64("uid", uid), zap.String("minuteCount", minuteCount))
return false, nil
}
return true, nil
}
與第一版的差異主要在于如下代碼：
- 在需要
incr操作前，我們先查看key是否存在（且沒有過期）- 確保存在后，立即
incr(這兩步間隔幾乎可以忽略，所以幾乎可以避免第一版中的問題） - 如果不存在則設(shè)置
key并設(shè)置過期時(shí)間。
- 確保存在后，立即
注：redis中的incr命令是不會(huì)改變key的過期時(shí)間的
// 調(diào)用成功則執(zhí)行 incr操作
exists, err := o.redis.Exists(ctx, buildUserOcrCountKey(uid)).Result()
if err != nil {
log.Error("doOcr: redis.Exists failed", zap.Error(err))
return nil, err
}
if exists == 1 { // key存在，計(jì)數(shù)加1
if err := o.redis.Incr(ctx,buildUserOcrCountKey(uid));err!=nil{
return nil,err
}
} else { // key不存在，設(shè)置key與過期時(shí)間
if err := o.redis.Set(ctx,buildUserOcrCountKey(uid),1,expireTime);err!=nil{
return nil,err
}
}
還有一種方式是查看key的過期時(shí)間,使用ttl，這樣即使在極端情況下通過incr設(shè)置出了沒有過期時(shí)間的key，也會(huì)在第二次訪問的時(shí)候通過Set設(shè)置過期時(shí)間了。
注：ttl命令返回值是鍵的剩余時(shí)間（單位是秒）。當(dāng)鍵不存在時(shí)，ttl命令會(huì)返回-2。沒有為鍵設(shè)置過期時(shí)間（即永久存在，這是建立一個(gè)鍵后的默認(rèn)情況）返回-1。
// 調(diào)用成功則執(zhí)行 incr操作
cnt, err := o.redis.Ttl(ctx, buildUserOcrCountKey(uid)).Result()
if err != nil {
log.Error("doOcr: redis.Ttl failed", zap.Error(err))
return nil, err
}
if cnt >= 1 { // key存在,且還沒有過期
if err := o.redis.Incr(ctx,buildUserOcrCountKey(uid));err!=nil{
return nil,err
}
} else { // key馬上過期0，key沒有過期時(shí)間-1， key不存在-2
if err := o.redis.Set(ctx,buildUserOcrCountKey(uid),1,expireTime);err!=nil{
return nil,err
}
}
第三版代碼(完美無(wú)瑕）
第二版代碼中的兩種方式其實(shí)已經(jīng)可以在工作中使用了，但如果追求完美無(wú)瑕的話，ttl版本的代碼在極端情況下還是有點(diǎn)瑕疵，比如極端情況下，key過期時(shí)間還有1s過期，然后我們用incr去累加，但是網(wǎng)絡(luò)延遲了，導(dǎo)致命令到達(dá)redis服務(wù)器的時(shí)候，key已經(jīng)過期了，盡管第二次訪問會(huì)用set重置key并設(shè)置過期時(shí)間，但是萬(wàn)一該用戶再也不來(lái)訪問了呢？這時(shí)候這個(gè)key就會(huì)永遠(yuǎn)占據(jù)著內(nèi)存了。文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-811885.html
將incr+expire放在lua腳本中執(zhí)行保證原子性是最完美的。廢話不多說(shuō)了，直接上代碼文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-811885.html
// 執(zhí)行ocr調(diào)用
func (o *ocrSvc)doOcr(ctx context.Context,uid int)(interface,err){
// 如果調(diào)用次數(shù)超過了指定限制，就直接拒絕此次請(qǐng)求
ok,err := o.checkMinute(uid)
if err != nil {
return nil,err
}
if !ok {
return nil,errors.News("frequently called")
}
// 執(zhí)行第三方ocr調(diào)用((偽代碼：模擬一個(gè)rpc接口))
ocrRes,err := doOcrByThird()
if err != nil {
return nil,err
}
// 調(diào)用成功則執(zhí)行 incr操作
if err := o.incrCount(ctx,buildUserOcrCountKey(uid));err!=nil{
return nil,err
}
return ocrRes,nil
}
func (o *ocrSvc) incrCount(ctx context.Context, uid int64) error {
/*
此段lua腳本的作用:
第一步，先執(zhí)行incr操作
local current = redis.call('incr',KEYS[1])
第二步，看下該key的ttl
local t = redis.call('ttl',KEYS[1]);
第三步，如果ttl為-1(永不過期)
if t == -1 then
則重新設(shè)置過期時(shí)間為 「一分鐘」
redis.call('expire',KEYS[1],ARGV[1])
end;
*/
script := redis.NewScript(
`local current = redis.call('incr',KEYS[1]);
local t = redis.call('ttl',KEYS[1]);
if t == -1 then
redis.call('expire',KEYS[1],ARGV[1])
end;
return current
`)
var (
expireTime = 60 // 60 秒
)
_, err := script.Run(ctx, b.redis.Client(), []string{buildUserOcrCountKey(uid)}, expireTime).Result()
if err != nil {
return err
}
return nil
}
// 校驗(yàn)每分鐘調(diào)用次數(shù)是否超過
func (o *ocrSvc)checkMinute (ctx context.Context,uid int) (bool, error) {
minuteCount, err := o.redis.Get(ctx, buildUserOcrCountKey(uid))
if err != nil && !errors.Is(err, eredis.Nil) {
elog.Error("checkMinute: redis.Get failed", zap.Error(err))
return false, constx.ErrServer
}
if errors.Is(err, eredis.Nil) {
// 第二版代碼中在check時(shí)不進(jìn)行初始化操作
// 過期了，或者沒有該用戶的調(diào)用次數(shù)記錄(設(shè)置初始值為0，過期時(shí)間為1分鐘)
// o.redis.Set(ctx, buildUserOcrCountKey(uid),0,time.Minute)
return true, nil
}
// 已經(jīng)超過每分鐘的調(diào)用次數(shù)
if cast.ToInt(minuteCount) >= config.UserOcrMinuteCount() {
elog.Warn("checkMinute: user FrequentlyCalled", zap.Int64("uid", uid), zap.String("minuteCount", minuteCount))
return false, nil
}
return true, nil
}
到了這里，關(guān)于68. redis計(jì)數(shù)與限流中incr+expire的坑以及解決辦法(Lua+TTL)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
