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

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

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

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

package main
import (
   "context"
   "fmt"
   "golang.org/x/sync/singleflight"
   "sync/atomic"
   "time"
   )
type Result string
// 模擬查詢數(shù)據(jù)庫
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)
}

在這段程序中，如果重復使用查詢結果，shared 會返回 true，穿透查詢會返回 false

上面的設計中還有一個問題，就是在 Do 阻塞時，所有請求都會阻塞，內存可能會出現(xiàn)大的問題。

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

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

在超時時主動返回，不阻塞。

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

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

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

最新評論