Pure Coder

Holiday

0%

go-zero 源码阅读之 SharedCalls

发表于 更新于

最近接触到微服务框架go-zero，大致翻看了下框架代码，感觉整个框架结构清晰、代码简洁，所以决定阅读源码学习下。

在go-zero中SharedCalls的作用是：将并发请求合并成一个请求，以减少对下层服务的压力。

应用场景

  1. 查询缓存时，合并请求，提升服务性能。
    假设有一个 ip 查询的服务，每次用户请求先在缓存中查询一个 ip 的归属地，如果缓存中存在则直接返回结果，不存在则进行 ip 解析操作。

    如上图所示，n 个用户请求查询同一个 ip（127.0.0.1）就会对应 n 个 redis 的查询，在高并发场景下，如果能将 n 个 redis 查询合并成一个 redis 查询，那么性能肯定能提升很多，而SharedCalls就是用来实现请求合并的，效果如下：
  2. 防止缓存击穿。

缓存击穿问题是指，在平常高并发的系统中，大量的请求同时查询一个 key 时，如果这个 key 正好过期失效了，就会导致大量的请求都打到数据库上。这就是缓存击穿。


通过SharedCalls可以将对同一个 Key 的并发请求的其中一个到数据库查询，其他请求共享同一个结果，可以很大程度提升并发能力。

应用方式

直接上代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
func main() {
	round := 10
	var wg sync.WaitGroup
	barrier := syncx.NewSharedCalls()
	wg.Add(round)
	for i := 0; i < round; i++ {
		go func() {
			defer wg.Done()
        // 启用10个协程模拟获取缓存操作
			val, err := barrier.Do("get_rand_int", func() (interface{}, error) {
				time.Sleep(time.Second)
				return rand.Int(), nil
			})
			if err != nil {
				fmt.Println(err)
			} else {
				fmt.Println(val)
			}
		}()
	}
	wg.Wait()
}

通过以上代码，模拟 10 个协程请求 redis 获取一个 key 的内容，代码很简单，就是执行Do方法。其中，接收两个参数，第一个参数是获取资源的标识，可以是 redis 中缓存的 key，第二个参数就是一个匿名函数，封装好要做的业务逻辑。最终获得的结果如下：

1
2
3
4
5
6
7
8
9
10
5577006791947779410
5577006791947779410
5577006791947779410
5577006791947779410
5577006791947779410
5577006791947779410
5577006791947779410
5577006791947779410
5577006791947779410
5577006791947779410

可以看出，10 个协程都获得了同一个结果，也就是只有一个协程真正执行了rand.Int()获取了随机数，其他的协程都共享了这个结果。

源码解析

先看定义的数据结构：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
type (
  // 定义接口，有2个方法Do和DoEx，其实逻辑是一样的，DoEx多了一个bool标识，主要看Do的逻辑就够了
	SharedCalls interface {
		Do(key string, fn func() (interface{}, error)) (interface{}, error)
		DoEx(key string, fn func() (interface{}, error)) (interface{}, bool, error)
	}
	// 定义call的结构，其中waitGroup是精髓
	call struct {
		wg  sync.WaitGroup // 用于实现通过1个call，其他call阻塞
		val interface{}    // 表示call操作的返回结果
		err error          // 表示call操作发生的错误
	}
	// 总控结构，实现SharedCalls接口
	sharedGroup struct {
		calls map[string]*call // 不同的call对应不同的key
		lock  sync.Mutex       // 利用锁控制请求
	}
)

然后看最核心的Do方法做了什么事情：

1
2
3
4
5
6
7
8
9
func (g *sharedGroup) Do(key string, fn func() (interface{}, error)) (interface{}, error) {
	c, done := g.createCall(key)
	if done {
		return c.val, c.err
	}

	g.makeCall(c, key, fn)
	return c.val, c.err
}

代码很简洁，利用g.createCall(key)对 key 发起 call 请求（其实就是做一件事情），如果此时已经有其他协程已经在发起 call 请求就阻塞住（done 为 true），等待拿到结果后直接返回。如果 done 是 false，说明当前协程是第一个发起 call 的协程，那么就执行g.makeCall(c, key, fn)真正地发起 call 请求（此后的其他协程就阻塞在了g.createCall(key))。

看到这里，其实关键就两步：

  1. 判断是第一个请求的协程（利用 map）
  2. 阻塞住其他所有协程（利用 sync.WaitGroup）

来看下g.createCall(key)如何实现的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func (g *sharedGroup) createCall(key string) (c *call, done bool) {
	g.lock.Lock()
	if c, ok := g.calls[key]; ok {
		g.lock.Unlock()
		c.wg.Wait()
		return c, true
	}

	c = new(call)
	c.wg.Add(1)
	g.calls[key] = c
	g.lock.Unlock()

	return c, false
}

先看第一步：判断是第一个请求的协程（利用 map）

1
2
3
4
5
6
g.lock.Lock()
if c, ok := g.calls[key]; ok {
	g.lock.Unlock()
	c.wg.Wait()
	return c, true
}

此处就是利用了判断了 map 中是否有 key，如果已经存在，则认为已经有其他协程在请求了，当前这个协程就只需要等待，并且利用了sync.WaitGroup的Wait()方法，此处还是很巧妙的。要注意的是，map 在 go 中是非并发安全的，所以需要加锁，并且因为只有一个协程去请求，其他协程都阻塞，所以用的是 Mutex，不是 RWMutex。

再看第二步：阻塞住其他所有协程（利用 sync.WaitGroup）

1
2
3
c = new(call)
c.wg.Add(1)
g.calls[key] = c

因为是第一个发起 call 的协程，所以需要 new 这个 call，然后将wg.Add(1)，这样就对应了上面的wg.Wait()，阻塞剩下的协程。然后将 new 的 call 放在 map 中，注意此时只是完成了初始化，并没有真正去执行 call 请求，再往下看。

1
2
3
4
5
6
7
8
9
10
func (g *sharedGroup) makeCall(c *call, key string, fn func() (interface{}, error)) {
	defer func() {
		g.lock.Lock()
		delete(g.calls, key)
		g.lock.Unlock()
		c.wg.Done()
	}()

	c.val, c.err = fn()
}

这个方法中做的事情很简单，就是执行了传递的匿名函数fn()（也就是真正 call 请求要做的事情）。最后就是处理收尾的事情（通过 defer），也是分成两步：

  1. 删除 map 中的 key，使得下次发起请求可以获取新的值。
  2. 调用wg.Done()，让之前阻塞的协程全部获得结果并返回。

至此，SharedCall的核心代码就解析完毕了，虽然代码不长，但是这个思想还是很棒的，可以在实际工作中借鉴。

总结

  • map 非并发安全，记得加锁。
  • 巧用 sync.WaitGroup 去完成需要阻塞控制协程的应用场景。
  • 通过匿名函数 fn 去封装传递具体业务逻辑，在调用 fn 的上层函数中去完成统一的逻辑处理。