概述

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 也都会接收到同样的错误