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方法的调用， 只能由一个协程完成，其余协程阻塞直到该协程执行成功后，直接获取其生成的值，以下是一个避免缓存击穿的常见使用方法：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 |
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后不会阻塞到其中一个协程完成任务，而是异步执行任务，最后需要结果时直接从通道中获取，避免长时间等待。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 |
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
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
// 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)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 |
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
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 |
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，并通过通道来实现结果的通知
