在 go 的 sync 包中，有一個(gè) singleflight 包，里面有一個(gè) singleflight.go 文件，代碼加注釋，一共 200 行出頭。內(nèi)容包括以下幾塊兒：

1. Group 結(jié)構(gòu)體管理一組相關(guān)的函數(shù)調(diào)用工作,它包含一個(gè)互斥鎖和一個(gè) map,map 的 key 是函數(shù)的名稱,value 是對(duì)應(yīng)的 call 結(jié)構(gòu)體。
2. call 結(jié)構(gòu)體表示一個(gè) inflight 或已完成的函數(shù)調(diào)用,包含等待組件 WaitGroup、調(diào)用結(jié)果 val 和 err、調(diào)用次數(shù) dups 和通知通道 chans。
3. Do 方法接收一個(gè) key 和函數(shù) fn,它會(huì)先查看 map 中是否已經(jīng)有這個(gè) key 的調(diào)用在 inflight,如果有則等待并返回已有結(jié)果,如果沒有則新建一個(gè) call 并執(zhí)行函數(shù)調(diào)用。
4. DoChan 類似 Do 但返回一個(gè) channel 來接收結(jié)果。
5. doCall 方法包含了具體處理調(diào)用的邏輯,它會(huì)在函數(shù)調(diào)用前后添加 defer 來 recover panic 和區(qū)分正常 return 與 runtime.Goexit。
6. 如果發(fā)生 panic,會(huì)將 panicwraps 成錯(cuò)誤返回給等待的 channel,如果是 goexit 會(huì)直接退出。正常 return 時(shí)會(huì)將結(jié)果發(fā)送到所有通知 channel。
7. Forget 方法可以忘記一個(gè) key 的調(diào)用,下次 Do 時(shí)會(huì)重新執(zhí)行函數(shù)。

這個(gè)包通過互斥鎖和 map 實(shí)現(xiàn)了對(duì)相同 key 的函數(shù)調(diào)用去重,可以避免對(duì)已有調(diào)用的重復(fù)計(jì)算,同時(shí)通過 channel 機(jī)制可以通知調(diào)用者函數(shù)執(zhí)行結(jié)果。在一些需要確保單次執(zhí)行的場(chǎng)景中，可以使用這個(gè)包中的方法。

通過 singleflight 可以很容易實(shí)現(xiàn)緩存和去重的效果，避免重復(fù)計(jì)算，接下來，我們來模擬一下并發(fā)請(qǐng)求可能導(dǎo)致的緩存穿透場(chǎng)景,以及如何用 singleflight 包來解決這個(gè)問題：

package main
import (
   "context"
   "fmt"
   "golang.org/x/sync/singleflight"
   "sync/atomic"
   "time"
   )
type Result string
// 模擬查詢數(shù)據(jù)庫(kù)
func find(ctx context.Context, query string) (Result, error) {
   return Result(fmt.Sprintf("result for %q", query)), nil
}
func main() {
   var g singleflight.Group
   const n = 200
   waited := int32(n)
   done := make(chan struct{})
   key := "this is key"
   for i := 0; i < n; i++ {
      go func(j int) {
         v, _, shared := g.Do(key, func() (interface{}, error) {
            ret, err := find(context.Background(), key)
            return ret, err
         })
         if atomic.AddInt32(&waited, -1) == 0 {
            close(done)
         }
         fmt.Printf("index: %d, val: %v, shared: %v\n", j, v, shared)
      }(i)
   }
   select {
   case <-done:
   case <-time.After(time.Second):
      fmt.Println("Do hangs")
   }
   time.Sleep(time.Second * 4)
}

在這段程序中，如果重復(fù)使用查詢結(jié)果，shared 會(huì)返回 true，穿透查詢會(huì)返回 false

上面的設(shè)計(jì)中還有一個(gè)問題，就是在 Do 阻塞時(shí)，所有請(qǐng)求都會(huì)阻塞，內(nèi)存可能會(huì)出現(xiàn)大的問題。

此時(shí)，Do 可以更換為DoChan，兩者實(shí)現(xiàn)上完全一樣，不同的是，DoChan() 通過 channel 返回結(jié)果。因此可以使用 select 語(yǔ)句實(shí)現(xiàn)超時(shí)控制

ch := g.DoChan(key, func() (interface{}, error) {
   ret, err := find(context.Background(), key)
   return ret, err
})
// Create our timeout
timeout := time.After(500 * time.Millisecond)
var ret singleflight.Result
select {
case <-timeout: // Timeout elapsed
   fmt.Println("Timeout")
   return
case ret = <-ch: // Received result from channel
   fmt.Printf("index: %d, val: %v, shared: %v\n", j, ret.Val, ret.Shared)
}

在超時(shí)時(shí)主動(dòng)返回，不阻塞。

此時(shí)又引入了另一個(gè)問題，這樣的每一次的請(qǐng)求，并不是高可用的，成功率是無法保證的。這時(shí)候可以增加一定的請(qǐng)求飽和度來保證業(yè)務(wù)的最終成功率，此時(shí)一次請(qǐng)求還是多次請(qǐng)求，對(duì)于下游服務(wù)而言并沒有太大區(qū)別，此時(shí)使用  singleflight  只是為了降低請(qǐng)求的數(shù)量級(jí)，那么可以使用 Forget() 來提高下游請(qǐng)求的并發(fā)。

ch := g.DoChan(key, func() (interface{}, error) {
   go func() {
      time.Sleep(10 * time.Millisecond)
      fmt.Printf("Deleting key: %v\n", key)
      g.Forget(key)
   }()
   ret, err := find(context.Background(), key)
   return ret, err
})

當(dāng)然，這種做法依然無法保證100%的成功，如果單次的失敗無法容忍，在高并發(fā)的場(chǎng)景下需要使用更好的處理方案，比如犧牲一部分實(shí)時(shí)性、完全使用緩存查詢 + 異步更新等。

到此這篇關(guān)于golang使用sync.singleflight解決熱點(diǎn)緩存穿透問題的文章就介紹到這了,更多相關(guān)golang sync.singleflight緩存穿透內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論