講完Go并發(fā)控制，講講并發(fā)抑制
已知有一個(gè)函數(shù)search，能夠按照關(guān)鍵詞執(zhí)行搜索，coSearch能夠批量并發(fā)查詢。
讓我們把目光定位到search上，search通過(guò)查詢數(shù)據(jù)庫(kù)或者調(diào)用其他api來(lái)完成搜索，這是一個(gè)相對(duì)耗時(shí)和消耗資源的操作。
當(dāng)多個(gè)相同的關(guān)鍵詞并發(fā)查詢（調(diào)用search函數(shù)）時(shí)，我們希望只產(chǎn)生一次數(shù)據(jù)庫(kù)調(diào)用（調(diào)用query），第一個(gè)查詢未完成時(shí)后續(xù)的重復(fù)查詢會(huì)等待，當(dāng)?shù)谝粋€(gè)查詢完成時(shí)則會(huì)與其他查詢分享結(jié)果，這樣一來(lái)雖然只執(zhí)行了一次數(shù)據(jù)庫(kù)調(diào)用但是所有查詢都拿到了最終的結(jié)果。
圖片
什么是并發(fā)抑制：
package main
import (
"context"
"fmt"
"sync"
"time"
"golang.org/x/sync/errgroup"
)
func query(ctx context.Context, word string) (string, error) {
fmt.Println("searching: ", word)
time.Sleep(5 * time.Second)
return fmt.Sprintf("result: %s", word), nil // 模擬結(jié)果
}
// 實(shí)現(xiàn)search，在重復(fù)并發(fā)調(diào)用下僅執(zhí)行一次query
// 其他并發(fā)共享這次query的結(jié)果
func search(ctx context.Context, word string) (string, error) {
return query(ctx, word)
}
func coSearch(ctx context.Context, words []string) ([]string, error) {
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(10)
results := make([]string, len(words))
for i, word := range words {
i, word := i, word
g.Go(func() error {
result, err := search(ctx, word)
if err != nil {
return err
}
results[i] = result
return nil
})
}
err := g.Wait()
return results, err
}
func main() {
words := []string{"Go","Go", "Go", "Rust", "PHP", "JavaScript", "Java"}
results, err := coSearch(context.Background(), words)
if err != nil {
fmt.Println(err)
return
}
fmt.Println(results)
}好了，可以先暫停想想該如何實(shí)現(xiàn)search函數(shù)了。
一步一步實(shí)現(xiàn)并發(fā)抑制
我們先假設(shè)所有查詢關(guān)鍵詞都一樣，那么問(wèn)題簡(jiǎn)化成并發(fā)執(zhí)行search時(shí)，只在第一次search時(shí)調(diào)用query，其他的search并發(fā)調(diào)用等待并共享這次的查詢結(jié)果。
通過(guò)waiting變量，其他goroutine等待第一個(gè)goroutine數(shù)據(jù)庫(kù)調(diào)用完成，那么如何讓其他goroutine等待在這個(gè)位置呢？
func main() {
words := []string{"Go", "Go", "Go", "Go", "Go"}
results, err := coSearch(context.Background(), words)
if err != nil {
fmt.Println(err)
return
}
fmt.Println(results)
}
var (
waiting bool
resp string
err error
)
func search(ctx context.Context, word string) (string, error) {
if waiting {
// 等待resp, err被賦值，即第一個(gè)query完成后再返回
// ...?
return resp, err
}
waiting = true
resp, err = query(ctx, word)
waiting = false
return resp, err
}
func query(ctx context.Context, word string) (string, error) {
fmt.Println("searching: ", word)
time.Sleep(5 * time.Second)
return fmt.Sprintf("result: %s", word), nil // 模擬結(jié)果
}sync.WaitGroup{}并發(fā)控制
sync.WaitGroup{}是并發(fā)控制的核心，這里再次重申下用法：
- 當(dāng)新運(yùn)行一個(gè)goroutine時(shí)，我們需要調(diào)用wg.Add(1)。
- 當(dāng)一個(gè)goroutine運(yùn)行完成的時(shí)候，我們需要調(diào)用wg.Done()。
- wg.Wait()讓程序阻塞在此處，直到所有的goroutine運(yùn)行完畢。
利用 sync.WaitGroup{}便可實(shí)現(xiàn)上文代碼中等待的效果：
var (
wg sync.WaitGroup
waiting bool
resp string
err error
)
func search(ctx context.Context, word string) (string, error) {
if waiting {
// 其他goroutine等待第一個(gè)goroutine執(zhí)行完成
wg.Wait()
return resp, err
}
waiting = true
wg.Add(1)
resp, err = query(ctx, word)
wg.Done() // 第一個(gè)goroutine執(zhí)行完成
waiting = false
return resp, err
}并發(fā)安全
當(dāng)多個(gè)goroutine對(duì)同一個(gè)內(nèi)存區(qū)域進(jìn)行讀寫(xiě)時(shí)，就會(huì)產(chǎn)生并發(fā)安全的問(wèn)題，它會(huì)導(dǎo)致程序運(yùn)行的結(jié)果不符合預(yù)期，而上文的程序并發(fā)的讀寫(xiě)了waiting變量，需要給waiting變量加把鎖。
釋放鎖的位置非常的有技巧，如果在在wg.Add(1)之前mu.Unlock()，可能 wg.Add(1)還未來(lái)得執(zhí)行其他goroutine已經(jīng)執(zhí)行了wg.Wait()，并獲取到了錯(cuò)誤的數(shù)據(jù)。
unlock在add之前；
var (
wg sync.WaitGroup
mu sync.Mutex
waiting bool
resp string
err error
)
func search(ctx context.Context, word string) (string, error) {
mu.Lock()
if waiting {
mu.Unlock()
wg.Wait()
return resp, err
}
waiting = true
wg.Add(1)
// 在wg.Add(1)之后釋放鎖，保證其他goroutine被wg.Wait()阻塞
mu.Unlock()
resp, err = query(ctx, word)
wg.Done()
mu.Lock()
waiting = false
mu.Unlock()
return resp, err
}完整版本
現(xiàn)在可以針對(duì)不同的關(guān)鍵詞做區(qū)分了，使用一個(gè)map來(lái)代替原有的waiting，并將每一個(gè)關(guān)鍵詞查詢的WaitGroup和結(jié)果打包到map的value中。
type call struct {
wg sync.WaitGroup
resp string
err error
}
var (
mu sync.Mutex
m = make(map[string]*call)
)
func search(ctx context.Context, word string) (string, error) {
mu.Lock()
if c, ok := m[word]; ok {
mu.Unlock()
c.wg.Wait()
return c.resp, c.err
}
c := &call{}
m[word] = c
c.wg.Add(1)
// 在wg.Add(1)之后才釋放鎖，保證其他goroutine被wg.Wait()阻塞
mu.Unlock()
c.resp, c.err = query(ctx, word)
c.wg.Done()
mu.Lock()
delete(m, word)
mu.Unlock()
return c.resp, c.err
}開(kāi)源庫(kù) golang.org/x/sync/singleflight
上面一步一步教大家手搓了一個(gè)并發(fā)抑制的邏輯，我們的基本邏輯和開(kāi)源庫(kù)golang.org/x/sync/singleflight沒(méi)有區(qū)別，只是singleflight內(nèi)部實(shí)現(xiàn)更加嚴(yán)謹(jǐn)
直接使用singleflight非常簡(jiǎn)單的就可以實(shí)現(xiàn)我們的訴求
- singleflight.Group 創(chuàng)建一個(gè)需要并發(fā)控制的范圍
- Do函數(shù)
第一個(gè)參數(shù)接收一個(gè)key來(lái)判斷否重復(fù)調(diào)用
第二個(gè)參數(shù)為要執(zhí)行的函數(shù)，函數(shù)可以返回正常值或者error
Do函數(shù)返回值除了閉包函數(shù)的返回值之外，還返回了此次返回值是否由其他goroutine共享
import (
"golang.org/x/sync/singleflight"
)
var g = new(singleflight.Group)
func search(ctx context.Context, word string) (string, error) {
resp, err, _ := g.Do(word, func() (interface{}, error) {
return query(ctx, word)
})
return resp.(string), err
}錯(cuò)誤處理
因?yàn)楣蚕淼谝粋€(gè)goroutine的結(jié)果，因此如果第一次調(diào)用失敗，那其他goroutine也都會(huì)失敗
如果在某些場(chǎng)景下允許第一個(gè)調(diào)用失敗后再次嘗試調(diào)用該函數(shù)，那么可以通過(guò)調(diào)用Forget方法來(lái)忘記這個(gè)key
var g = new(singleflight.Group)
func search(ctx context.Context, word string) (string, error) {
resp, err, _ := g.Do(word, func() (interface{}, error) {
val, err := query(ctx, word)
// 當(dāng)出錯(cuò)并且允許重試時(shí)
if err != nil && true {
g.Forget(word)
return "", err
}
return val, err
})
return resp.(string), err
}超時(shí)控制
當(dāng)使用Do函數(shù)時(shí)，如果query長(zhǎng)時(shí)間未響應(yīng)（這里假設(shè)qeury不具備超時(shí)能力），那么所有的goroutine都會(huì)被阻塞并等待，利用DoChan+select可以實(shí)現(xiàn)超時(shí)邏輯
var g = new(singleflight.Group)
func search(ctx context.Context, word string) (string, error) {
ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
defer cancel()
result := g.DoChan(word, func() (interface{}, error) {
return query(ctx, word)
})
select {
case r := <-result:
return r.Val.(string), r.Err
case <-ctx.Done():
return "", ctx.Err()
}
}使用場(chǎng)景
預(yù)防緩存穿透
在高并發(fā)的狀態(tài)下，一般會(huì)給熱點(diǎn)數(shù)據(jù)設(shè)置緩存。但數(shù)據(jù)第一次訪問(wèn)或者緩存失效的狀態(tài)下，如果直接去查詢數(shù)據(jù)庫(kù)，會(huì)給數(shù)據(jù)庫(kù)造成極大壓力，甚至直接打爆數(shù)據(jù)庫(kù)。
以上各種分享中被反復(fù)提到的場(chǎng)景，但！注意！使用singleflight就一勞永逸了么，不是的，在大規(guī)模集群下可能有數(shù)百臺(tái)機(jī)器，當(dāng)處在高并發(fā)狀態(tài)時(shí)，即使每臺(tái)機(jī)器只發(fā)起一個(gè)請(qǐng)求，也足以打爆你的數(shù)據(jù)庫(kù)！結(jié)合實(shí)際，搭配適當(dāng)?shù)木彺娌呗?、?shù)據(jù)預(yù)熱、限流等手段才能避免潛在的風(fēng)險(xiǎn)。挖個(gè)坑，以后有機(jī)會(huì)聊聊這些問(wèn)題
總結(jié)
本篇作為一個(gè)例子，給你講透典型的Go并發(fā)控制的姊妹篇，講述了另外一種并發(fā)控制模型，并介紹了開(kāi)源庫(kù)golang.org/x/sync/singleflight。
當(dāng)由一個(gè)goroutine并發(fā)向下發(fā)展成多個(gè)goroutine時(shí)，使用golang.org/x/sync/errgroup
當(dāng)多個(gè)goroutine并發(fā)向下抑制成一個(gè)goroutine時(shí)，使用golang.org/x/sync/singleflight
