Monthly Archives: November 2020

Golang中的并发原语 Singleflight

Posted on by

在Golang中有一个并发原语是Singleflight，好像知道的开发者并不多。其中著名的 https://github.com/golang/groupcache 就用到了这个并发原语。

Golang版本

go1.15.5

相关知识点

map、Mutex、channel、

使用场景

一般用在对指定资源频繁操作的情况下，如高并发下的“缓存击穿”问题。

缓存击穿：一个存在的key，在缓存过期的瞬间，同时有大量的请求过来，造成所有请求都去DB读取数据，这些请求都会击穿缓存到DB，造成瞬时DB请求量大、压力瞬间骤增，导致数据库负载过高，影响整个系统正常运行。（缓存击穿不同于 缓存雪崩 和 缓存穿透）

缓存击穿

怎么理解这个原语呢，简单的讲就是将对同一个资源的多个请求合并为一个请求。

举例说明，假如当有10万个请求来获取同一个key的值的时候，正常情况下会执行10万次get操作。而使用singleflight并发语后，只需要首次的地个请求执行一次get操作就可以了，其它请求再过来时，只需要只需要等待即可。待执行结果返回后，再把结果分别返回给等待中的请求，每个请求再返回给客户端，由此看看，在一定的高并发场景下可以大大减少系统的负载，节省大量的资源。

注意这个与 sync.Once是不一样的，sync.Once 是全局只能有一个，但本并发原语则是根据key来划分的，并且可以根据需求来决定什么情况下共用一个。

实现原理

主要使用 Mutext 和 Map 来实现，以 key 为键，值为 *call。每个*call中存储有一个请求 chans 字段，用来存储所有请求此key的客户端，等有返回结果的时候，再从chans字段中读取出来，分别写入即可。

源文件为 /src/internal/singleflight/singleflight.go

Singleflight 数据结构如下

  1. Do()
    这个方法是一个执行函数并返回执行结果
    参数
    key 要请求的key，多个请求可能请求的是同一个key，同时也只有一个函数在执行，
    fn key对应的执行函数，此函数有三个返回值v, err, shared。其中shared表示当前返回结果值是只有一个请求还是由于多个请求发生返回的
  2. DoChan()
    类型与Do()方法，但返回的是个 ch 类型，等函数 fn 执行后，可以通过读取返回的ch来获取函数结果
  3. Forget()
    在官方库internal/singleflight/singleflight.go中这个名字是ForgetUnshared， 告诉 Group 忘记请求的这个key，下次再请求时，直接当作新的key来处理就可以了。其实就是将这个key从map中删除，后续再有这个key的操作的话，即视为新一轮的处理逻辑。

其中Do() 和 DoChan() 的功能一样，只是获取数据的方式有所区别，开发者可以根据自己的情况来选择使用哪一个。另外还包含一个由Do()方法调用的私有方法 doCall()，直接执行key处理方法的函数

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

实现数据结构

实现内部主要的数据结构一共三个，分别是 call、Group 和 Result。

// 定义 map[key] 的值 call 类型
// 在call 内部存储对应key相关的所有请求客户端信息
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.
        // 请求返回结果，会在sync.WaitGroup为Done的时候执行
	val interface{}
	err error

	// 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.
        // 可以理解为请求key的个数，每增加一个请求，则值加1 
	dups  int

        // 存储所有key对应的 Result{}, 一个请求对应一个 Result
	chans []chan<- Result
}

// Group represents a class of work and forms a namespace in
// which units of work can be executed with duplicate suppression.
// SingleFlight 的主结构体
type Group struct {
	mu sync.Mutex       // protects m
	m  map[string]*call // lazily initialized
}

// Result holds the results of Do, so they can be passed
// on a channel.
// 定义请求结果数据结构
type Result struct {
	Val    interface{}
	Err    error
	Shared bool
}

call 数据结构是用来记录key相关数据，则请求当前key的个数和请求结果
Group 是并发原语 SingleFlight 的主数据结构, m 字段用来存储key与请求的关系，而mu则是Mutex锁
Result 是请求结果的数据结构，其中Val和Err 是实现请求后返回的结果，而 Shared 字段则是用来表示当前结果是否进行了合并处理，如果当前key只有一个请求在使用的话，则返回false, 否则返回 true

用法

package main

import (
	"fmt"
	"golang.org/x/sync/singleflight"
	"log"
	"sync"
	"time"
)

func getDataFromDB(key string) (string, error) {
	defer func() {
		log.Println("db query end")
	}()

	log.Println("db query begin")
	time.Sleep(2 * time.Second)
	result := key + "abcxyz"

	return result, nil
}

func main() {
	var singleRequest singleflight.Group

	getData := func(requestID int, key string) (string, error) {
		log.Printf("request %v start request ...", requestID)

		// 合并请求
		value, _, _ := singleRequest.Do(key, func() (ret interface{}, err error) {
			log.Printf("request %v is begin...", requestID)
			ret, err = getDataFromDB(key)
			log.Printf("request %v end!", requestID)
			return
		})

		return value.(string), nil
	}

	var wg sync.WaitGroup
	key := "orderID"
	for i := 1; i < 10; i++ {
		wg.Add(1)
		go func(wg *sync.WaitGroup, requestID int) {
			defer wg.Done()
			value, _ := getData(requestID, key)
			log.Printf("request %v get value: %v", requestID, value)
		}(&amp;wg, i)
	}
	wg.Wait()

	fmt.Println("main end")
}

输出结果

2020/11/20 12:11:44 request 6 start request …
2020/11/20 12:11:44 request 6 is begin…
2020/11/20 12:11:44 request 2 start request …
2020/11/20 12:11:44 request 9 start request …
2020/11/20 12:11:44 request 7 start request …
2020/11/20 12:11:44 request 3 start request …
2020/11/20 12:11:44 request 4 start request …
2020/11/20 12:11:44 request 1 start request …
2020/11/20 12:11:44 request 5 start request …
2020/11/20 12:11:44 request 8 start request …
2020/11/20 12:11:44 db query begin
2020/11/20 12:11:46 db query end
2020/11/20 12:11:46 request 6 end!
2020/11/20 12:11:46 request 6 get value: orderIDabcxyz
2020/11/20 12:11:46 request 8 get value: orderIDabcxyz
2020/11/20 12:11:46 request 2 get value: orderIDabcxyz
2020/11/20 12:11:46 request 9 get value: orderIDabcxyz
2020/11/20 12:11:46 request 7 get value: orderIDabcxyz
2020/11/20 12:11:46 request 3 get value: orderIDabcxyz
2020/11/20 12:11:46 request 4 get value: orderIDabcxyz
2020/11/20 12:11:46 request 1 get value: orderIDabcxyz
2020/11/20 12:11:46 request 5 get value: orderIDabcxyz
main end

从上面的输出内容可以看到，我们同时发起了10个gorourtine来查询同一个订单信息，真正在DB 层查询操作只有一次，大大减少了DB的压力。

总结

使用SingleFlight时，在高并发下且对少量key频繁读取的话，可以大大减少服务器负载。在一定场景下特别的有用，如 CDN。