【Go】singleflight 100行代码解决缓存击穿问题

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

package main

import (
	"log"
	"sync"
	"time"

	"golang.org/x/sync/singleflight"
)

func main() {
	var singleSetCache singleflight.Group

	var wg sync.WaitGroup
	cacheKey := "cacheKey"
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			value, err, shared := singleSetCache.Do(cacheKey, func() (ret interface{}, err error) { //do的入参key，可以直接使用缓存的key，这样同一个缓存，只有一个协程会去读DB
				log.Printf("%v exec", i)
				time.Sleep(3 * time.Second)
				return "value", nil
			})
			log.Printf("value %v, err %v, shared:%v", value.(string), err, shared)
		}(i)
	}
	wg.Wait()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

2021/01/05 23:18:32 0 exec
2021/01/05 23:18:35 value value, err <nil>, shared:true
2021/01/05 23:18:35 value value, err <nil>, shared:true
2021/01/05 23:18:35 value value, err <nil>, shared:true
2021/01/05 23:18:35 value value, err <nil>, shared:true
2021/01/05 23:18:35 value value, err <nil>, shared:true
2021/01/05 23:18:35 value value, err <nil>, shared:true
2021/01/05 23:18:35 value value, err <nil>, shared:true
2021/01/05 23:18:35 value value, err <nil>, shared:true
2021/01/05 23:18:35 value value, err <nil>, shared:true
2021/01/05 23:18:35 value value, err <nil>, shared:true

1
2
3
4
5

func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool)

func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result

func (g *Group) Forget(key string)

1
2
3
4
5
6
7
8

type Resolver struct {
  ......
 // 源码地址 https://github.com/golang/go/blob/master/src/net/lookup.go#L151
 // lookupGroup merges LookupIPAddr calls together for lookups for the same
 // host. The lookupGroup key is the LookupIPAddr.host argument.
 // The return values are ([]IPAddr, error).
 lookupGroup singleflight.Group
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

func LookupIP(host string) ([]IP, error) {
 addrs, err := DefaultResolver.LookupIPAddr(context.Background(), host)
  ......
}

func (r *Resolver) lookupIPAddr(ctx context.Context, network, host string) ([]IPAddr, error) {
  ......
  // 使用SingleFlight的DoChan合并多个查询请求
 ch, called := r.getLookupGroup().DoChan(lookupKey, func() (interface{}, error) {
  defer dnsWaitGroup.Done()
  return testHookLookupIP(lookupGroupCtx, resolverFunc, network, host)
 })
 if !called {
  dnsWaitGroup.Done()
 }
 
 select {
 case <-ctx.Done():
  ......
 case r := <-ch:
  lookupGroupCancel()
  if trace != nil && trace.DNSDone != nil {
   addrs, _ := r.Val.([]IPAddr)
   trace.DNSDone(ipAddrsEface(addrs), r.Shared, r.Err)
  }
  return lookupIPReturn(r.Val, r.Err, r.Shared)
 }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// 模拟一个Redis客户端
type client struct {
  // ... 其他的配置省略
  requestGroup singleflight.Group
}

// 普通查询
func (c *client) Get(key string) (interface{}, error) {
  fmt.Println("Querying Database")
  time.Sleep(time.Second)
  v := "Content of key" + key
  return  v, nil
}

// SingleFlight查询
func (c *client) SingleFlightGet(key string) (interface{}, error) {
  v, err, _ := c.requestGroup.Do(key, func() (interface{}, error) {
    return c.Get(key)

  })
  if err != nil {
    return nil, err
  }
  return v, err
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

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
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

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 {
    // 存在相同的key, 增加计数
    c.dups++
    g.mu.Unlock()
    c.wg.Wait() //等待这个key对应的fn调用完成
    return c.val, c.err, true // 返回fn调用的结果
  }
  c := new(call) // 不存在key, 是第一个请求, 创建一个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
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

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 { // 已调用完，删除这个key
    delete(g.m, key)
  }
  for _, ch := range c.chans {
    ch <- Result{c.val, c.err, c.dups > 0}
  }
  g.mu.Unlock()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

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)s
  g.m[key] = c
  g.mu.Unlock()

  go g.doCall(c, key, fn)

  return ch
}

type Result struct {
  Val    interface{}
  Err    error
  Shared bool
}