写点什么

sync.singleflight 到底怎么用才对?

用户头像
cyningsun
关注
发布于: 2021 年 01 月 12 日

背景


缓存 在各种场景中被大量使用,在 Cache Miss(缓存未命中)的情况下,就会出现下图的情况:

所有的请求被同时打到下游存储上,将会影响下游存储的服务质量,因此需要严格限制访问下游存储的并发量。使用 Golang 编程的人,倾向于不假思索的使用 singleflight 应对 Cache Miss(缓存未命中),即:

在绝大多数场景下,singleflight 都很好用,因此让很多人相信 singleflight 是完美无缺的银弹。在 2020 年的电商大促中,因为此种认知,导致线上业务出现了严重故障。之前只是直觉性的觉得这么设计不妥,因为该问题促使我回过头来梳理该类型设计的存在的不足。

了解 singleflight

在此之前,先了解下 singleflight 来源于准官方库golang.org/x/sync/singleflight,能够在抑制对下游的多次重复请求.主要提供了以下三个方法:


// Do():  相同的 key,fn 同时只会执行一次,返回执行的结果给fn执行期间,所有使用该 key 的调用// v: fn 返回的数据// err: fn 返回的err// shared: 表示返回数据是调用 fn 得到的还是其他相同 key 调用返回的func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {// DoChan(): 类似Do方(),以 chan 返回结果func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result {// Forget(): 失效 key,后续对此 key 的调用将执行 fn,而不是等待前面的调用完成func (g *Group) Forget(key string)
复制代码


通常的使用方式如下:


package main
import ( "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.Group const n = 5 waited := int32(n) done := make(chan struct{}) key := "https://weibo.com/1227368500/H3GIgngon" 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") }}
复制代码


输出结果如下:


index: 1, val: result for "https://weibo.com/1227368500/H3GIgngon", shared: trueindex: 2, val: result for "https://weibo.com/1227368500/H3GIgngon", shared: trueindex: 3, val: result for "https://weibo.com/1227368500/H3GIgngon", shared: falseindex: 4, val: result for "https://weibo.com/1227368500/H3GIgngon", shared: falseindex: 0, val: result for "https://weibo.com/1227368500/H3GIgngon", shared: false
复制代码


如果函数执行一切正常,则所有请求都能顺利获得正确的数据。相反,如果函数执行遇到问题呢?由于 singleflight 是以阻塞读的方式来控制向下游请求的并发量,在第一个下游请求没有返回之前,所有请求都将被阻塞。

问题分析

假设服务正常情况下处理能力为 1W QPS,每次请求会发起 3 次 下游调用,其中一个下游调用使用 singleflight 获取控制并发获取数据,请求超时时间为 3S。那么在出现请求超时的情况下,会出现以下几个问题:


  • 协程暴增,最小协程数为 3W(1 W/S * 3S)

  • 内存暴涨,内存总大小为:协程内存大小 + 1W/S 3S (3+1)次 * (请求包+响应包)大小

  • 大量超时报错:1W/S * 3S

  • 后续请求耗时增加(调度等待)


如果类似问题出现在重要程度高的接口上,例如:读取游戏配置获取博主信息 等关键接口,那么问题将是非常致命的。出现该情况的根本原因有以下两点:


  • 阻塞读:缺少超时控制,难以快速失败

  • 单并发:控制了并发量,但牺牲了成功率


那么如何应对以上问题呢?

阻塞读

作为 Do() 的替代函数,singleflight 提供了 DoChan()。两者实现上完全一样,不同的是,DoChan() 通过 channel 返回结果。因此可以使用 select 语句实现超时控制


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

单并发

在一些对可用性要求极高的场景下,往往需要一定的请求饱和度来保证业务的最终成功率。一次请求还是多次请求,对于下游服务而言并没有太大区别,此时使用 singleflight 只是为了降低请求的数量级,那么使用 Forget() 提高下游请求的并发:


v, _, shared := g.Do(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})
复制代码


当有一个并发请求超过 10ms,那么将会有第二个请求发起,此时只有 10ms 内的请求最多发起一次请求,即最大并发:100 QPS。单次请求失败的影响大大降低。

总结

当然,如果单次的失败无法容忍,在高并发的场景下更好的处理方案是:


  1. 放弃使用同步请求,牺牲数据更新的实时性

  2. “缓存” 存储准实时的数据 + “异步更新” 数据到缓存


以上模式,在设计弹幕系统的时候有使用到,详细介绍参考:弹幕系统设计实践


本文作者:cyningsun

本文地址https://www.cyningsun.com/01-11-2021/golang-concurrency-singleflight.html

版权声明:本博客所有文章除特别声明外,均采用 CC BY-NC-ND 3.0 CN 许可协议。转载请注明出处!


用户头像

cyningsun

关注

还未添加个人签名 2018.09.19 加入

还未添加个人简介

评论

发布
暂无评论
sync.singleflight 到底怎么用才对?