網站首頁 編程語言 正文
1.背景
1.1. 項目介紹
golang/sync庫拓展了官方自帶的sync庫,提供了errgroup、semaphore、singleflight及syncmap四個包,本次分析singlefliht的源代碼。
singlefliht用于解決單機協程并發調用下的重復調用問題,常與緩存一起使用,避免緩存擊穿。
1.2.使用方法
go get -u golang.org/x/sync
- 核心API:Do、DoChan、Forget
- Do:同一時刻對某個Key方法的調用, 只能由一個協程完成,其余協程阻塞直到該協程執行成功后,直接獲取其生成的值,以下是一個避免緩存擊穿的常見使用方法:
func main() { var flight singleflight.Group var errGroup errgroup.Group // 模擬并發獲取數據緩存 for i := 0; i < 10; i++ { i := i errGroup.Go(func() error { fmt.Printf("協程%v準備獲取緩存\n", i) v, err, shared := flight.Do("getCache", func() (interface{}, error) { // 模擬獲取緩存操作 fmt.Printf("協程%v正在讀數據庫獲取緩存\n", i) time.Sleep(100 * time.Millisecond) fmt.Printf("協程%v讀取數據庫生成緩存成功\n", i) return "mockCache", nil }) if err != nil { fmt.Printf("err = %v", err) return err } fmt.Printf("協程%v獲取緩存成功, v = %v, shared = %v\n", i, v, shared) return nil }) } if err := errGroup.Wait(); err != nil { fmt.Printf("errGroup wait err = %v", err) } } // 輸出:只有0號協程實際生成了緩存,其余協程讀取生成的結果 協程0準備獲取緩存 協程4準備獲取緩存 協程3準備獲取緩存 協程2準備獲取緩存 協程6準備獲取緩存 協程5準備獲取緩存 協程7準備獲取緩存 協程1準備獲取緩存 協程8準備獲取緩存 協程9準備獲取緩存 協程0正在讀數據庫獲取緩存 協程0讀取數據庫生成緩存成功 協程0獲取緩存成功, v = mockCache, shared = true 協程8獲取緩存成功, v = mockCache, shared = true 協程2獲取緩存成功, v = mockCache, shared = true 協程6獲取緩存成功, v = mockCache, shared = true 協程5獲取緩存成功, v = mockCache, shared = true 協程7獲取緩存成功, v = mockCache, shared = true 協程9獲取緩存成功, v = mockCache, shared = true 協程1獲取緩存成功, v = mockCache, shared = true 協程4獲取緩存成功, v = mockCache, shared = true 協程3獲取緩存成功, v = mockCache, shared = true
DoChan:將執行結果返回到通道中,可通過監聽通道結果獲取方法執行值,這個方法相較于Do來說的區別是執行DoChan后不會阻塞到其中一個協程完成任務,而是異步執行任務,最后需要結果時直接從通道中獲取,避免長時間等待。
func testDoChan() { var flight singleflight.Group var errGroup errgroup.Group // 模擬并發獲取數據緩存 for i := 0; i < 10; i++ { i := i errGroup.Go(func() error { fmt.Printf("協程%v準備獲取緩存\n", i) ch := flight.DoChan("getCache", func() (interface{}, error) { // 模擬獲取緩存操作 fmt.Printf("協程%v正在讀數據庫獲取緩存\n", i) time.Sleep(100 * time.Millisecond) fmt.Printf("協程%v讀取數據庫獲取緩存成功\n", i) return "mockCache", nil }) res := <-ch if res.Err != nil { fmt.Printf("err = %v", res.Err) return res.Err } fmt.Printf("協程%v獲取緩存成功, v = %v, shared = %v\n", i, res.Val, res.Shared) return nil }) } if err := errGroup.Wait(); err != nil { fmt.Printf("errGroup wait err = %v", err) } } // 輸出結果 協程9準備獲取緩存 協程0準備獲取緩存 協程1準備獲取緩存 協程6準備獲取緩存 協程5準備獲取緩存 協程2準備獲取緩存 協程7準備獲取緩存 協程8準備獲取緩存 協程4準備獲取緩存 協程9正在讀數據庫獲取緩存 協程9讀取數據庫獲取緩存成功 協程3準備獲取緩存 協程3獲取緩存成功, v = mockCache, shared = true 協程8獲取緩存成功, v = mockCache, shared = true 協程0獲取緩存成功, v = mockCache, shared = true 協程1獲取緩存成功, v = mockCache, shared = true 協程6獲取緩存成功, v = mockCache, shared = true 協程5獲取緩存成功, v = mockCache, shared = true 協程2獲取緩存成功, v = mockCache, shared = true 協程7獲取緩存成功, v = mockCache, shared = true 協程4獲取緩存成功, v = mockCache, shared = true 協程9獲取緩存成功, v = mockCache, shared = true
2.源碼分析
2.1.項目結構
- singleflight.go:核心實現,提供相關API
- singleflight_test.go:相關API單元測試
2.2.數據結構
- singleflight.go
// singleflight.Group type Group struct { mu sync.Mutex // map的鎖 m map[string]*call // 保存每個key的調用 } // 一次Do對應的響應結果 type Result struct { Val interface{} Err error Shared bool } // 一個key會對應一個call type call struct { wg sync.WaitGroup val interface{} // 保存調用的結果 err error // 調用出現的err // 該call被調用的次數 dups int // 每次DoChan時都會追加一個chan在該列表 chans []chan<- Result }
2.3.API代碼流程
func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool)
func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) { g.mu.Lock() if g.m == nil { // 第一次執行Do的時候創建map g.m = make(map[string]*call) } // 已經存在該key,對應后續的并發調用 if c, ok := g.m[key]; ok { // 執行次數自增 c.dups++ g.mu.Unlock() // 等待執行fn的協程完成 c.wg.Wait() // ... // 返回執行結果 return c.val, c.err, true } // 不存在該key,說明第一次調用,初始化一個call c := new(call) // wg添加1,后續其他協程在該wg上阻塞 c.wg.Add(1) // 保存key和call的關系 g.m[key] = c g.mu.Unlock() // 真正執行fn函數 g.doCall(c, key, fn) return c.val, c.err, c.dups > 0 } func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) { normalReturn := false recovered := false // 第三步、最后的設置和清理工作 defer func() { // ... g.mu.Lock() defer g.mu.Unlock() // 執行完成,調用wg.Done,其他協程此時不再阻塞,讀到fn執行結果 c.wg.Done() // 二次校驗map中key的值是否為當前call,并刪除該key if g.m[key] == c { delete(g.m, key) } // ... // 如果c.chans存在,則遍歷并寫入執行結果 for _, ch := range c.chans { ch <- Result{c.val, c.err, c.dups > 0} } } }() // 第一步、執行fn獲取結果 func() { // 3、如果fn執行過程中panic,將c.err設置為PanicError defer func() { if !normalReturn { if r := recover(); r != nil { c.err = newPanicError(r) } } }() // 1、執行fn,獲取到執行結果 c.val, c.err = fn() // 2、設置正常返回結果標識 normalReturn = true }() // 第二步、fn執行出錯,將recovered標識設置為true if !normalReturn { recovered = true } }
func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result
func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result { // 一次調用對應一個chan ch := make(chan Result, 1) g.mu.Lock() if g.m == nil { // 第一次調用,初始化map g.m = make(map[string]*call) } // 后續調用,已存在key if c, ok := g.m[key]; ok { // 調用次數自增 c.dups++ // 將chan添加到chans列表 c.chans = append(c.chans, ch) g.mu.Unlock() // 直接返回chan,不等待fn執行完成 return ch } // 第一次調用,初始化call及chans列表 c := &call{chans: []chan<- Result{ch}} // wg加一 c.wg.Add(1) // 保存key及call的關系 g.m[key] = c g.mu.Unlock() // 異步執行fn函數 go g.doCall(c, key, fn) // 直接返回該chan return ch }
3.總結
- singleflight經常和緩存獲取配合使用,可以緩解緩存擊穿問題,避免同一時刻單機大量的并發調用獲取數據庫構建緩存
- singleflight的實現很精簡,核心流程就是使用map保存每次調用的key與call的映射關系,每個call中通過wg控制只存在一個協程執行fn函數,其他協程等待執行完成后,直接獲取執行結果,在執行完成后會刪去map中的key
- singleflight的Do方法會阻塞直到fn執行完成,DoChan方法不會阻塞,而是異步執行fn,并通過通道來實現結果的通知
原文鏈接:https://blog.csdn.net/pbrlovejava/article/details/127717139
相關推薦
- 2022-05-20 Spring注入bean的常用的六種方式
- 2022-12-27 python中getopt()函數用法詳解_python
- 2022-12-21 k8s安裝CICD?devtron過程詳解_云其它
- 2022-10-31 sql查詢給表起別名要點小結(涉及嵌套查詢)_oracle
- 2022-08-10 Go語言pointer及switch?fallthrough實戰詳解_Golang
- 2022-12-06 docker多容器操作與強制刪除容器的方法步驟_docker
- 2022-08-02 goFrame的gqueue與channe的區別_Golang
- 2022-09-04 Go語言之init函數_Golang
- 最近更新
-
- window11 系統安裝 yarn
- 超詳細win安裝深度學習環境2025年最新版(
- Linux 中運行的top命令 怎么退出?
- MySQL 中decimal 的用法? 存儲小
- get 、set 、toString 方法的使
- @Resource和 @Autowired注解
- Java基礎操作-- 運算符,流程控制 Flo
- 1. Int 和Integer 的區別,Jav
- spring @retryable不生效的一種
- Spring Security之認證信息的處理
- Spring Security之認證過濾器
- Spring Security概述快速入門
- Spring Security之配置體系
- 【SpringBoot】SpringCache
- Spring Security之基于方法配置權
- redisson分布式鎖中waittime的設
- maven:解決release錯誤:Artif
- restTemplate使用總結
- Spring Security之安全異常處理
- MybatisPlus優雅實現加密?
- Spring ioc容器與Bean的生命周期。
- 【探索SpringCloud】服務發現-Nac
- Spring Security之基于HttpR
- Redis 底層數據結構-簡單動態字符串(SD
- arthas操作spring被代理目標對象命令
- Spring中的單例模式應用詳解
- 聊聊消息隊列,發送消息的4種方式
- bootspring第三方資源配置管理
- GIT同步修改后的遠程分支