Go扩展原语之SingleFlight的用法详解

概述

singleflight.Group 是 Go 语言扩展包的另一种同步原语，它能够再一个服务中抑制对下游的多次重复请求。一个比较常见的使用场景是，我们使用 Redis 对数据库中的数据进行缓存，发生缓存击穿时，大量请求会打到数据库上进而影响服务的尾延时。

而 singleflight.Group 能够有效地解决这个问题，它能够限制对同一个键值对的多次重复请求，减少对下游的瞬时流量。

在资源的获取非常昂贵时（例如访问缓存、数据库），就很适合使用 singleflight.Group 优化服务。它的使用方法如下：

type service struct {
    requestGroup singleflight.Group
}
func (s *service) handleRequest(ctx context.Context, request Request) (Response, error) {
    v, err, _ := requestGroup.Do(request.Hash(), func() (interface{}, error) {
        rows, err := // select * from tables
        if err != nil {
            return nil, err
        }
        return rows, nil
    })
    if err != nil {
        return nil, err
    }
    return Response{
        rows: rows,
    }, nil
}

因为请求的哈希在业务上一般表示相同的请求，所以上述代码使用它作为请求的键。当然，我们也可以选择其他的字段作为 singleflight.Group.Do 方法的第一个参数减少重复的请求。

结构体

singleflight.Group 结构体由一个互斥锁 sync.Mutex 和一个映射表组成，每一个 singleflight.call 结构体都保存了当前调用对应的信息：

type Group struct {
	mu sync.Mutex
	m  map[string]*call
}
type call struct {
	wg sync.WaitGroup
	val interface{}
	err error
	dups  int
	chans []chan<- Result
}

singleflight.call 结构体中的 val 和 err 字段都只会在执行传入的函数时赋值一次并在 sync.WaitGroup.Wait 返回时被读取。

dups 和 chans 两个字段分别存储了抑制的请求数量以及用于同步结果的 Channel。

接口

singleflight.Group求的方法：

• singleflight.Group.Do  — 同步等待的方法；
• singleflight.Group.DoChan — 返回 Channel 异步等待的方法；

这两个方法在功能上没有太多的区别，只是在接口的表现上稍有不同。

每次调用 singleflight.Group.Do 方法时都会获取互斥锁，随后判断是否已经存在键对应的 singleflight.call：

• 当不存在对应的 singleflight.call 时：

  • 初始化一个新的 singleflight.call 指针
  • 增加 sync.WaitGroup 持有的计数器
  • 将 singleflight.call 指针添加到映射表
  • 释放持有的互斥锁
  • 阻塞地调用 singleflight.Group.doCall 方法等待结果的返回

• 当存在对应的 singleflight.call 时：

  • 增加 dups 计数器，它表示当前重复的调用次数
  • 释放持有的互斥锁
  • 通过 sync.WaitGroup.Wait 等待请求的返回

func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {
    g.mu.Lock()
    if g.m == nil {
            g.m = make(map[string]*call)
    }
    if c, ok := g.m[key]; ok {
            c.dups++
            g.mu.Unlock()
            c.wg.Wait()
            return c.val, c.err, true
    }
    c := new(call)
    c.wg.Add(1)
    g.m[key] = c
    g.mu.Unlock()
    g.doCall(c, key, fn)
    return c.val, c.err, c.dups > 0
}

因为 val 和 err 两个字段都只会在 singleflight.Group.doCall 方法中赋值，所以当 singleflight.Group.doCall 和 sync.WaitGroup.Wait 返回时，函数调用的结果和错误都会返回给 singleflight.Group.Do 的调用方。

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
	c.val, c.err = fn()
	c.wg.Done()
	g.mu.Lock()
	delete(g.m, key)
	for _, ch := range c.chans {
		ch <- Result{c.val, c.err, c.dups > 0}
	}
	g.mu.Unlock()
}

• 行传入的函数 fn，该函数的返回值会赋值给 c.val 和 c.err
• 调用 sync.WaitGroup.Done 方法通知所有等待结果的 Goroutine — 当前函数已经执行完成，可以从 call 结构体中取出返回值并返回了
• 获取持有的互斥锁并通过管道将信息同步给使用 singleflight.Group.DoChan 方法的 Goroutine

func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result {
	ch := make(chan Result, 1)
	g.mu.Lock()
	if g.m == nil {
		g.m = make(map[string]*call)
	}
	if c, ok := g.m[key]; ok {
		c.dups++
		c.chans = append(c.chans, ch)
		g.mu.Unlock()
		return ch
	}
	c := &call{chans: []chan<- Result{ch}}
	c.wg.Add(1)
	g.m[key] = c
	g.mu.Unlock()
	go g.doCall(c, key, fn)
	return ch
}

singleflight.Group.Do 和 singleflight.Group.DoChan 分别提供了同步和异步的调用方式，这让我们使用起来也更加灵活。

小结

当我们需要减少对下游的相同请求时，可以使用 singleflight.Group 来增加吞吐量和服务质量，不过在使用的过程中我们也需要注意以下的几个问题：

• singleflight.Group.Do 和 singleflight.Group.DoChan 一个用于同步阻塞调用传入的函数，一个用于异步调用传入的参数并通过 Channel 接收函数的返回值
• singleflight.Group.Forget 可以通知 singleflight.Group 在持有的映射表中删除某个键，接下来对该键的调用就不会等待前面的函数返回了
• 一旦调用的函数返回了错误，所有在等待的 Goroutine 也都会接收到同样的错误

到此这篇关于Go扩展原语之SingleFlight的用法详解的文章就介绍到这了,更多相关Go SingleFlight内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！