前言

最近从java转到go，来公司第一个开发工作就是对一个资源请求去重复，最终发现这个singleflight这个好东西，分享一下。

singleflight使用场景

缓存击穿：缓存在某个时间点过期的时候，恰好在这个时间点对这个Key有大量的并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。
- 绝大多数公司都是这么用的
请求资源去重复
- 我们的用法，需要改动一行代码。

singleflight 简介

singleflight 在 golang.org/x/sync/singleflight 项目下，对外提供了以下几个方法

//Do方法，传入key，以及回调函数，如果key相同，fn方法只会执行一次，同步等待

//返回值v:表示fn执行结果

//返回值err:表示fn的返回的err

//返回值shared：表示是否是真实fn返回的还是从保存的map[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关联的值是否失效，默认以上两个方法只要fn方法执行完成后，内部维护的fn的值也删除（即并发结束后就失效了）

func (g *Group) Forget(key string)

singleflight的使用

从singleflight的test最简单用法

func TestDo(t *testing.T) {

	var g Group

    // key 可以理解资源的id

	v, err, _ := g.Do("key", func() (interface{}, error) {

    // do what you want

		return "bar", nil

	})

	if got, want := fmt.Sprintf("%v (%T)", v, v), "bar (string)"; got != want {

		t.Errorf("Do = %v; want %v", got, want)

	}

	if err != nil {

		t.Errorf("Do error = %v", err)

	}

}

验证并发重复请求

func process(g *Group, t *testing.T, ch chan int, key string) {

	for count := 0; count < 10; count++ {

		v, err, shared := g.Do(key, func() (interface{}, error) {

			time.Sleep(1000 * time.Millisecond)

			return "bar", nil

		})

		t.Log("v = ", v, " err = ", err, " shared =", shared, " ch :", ch, "g ", len(g.m))

		if got, want := fmt.Sprintf("%v (%T)", v, v), "bar (string)"; got != want {

			t.Errorf("Do = %v; want %v", got, want)

		}

		if err != nil {

			t.Errorf("Do error = %v", err)

		}

	}

	ch <- 1

}

func TestDo1(t *testing.T) {

	var g Group

	channels := make([]chan int, 10)

	key := "key"

	for i := 0; i < 10; i++ {

		channels[i] = make(chan int)

		go process(&g, t, channels[i], key)

	}

	for i, ch := range channels {

		<-ch

		fmt.Println("routine ", i, "quit!")

	}

}

结果

singleflight的原理

call

call 用来表示一个正在执行或已完成的函数调用。

// call is an in-flight or completed singleflight.Do call

type call struct {

	wg sync.WaitGroup

	// These fields are written once before the WaitGroup is done

	// and are only read after the WaitGroup is done.

    //val和err用来记录fn发放执行的返回值

	val interface{}

	err error

	// forgotten indicates whether Forget was called with this call's key

	// while the call was still in flight.

    // 用来标识fn方法执行完成之后结果是否立马删除还是保留在singleflight中

	forgotten bool

	// These fields are read and written with the singleflight

	// mutex held before the WaitGroup is done, and are read but

	// not written after the WaitGroup is done.

    //dups 用来记录fn方法执行的次数

	dups  int

    //用来记录DoChan中调用次数以及需要返回的数据

	chans []chan<- Result

}

Group

Group 可以看做是任务的分类。

// Group represents a class of work and forms a namespace in which

// units of work can be executed with duplicate suppression.

type Group struct {

	mu sync.Mutex       // protects m

	m  map[string]*call // lazily initialized

}

Do 函数

// Do executes and returns the results of the given function, making

// sure that only one execution is in-flight for a given key at a

// time. If a duplicate comes in, the duplicate caller waits for the

// original to complete and receives the same results.

// The return value shared indicates whether v was given to multiple callers.

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)

	// 设置forgotten = true, doCall时 不再调用delete(g.m, key)

	// c.forgotten = true

	c.wg.Add(1)

	g.m[key] = c

	g.mu.Unlock()

	g.doCall(c, key, fn)

	return c.val, c.err, c.dups > 0

}

// doCall handles the single call for a key.

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {

	c.val, c.err = fn()

	c.wg.Done()

	g.mu.Lock()

	if !c.forgotten {

		delete(g.m, key)

	}

	for _, ch := range c.chans {

		ch <- Result{c.val, c.err, c.dups > 0}

	}

	g.mu.Unlock()

}

在Do方法中是通过waitgroup来控制的，主要流程如下：

在Group中设置了一个map，如果key不存在，则实例化call(用来保存值信息)，并将key=>call的对应关系存入map中通过mutex保证了并发安全
如果已经在调用中则key已经存在map，则wg.Wait
在fn执行结束之后（在doCall方法中执行）执行wg.Done
卡在第2步的方法得到执行，返回结果

其他的DoChan方法也是类似的逻辑，只是返回的是一个chan。

参考

singleflight包原理解析

使用Golang的singleflight防止缓存击穿

你的鼓励也是我创作的动力

打赏地址

巴特西

[go语言]-深入理解singleflight

前言