大家好，我是 polarisxu。

并發(fā)是 Go 的優(yōu)勢，但并發(fā)也需要很好的進(jìn)行控制。標(biāo)準(zhǔn)庫中有 sync 包，經(jīng)常使用的功能有 sync.Mutex、sync.WaitGroup 等。其實(shí)，除了標(biāo)準(zhǔn)庫，還有一個(gè)官方的擴(kuò)展庫，也叫 sync，其中有一個(gè)子包：sync/singleflight，專門做并發(fā)控制，比如防止緩存擊穿。

01 從例子說起

看一個(gè)模擬緩存的例子，有如下代碼：

1. package main
3. import (
4. "errors"
5. "flag"
6. "log"
7. "sync"
8. )
10. var errorNotExist = errors.New("not exist")
12. var n int
14. func init() {
15. flag.IntVar(&n, "n", 5, "模擬的并發(fā)數(shù)，默認(rèn) 5")
16. }
18. func main() {
19. flag.Parse()
21. var wg sync.WaitGroup
22. wg.Add(n)
24. // 模擬并發(fā)訪問
25. for i := 0; i < n; i++ {
26. go func() {
27. defer wg.Done()
28. // 假設(shè)都獲取 id = 1 這篇文章
29. article := fetchArticle(1)
30. log.Println(article)
31. }()
32. }
33. wg.Wait()
34. }
36. type Article struct {
37. ID int
38. Content string
39. }
42. func fetchArticle(id int) *Article {
43. article := findArticleFromCache(id)
45. if article != nil && article.ID > 0 {
46. return article
47. }
49. return findArticleFromDB(id)
50. }
52. var (
53. cache = make(map[int]*Article)
54. rwmutex sync.RWMutex
55. )
57. // 模擬從緩存獲取數(shù)據(jù)
58. func findArticleFromCache(id int) *Article {
59. rwmutex.RLock()
60. defer rwmutex.RUnlock()
61. return cache[id]
62. }
64. // 模擬從數(shù)據(jù)庫中獲取數(shù)據(jù)
65. func findArticleFromDB(id int) *Article {
66. log.Printf("SELECT * FROM article WHERE id=%d", id)
67. article := &Article{ID: id, Content: "polarisxu"}
68. rwmutex.Lock()
69. defer rwmutex.Unlock()
70. cache[id] = article
71. return article
72. }

我們模擬 5 個(gè)用戶并發(fā)訪問，同時(shí)獲取 ID=1 的文章，因?yàn)榫彺嬷胁淮嬖?，因此都到后?DB 獲取具體數(shù)據(jù)。從運(yùn)行結(jié)果可以看出這一點(diǎn)：

1. $ go run main.go
2. 2021/12/30 10:32:36 SELECT * FROM article WHERE id=1
3. 2021/12/30 10:32:36 SELECT * FROM article WHERE id=1
4. 2021/12/30 10:32:36 &{1 polarisxu}
5. 2021/12/30 10:32:36 &{1 polarisxu}
6. 2021/12/30 10:32:36 SELECT * FROM article WHERE id=1
7. 2021/12/30 10:32:36 &{1 polarisxu}
8. 2021/12/30 10:32:36 SELECT * FROM article WHERE id=1
9. 2021/12/30 10:32:36 &{1 polarisxu}
10. 2021/12/30 10:32:36 SELECT * FROM article WHERE id=1
11. 2021/12/30 10:32:36 &{1 polarisxu}

顯然這是我們不希望看到的。

02 使用 singleflight

官方的擴(kuò)展包 golang.org/x/sync 下面有一個(gè)子包 singleflight：

1. Package singleflight provides a duplicate function call suppression mechanism.

它用來抑制函數(shù)的重復(fù)調(diào)用，這正好符合上面的場景：希望從數(shù)據(jù)庫獲取數(shù)據(jù)的函數(shù)只調(diào)用一次。

將 fetchArticle 函數(shù)改成這樣：

1. var g singleflight.Group
3. func fetchArticle(id int) *Article {
4. article := findArticleFromCache(id)
6. if article != nil && article.ID > 0 {
7. return article
8. }
10. v, err, shared := g.Do(strconv.Itoa(id), func() (interface{}, error) {
11. return findArticleFromDB(id), nil
12. })
14. // 打印 shared，看看都什么值
15. fmt.Println("shared===", shared)
17. if err != nil {
18. log.Println("singleflight do error:", err)
19. return nil
20. }
22. return v.(*Article)
23. }

singleflight.Group 是一個(gè)結(jié)構(gòu)體類型，沒有導(dǎo)出任何字段，它代表一類工作并形成一個(gè)命名空間，在該命名空間中可以抑制工作單元的重復(fù)執(zhí)行。

該類型有三個(gè)方法，它們的功能見注釋：

1. // 執(zhí)行并返回給定函數(shù)的結(jié)果，確保對于給定的鍵，fn 函數(shù)只會(huì)執(zhí)行一次。
2. // 如果有重復(fù)的進(jìn)來，重復(fù)的調(diào)用者會(huì)等待最原始的調(diào)用完成并收到相同的結(jié)果。
3. // 返回值 shared 指示是否將 v 提供給多個(gè)調(diào)用者。
4. // 返回值 v 是 fn 的執(zhí)行結(jié)果
5. // 返回值 err 是 fn 返回的 err
6. func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool)
7. // 和 Do 類似，但返回一個(gè) channel（只能接收），用來接收結(jié)果。Result 是一個(gè)結(jié)構(gòu)體，有三個(gè)字段，即 Do 返回的那三個(gè)。
8. func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result
9. func (g *Group) Forget(key string)

因此，改后的代碼，通過 Group.Do，即使并發(fā)多次調(diào)用，findArticleFromDB 也只會(huì)執(zhí)行一次，并且這一次的結(jié)果會(huì)被并發(fā)多次執(zhí)行共享。

運(yùn)行后，結(jié)果如下：

1. $ go run main.go
2. 2021/12/30 11:55:44 SELECT * FROM article WHERE id=1
3. shared=== true
4. 2021/12/30 11:55:44 &{1 polarisxu}
5. shared=== true
6. 2021/12/30 11:55:44 &{1 polarisxu}
7. shared=== true
8. 2021/12/30 11:55:44 &{1 polarisxu}
9. shared=== true
10. 2021/12/30 11:55:44 &{1 polarisxu}
11. shared=== true
12. 2021/12/30 11:55:44 &{1 polarisxu}

和預(yù)期一樣，findArticleFromDB 只執(zhí)行了一次，shared 的值也表示結(jié)果被多個(gè)調(diào)用者共享。

所以，使用 Go 后，再也不需要通過類似 Redis 中的 SETNX 這樣的命令來實(shí)現(xiàn)類似的功能了。

03 Forget 的用途

上面 Group 的方法中，有一個(gè)沒有給任何注釋，即 Forget。從名字猜到，用來忘掉什么，那具體什么意思呢?

通過上面的例子，我們知曉，通過 Do，可以實(shí)現(xiàn)多個(gè)并發(fā)調(diào)用只執(zhí)行回調(diào)函數(shù)一次，并共享相同的結(jié)果。而 Forget 的作用是：

Forget tells the singleflight to forget about a key. Future calls to Do for this key will call the function rather than waiting for an earlier call to complete.

即告訴 singleflight 忘記一個(gè) key，未來對此 key 的 Do 調(diào)用將調(diào)用 fn 回調(diào)函數(shù)，而不是等待更早的調(diào)用完成，即相當(dāng)于廢棄 Do 原本的作用。

可以在上面例子中 Do 調(diào)用之前，調(diào)用 g.Forget，驗(yàn)證是否 Do 的調(diào)用都執(zhí)行 fn 函數(shù)即 findArticleFromDB 函數(shù)了。

04 泛型版本

細(xì)心的讀者可能會(huì)發(fā)現(xiàn)，Do 方法返回的 v 是 interface{}，在 fetchArticle 函數(shù)最后，我們做了類型斷言：v.(*Article)。

既然 Go1.18 馬上要來了，有了泛型，可以有泛型版本的 singleflight，不需要做類型斷言了。GitHub 已經(jīng)有人實(shí)現(xiàn)并開源：https://github.com/marwan-at-work/singleflight。

改成這個(gè)泛型版本，要改以下幾處：

• 導(dǎo)入包 marwan.io/singleflight，而非 github.com/marwan-at-work/singleflight，同時(shí)移除 golang.org/x/sync/singleflight
• g 的聲明改為：var g singleflight.Group[*Article]
• Do 的調(diào)用，返回值由 interface{} 類型改為：*Article：

1. article, err, shared := g.Do(strconv.Itoa(id), func() (*Article, error) {
2. return findArticleFromDB(id), nil
3. })

• 最后返回時(shí)，直接返回 article，不需要做類型斷言

05 總結(jié)

singleflight 很常用，你在 pkg.go.dev 搜索 singleflight，發(fā)現(xiàn)有很多輪子：https://pkg.go.dev/search?q=singleflight，好些項(xiàng)目不是使用官方的 golang.org/x/sync/singleflight，而是自己實(shí)現(xiàn)一個(gè)，不過這些實(shí)現(xiàn)基本只實(shí)現(xiàn)了最常用的 Do 方法。感興趣的可以查看他們的實(shí)現(xiàn)。

下次項(xiàng)目中需要類似功能，記得使用 singleflight 哦!

原文鏈接：https://mp.weixin.qq.com/s/zwTErb_kiEEsurscYQ6eRw