在 go
的 sync
包中,有一个 singleflight
包,里面有一个 singleflight.go
文件,代码加注释,一共 200 行出头。内容包括以下几块儿:
Group
结构体管理一组相关的函数调用工作,它包含一个互斥锁和一个map
,map
的key
是函数的名称,value
是对应的call
结构体。call
结构体表示一个inflight
或已完成的函数调用,包含等待组件WaitGroup
、调用结果val
和err
、调用次数dups
和通知通道chans
。Do
方法接收一个key
和函数fn
,它会先查看map
中是否已经有这个key
的调用在inflight
,如果有则等待并返回已有结果,如果没有则新建一个call
并执行函数调用。DoChan
类似Do
但返回一个channel
来接收结果。doCall
方法包含了具体处理调用的逻辑,它会在函数调用前后添加defer
来recover
panic
和区分正常return
与runtime.Goexit
。- 如果发生
panic
,会将panicwraps
成错误返回给等待的channel
,如果是goexit
会直接退出。正常return
时会将结果发送到所有通知channel
。 Forget
方法可以忘记一个key
的调用,下次Do
时会重新执行函数。
这个包通过互斥锁和 map
实现了对相同 key
的函数调用去重,可以避免对已有调用的重复计算,同时通过 channel
机制可以通知调用者函数执行结果。在一些需要确保单次执行的场景中,可以使用这个包中的方法。
通过 singleflight
可以很容易实现缓存和去重的效果,避免重复计算,接下来,我们来模拟一下并发请求可能导致的缓存穿透场景,以及如何用 singleflight
包来解决这个问题:
package mainimport ("context""fmt""golang.org/x/sync/singleflight""sync/atomic""time")type Result string
// 模拟查询数据库
func find(ctx context.Context, query string) (Result, error) {return Result(fmt.Sprintf("result for %q", query)), nil
}func main() {var g singleflight.Groupconst n = 200waited := 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 阻塞时,所有请求都会阻塞,内存可能会出现大的问题。
此时,Do
可以更换为DoChan
,两者实现上完全一样,不同的是,DoChan()
通过 channel
返回结果。因此可以使用 select
语句实现超时控制
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 elapsedfmt.Println("Timeout")return
case ret = <-ch: // Received result from channelfmt.Printf("index: %d, val: %v, shared: %v\n", j, ret.Val, ret.Shared)
}
在超时时主动返回,不阻塞。
此时又引入了另一个问题,这样的每一次的请求,并不是高可用的,成功率是无法保证的。这时候可以增加一定的请求饱和度来保证业务的最终成功率,此时一次请求还是多次请求,对于下游服务而言并没有太大区别,此时使用 singleflight
只是为了降低请求的数量级,那么可以使用 Forget()
来提高下游请求的并发。
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%的成功,如果单次的失败无法容忍,在高并发的场景下需要使用更好的处理方案,比如牺牲一部分实时性、完全使用缓存查询 + 异步更新等。