源碼剖析Golang中singleflight的應(yīng)用

更新時(shí)間：2024年03月29日 08:34:54 作者：陳明勇

這篇文章主要為大家詳細(xì)介紹了如何利用singleflight來避免緩存擊穿,并剖析singleflight包的源碼實(shí)現(xiàn)和工作原理,感興趣的可以了解下

前言

前面的一篇文章 Go singleflight：防緩存擊穿利器詳細(xì)介紹 singleflight 包的使用，展示如何利用它來避免緩存擊穿。而本篇文章，我們來剖析 singleflight 包的源碼實(shí)現(xiàn)和工作原理，探索單飛的奧秘。

singleflight 版本：golang.org/x/sync v0.6.0

結(jié)構(gòu)體解析

Group

Group 是 singleflight 包的一個(gè)核心結(jié)構(gòu)體，它管理著所有的請(qǐng)求，確保同一時(shí)刻，對(duì)同一資源的請(qǐng)求只會(huì)被執(zhí)行一次。該結(jié)構(gòu)體的源碼如下所示：

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

Group 結(jié)構(gòu)體有兩個(gè)字段：

mu sync.Mutex：一個(gè)互斥鎖，用于保護(hù)下面的 m 映射。在并發(fā)環(huán)境下，多個(gè) goroutine 可能會(huì)同時(shí)對(duì) m 進(jìn)行讀寫操作，所以需要通過互斥鎖來確保對(duì) m 的操作是安全的。
m map[string]*call：一個(gè) map 映射，其鍵是字符串，值是指向 call 結(jié)構(gòu)體的指針。Do 和 DoCHan 方法的參數(shù)里，其中一個(gè)是 key，m 的鍵保存的就是這個(gè) key。m 是惰性初始化的，意味著它在第一次使用時(shí)才會(huì)被創(chuàng)建。

Group 通過維護(hù) m 字段來跟蹤每個(gè) key 的調(diào)用狀態(tài)，從而實(shí)現(xiàn)將多個(gè)請(qǐng)求合并成一個(gè)請(qǐng)求，多個(gè)請(qǐng)求共享同一個(gè)結(jié)果。

call

call 結(jié)構(gòu)體表示一個(gè)針對(duì)特定 key 的正在進(jìn)行中或者已完成的請(qǐng)求，它確保所有同時(shí)對(duì)該key調(diào)用 Do 或 DoChan 方法的 goroutine 共享同一個(gè)執(zhí)行結(jié)果。該結(jié)構(gòu)體的源碼如下所示：

// call is an in-flight or completed singleflight.Do call
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.
    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.
    dups  int
    chans []chan<- Result
}

call 結(jié)構(gòu)體有五個(gè)字段：

wg sync.WaitGroup：一個(gè)等待組，用于等待當(dāng)前 call 的完成。當(dāng)調(diào)用 Do 或 DoChan 方法后，內(nèi)部會(huì)增加 WaitGroup 的計(jì)數(shù)器，當(dāng)調(diào)用完成后，會(huì)減少計(jì)數(shù)器。在調(diào)用完成之前，其他想要獲取當(dāng)前 call 的結(jié)果的 goroutine 會(huì)等待 WaitGroup 的完成。
val interface{}：調(diào)用 Do 或 DoChan 方法的結(jié)果值之一，對(duì)應(yīng)著 fn 函數(shù)（Do 或 DoChan 方法的參數(shù)）的第一個(gè)返回值 val。這個(gè)字段在 WaitGroup 完成之前被寫入一次，只有在 WaitGroup 完成后才會(huì)被讀取。
err error：這是在調(diào)用 Do 或者 DoChan 方法時(shí)可能發(fā)生的錯(cuò)誤。和 val 類似，這個(gè)字段在 WaitGroup 完成之前被寫入一次，只有在 WaitGroup 完成后才會(huì)被讀取。
dups int：用于計(jì)數(shù)當(dāng)前 call 的重復(fù)調(diào)用數(shù)量。這個(gè)計(jì)數(shù)是在 singleflight 的互斥鎖保護(hù)下進(jìn)行的，在 WaitGroup 完成之前可以讀寫，在 WaitGroup 完成后只能讀取。目前該字段的作用是判斷 call 的結(jié)果是否被共享。
chans []chan<- Result：一個(gè)通道切片，用于存儲(chǔ)所有等待當(dāng)前 call 結(jié)果的通道。這些通道在 call 完成時(shí)接收到結(jié)果。這個(gè)字段同樣是在 singleflight 的互斥鎖保護(hù)下進(jìn)行的，在 WaitGroup 完成之前可以讀寫，在 WaitGroup 完成后只能讀取。

一句話概括就是：call 結(jié)構(gòu)體用于跟蹤 Do 或 DoChan 方法的調(diào)用狀態(tài)，包括等待其完成的 goroutine、調(diào)用的結(jié)果、發(fā)生的錯(cuò)誤以及跟蹤重復(fù)的調(diào)用次數(shù)，對(duì)于 singleflight 在共享調(diào)用結(jié)果中起到關(guān)鍵作用。

Result

Result 是一個(gè)封裝了請(qǐng)求調(diào)用結(jié)果的結(jié)構(gòu)體，在DoChan 方法返回結(jié)果時(shí)使用。該結(jié)構(gòu)體的源碼如下所示：

// Result holds the results of Do, so they can be passed
// on a channel.
type Result struct {
    Val    interface{}
    Err    error
    Shared bool
}

Result 結(jié)構(gòu)體有三個(gè)字段：

Val interface{}：存儲(chǔ) DoChan 方法調(diào)用的結(jié)果值之一，對(duì)應(yīng)著 fn 函數(shù)（DoChan 方法的參數(shù)）的第一個(gè)返回值 val。
Err error：存儲(chǔ) DoChan 方法調(diào)用過程中可能發(fā)生的錯(cuò)誤。
Shared bool：表示調(diào)用結(jié)果是否被多個(gè)請(qǐng)求（goroutine）共享。該字段的值，取決于 call 結(jié)構(gòu)體的 dups 字段值，如果 dups 大于 0，Shared 的值則為 true，否則為 false。

panicError

panicError 用于封裝從 panic 中恢復(fù)的任意值和在給定函數(shù)執(zhí)行期間產(chǎn)生的堆棧跟蹤信息。該結(jié)構(gòu)體的源碼如下所示：

// A panicError is an arbitrary value recovered from a panic
// with the stack trace during the execution of given function.
type panicError struct {
    value interface{}
    stack []byte
}

// Error implements error interface.
func (p *panicError) Error() string {
    return fmt.Sprintf("%v\n\n%s", p.value, p.stack)
}

func (p *panicError) Unwrap() error {
    err, ok := p.value.(error)
    if !ok {
        return nil
    }

    return err
}

func newPanicError(v interface{}) error {
    stack := debug.Stack()

    // The first line of the stack trace is of the form "goroutine N [status]:"
    // but by the time the panic reaches Do the goroutine may no longer exist
    // and its status will have changed. Trim out the misleading line.
    if line := bytes.IndexByte(stack[:], '\n'); line >= 0 {
            stack = stack[line+1:]
    }
    return &panicError{value: v, stack: stack}
}

字段

panicError 結(jié)構(gòu)體有兩個(gè)字段：

value interface{}：存儲(chǔ)從 panic 中恢復(fù)的值，這個(gè)值是任意類型的，可能是 error 類型，也可能是其它類型。
stack []byte：存儲(chǔ)堆棧跟蹤信息的字節(jié)切片，這個(gè)堆棧跟蹤提供了 panic 發(fā)生時(shí)的函數(shù)調(diào)用層次結(jié)構(gòu)和順序，有助于調(diào)試和診斷問題。

方法

panicError 結(jié)構(gòu)體有兩個(gè)方法：

Error() string：實(shí)現(xiàn)了 error 接口的 Error 方法。它將 panicError 結(jié)構(gòu)體的 value 和 stack 字段的格式化成一個(gè)字符串。
Unwrap() error：實(shí)現(xiàn)了 Wrapper 接口的 Unwrap 接包方法，嘗試將 value 字段斷言為 error 類型并返回。如果 value 不是一個(gè) error 類型，它將返回 nil。這個(gè)方法使得 panicError 能夠與 Go 的錯(cuò)誤處理機(jī)制（如 errors.Is 和 errors.As）更好地集成。

初始化函數(shù)

newPanicError(v interface{}) error：這個(gè)函數(shù)用于創(chuàng)建一個(gè)新的 panicError 實(shí)例。它接受從 panic 中恢復(fù)的值作為參數(shù)，然后通過 debug.Stack 獲取堆棧信息，并移除堆棧信息的第一行（如 goroutine 的編號(hào)和狀態(tài)），因?yàn)檫@一行包含的信息可能會(huì)因?yàn)?panic 的恢復(fù)而變得不準(zhǔn)確。最后，返回指向 panicError 實(shí)例的指針。

核心方法解析

Do

func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {
    // 獲取鎖
    g.mu.Lock()
    // 懶初始化 map
    if g.m == nil {
        g.m = make(map[string]*call)
    }
    // 判斷特定 key 的 call 是否正在進(jìn)行調(diào)用
    if c, ok := g.m[key]; ok {
        // 重復(fù)調(diào)用次數(shù)加 1
        c.dups++
        // 解鎖
        g.mu.Unlock()
        // 掛起，等待調(diào)用的完成
        c.wg.Wait()
        // 判斷是否發(fā)生了 panic
        if e, ok := c.err.(*panicError); ok {
                // panic
                panic(e)
        } else if c.err == errGoexit { // 判斷是否發(fā)生了 runtime.Goexit
                // 執(zhí)行 runtime.Goexit，停止當(dāng)前 goroutine 的執(zhí)行，并確保所有 defer 語句的執(zhí)行
                runtime.Goexit()
        }
        // 返回結(jié)果
        return c.val, c.err, true
    }
    // 創(chuàng)建一個(gè)新的調(diào)用
    c := new(call)
    // 等待組加 1
    c.wg.Add(1)
    // key 和 call 映射
    g.m[key] = c
    // 釋放鎖
    g.mu.Unlock()
    // 調(diào)用開始，執(zhí)行所接受的函數(shù) fn
    g.doCall(c, key, fn)
    // 返回結(jié)果
    return c.val, c.err, c.dups > 0
}

Do 方法的執(zhí)行流程如下所示：

1、獲取鎖：通過 g.mu.Lock() 加鎖，確保對(duì)內(nèi)部的 g.m（一個(gè) map，用于跟蹤 key 的調(diào)用狀態(tài)）和 c.dups（對(duì)于該 key 的重復(fù)調(diào)用次數(shù)）的訪問是并發(fā)安全的。

2、初始化 map：如果 g.m == nil，意味著是第一次調(diào)用 Do 方法且沒有調(diào)用過 DoChan 方法，所以初始化 g.m。

3、檢查是否有正在進(jìn)行的調(diào)用：通過 c, ok := g.m[key]; ok 檢查是否有一個(gè)對(duì)于該 key 的調(diào)用正在進(jìn)行，如果 ok 為 true，則說明有一個(gè)對(duì)于該 key 的調(diào)用正在進(jìn)行：

增加重復(fù)調(diào)用次數(shù) c.dups，表示來了一個(gè)新的 goroutine 在等待這個(gè)調(diào)用結(jié)果。
釋放鎖 g.mu.Unlock()，因?yàn)椴辉傩枰薷墓蚕碣Y源。
等待 c.wg.Wait()，直到當(dāng)前的調(diào)用完成。
檢查錯(cuò)誤類型，并按需處理（如果是 panicError 或 errGoexit，則分別觸發(fā) panic 或 Goexit）。
返回當(dāng)前進(jìn)行的調(diào)用的結(jié)果。

4、初始化并執(zhí)行新的調(diào)用：如果沒有一個(gè)對(duì)于該 key 的調(diào)用正在進(jìn)行，則：

創(chuàng)建一個(gè)新的 call 實(shí)例。
c.wg 等待組計(jì)數(shù)加 1，標(biāo)記新操作的開始，后續(xù)有相同調(diào)用的請(qǐng)求將會(huì)等待該操作的完成并共享結(jié)果。
在 g.m 中注冊(cè) key 和新創(chuàng)建的 call 實(shí)例的映射 g.m[key] = c。
釋放鎖。
調(diào)用 g.doCall(c, key, fn) 執(zhí)行實(shí)際的函數(shù)調(diào)用。
返回調(diào)用結(jié)果。

Do 方法的關(guān)鍵在于綜合使用等待組（sync.WaitGroup）、互斥鎖（sync.Mutex）以及一個(gè)映射（map），以確保：

對(duì)于相同的 key，fn 函數(shù)只會(huì)被執(zhí)行一次。這是通過 map 檢查當(dāng)前 key 是否存在對(duì)應(yīng)的 call 實(shí)例來實(shí)現(xiàn)的。如果已存在，意味著函數(shù)調(diào)用正在執(zhí)行或已完成，不需要再次執(zhí)行。
同一時(shí)刻，所有請(qǐng)求同一 key 的調(diào)用都能得到同一個(gè)結(jié)果。通過 map 追蹤特定 key 對(duì)應(yīng)的調(diào)用結(jié)果，確保所有的 goroutine 對(duì)同一 key 發(fā)起 Do 方法調(diào)用都能共享相同的結(jié)果。
正確地處理并發(fā)和同步。通過 sync.Mutex 保護(hù)并發(fā)環(huán)境下 map 的讀寫操作，避免并發(fā)沖突；通過 sync.WaitGroup 等待異步操作完成，保證所有請(qǐng)求都在函數(shù)執(zhí)行完成后才返回結(jié)果。

doCall

doCall 方法負(fù)責(zé)執(zhí)行給定 key 的函數(shù) fn，并處理可能的錯(cuò)誤和同步執(zhí)行結(jié)果，確保所有請(qǐng)求該key的goroutine 得到統(tǒng)一的結(jié)果。該方法的源碼如下所示：

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
    // 定義正常返回標(biāo)志
    normalReturn := false
    // 定義 panic 標(biāo)志
    recovered := false

    // 使用雙重 defer 來區(qū)分 panic 和 runtime.Goexit
    defer func() {
        // fn 函數(shù)里面調(diào)用了 runtime.Goexit 函數(shù)
        if !normalReturn && !recovered {
            // 將 errGoexit 的值賦給 c.err
            c.err = errGoexit
        }
        // 加鎖
        g.mu.Lock()
        // 函數(shù)執(zhí)行結(jié)束時(shí)釋放鎖
        defer g.mu.Unlock()
        // 標(biāo)記 call 的完成
        c.wg.Done()
        // 保險(xiǎn)起見，判斷當(dāng)前 key 對(duì)應(yīng)的 call 是否被覆蓋，沒有被覆蓋就從 map 中移除這個(gè) key 
        if g.m[key] == c {
            delete(g.m, key)
        }
        // 判斷執(zhí)行 fn 的時(shí)候是否發(fā)生 panic
        if e, ok := c.err.(*panicError); ok {
            // 避免等待中的通道永久阻塞，如果發(fā)生了 panic，需要確保這個(gè) panic 不能被捕獲
            if len(c.chans) > 0 {
                // 開一個(gè)新的協(xié)程去 panic，這個(gè) panic 就不會(huì)被捕獲了
                go panic(e)
                // 保持當(dāng)前 goroutine 的存活，這樣等到 panic 之后，關(guān)于當(dāng)前 goroutine 的信息就會(huì)出現(xiàn)在堆棧中
                select {}
            } else {
                // 直接 panic
                panic(e)
            }
        } else if c.err == errGoexit {
            // 如果是 errGoexit，什么都不用做，因?yàn)橹耙呀?jīng)執(zhí)行了 runtime.Goexit
        } else {
            // 向等待中的通道發(fā)送結(jié)果
            for _, ch := range c.chans {
                ch <- Result{c.val, c.err, c.dups > 0}
            }
        }
    }()

    func() {
        defer func() {
            // 如果 fn 沒有正常執(zhí)行完
            if !normalReturn {
                // 獲取從 panic 中恢復(fù)的值
                if r := recover(); r != nil {
                    // 創(chuàng)建一個(gè) `panicError` 實(shí)例并賦值給 c.err
                    c.err = newPanicError(r)
                }
            }
        }()
        // 執(zhí)行函數(shù)調(diào)用
        c.val, c.err = fn()
        // 設(shè)置正常返回標(biāo)志為 true
        normalReturn = true
    }()
    // 如果 fn 沒有正常執(zhí)行完，則發(fā)生了 panic
    if !normalReturn {
        // 設(shè)置 panic 標(biāo)志為 true
        recovered = true
    }
}

代碼剖析：

標(biāo)志位定義：定義 normalReturn 和 recovered 用來區(qū)分 fn 是否正常執(zhí)行完成或者發(fā)生了 panic。
雙重 defer 機(jī)制：目的是為了能夠區(qū)分 fn 函數(shù)的正常執(zhí)行完成、fn 函數(shù)里發(fā)生的 panic 以及 fn 函數(shù)里調(diào)用runtime.Goexit 終止協(xié)程的情況。
- 第一個(gè) defer 用于清理資源和處理結(jié)果。
  - 如果非正常函數(shù)執(zhí)行完成并且沒有發(fā)生 panic，則 fn 里執(zhí)行了 runtime.Goexit 函數(shù)。
  - 加鎖，調(diào)用 c.wg.Done() 以標(biāo)記 call 調(diào)用完成，然后從 g.m 映射中移除當(dāng)前 key。
  - 錯(cuò)誤處理。
    - 如果 fn 函數(shù)中發(fā)生了 panic，先判斷是否有通道正在等待結(jié)果，有的話，新開一個(gè)協(xié)程去 panic，確保 panic 不能被恢復(fù)，這里還用到了 select{} 來阻塞當(dāng)前線程，保證 panic 之后，當(dāng)前 goroutine 的信息會(huì)出現(xiàn)在堆棧中。如果沒有通道正在等待結(jié)果，則直接 panic。
    - 如果是 errGoexit 錯(cuò)誤，說明 fn 函數(shù)中執(zhí)行了 runtime.Goexit，這時(shí)什么都不用做。
  - 結(jié)果同步。如果沒有發(fā)生 error，就向正在等待的通道發(fā)送結(jié)果。
- 第二個(gè) defer 在一個(gè)匿名函數(shù)里，它的目的是執(zhí)行 fn 函數(shù)和捕獲 panic。如果 fn 函數(shù)正常執(zhí)行完成，normalReturn 就會(huì)被設(shè)置為 true；在 defer 里，如果 normalReturn 為 false，則說明可能發(fā)生了 panic，通過 recover() 函數(shù)嘗試恢復(fù) panic 并新建一個(gè) panicError 存儲(chǔ)信息。
recovered 標(biāo)志更新：如果 fn 函數(shù)非正常執(zhí)行成功（normalReturn 為 false），則將 recovered 賦值為 true，表示發(fā)生了 panic。

call 方法的關(guān)鍵在于使用了雙重 defer 機(jī)制，結(jié)合標(biāo)志 normalReturn 和 recovered 來判斷 fn 函數(shù)的狀態(tài)。normalReturn 和 recovered 有三組值：

normalReturn 為 true，recovered 為 false：表明 fn 函數(shù)執(zhí)行成功，后續(xù)執(zhí)行第一個(gè) defer 時(shí)，除了資源清理以外，還會(huì)向等待中的通道發(fā)送調(diào)用完成的結(jié)果。
normalReturn 為 false，recovered 為 true：表明在 fn 函數(shù)里發(fā)生了 panic，并且這個(gè) panic 被成功捕獲并恢復(fù)。后續(xù)執(zhí)行第一個(gè) defer 時(shí)，除了資源清理以外，會(huì)再次觸發(fā) panic。
normalReturn 為 false，recovered 為 false：這種情況說明在 fn 函數(shù)里，調(diào)用了 runtime.Goexit 函數(shù)終止當(dāng)前協(xié)程，不再執(zhí)行后續(xù)的代碼。這意味著 normalReturn = true 和 recovered = true 代碼都不可能被執(zhí)行，因此 normalReturn 和 recovered 的值都為 false。后續(xù)執(zhí)行第一個(gè) defer 時(shí)不會(huì)向等待的通道發(fā)送任何結(jié)果，僅僅是進(jìn)行資源清理。

DoChan

DoChan 方法與 Do 方法類似，但是它返回的是一個(gè)通道，通道在操作完成時(shí)接收到結(jié)果。返回值是通道，意味著我們能以非阻塞的方式等待結(jié)果。該方法的源碼如下所示：

func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result {
    // 創(chuàng)建一個(gè)通道，類型為 Result
    ch := make(chan Result, 1)
    // 加鎖
    g.mu.Lock()
    // 懶初始化 map
    if g.m == nil {
        g.m = make(map[string]*call)
    }
    // 判定該 key 是否有正在進(jìn)行的調(diào)用
    if c, ok := g.m[key]; ok {
        // 重復(fù)調(diào)用次數(shù)加 1
        c.dups++
        // 將新通道添加到通道切片里
        c.chans = append(c.chans, ch)
        // 釋放鎖
        g.mu.Unlock()
        // 返回通道
        return ch
    }
    // 創(chuàng)建一個(gè) call 實(shí)例，并將 ch 通道作為參數(shù)傳遞
    c := &call{chans: []chan<- Result{ch}}
    // 等待組加 1
    c.wg.Add(1)
    // key 和 call 映射
    g.m[key] = c
    // 釋放鎖
    g.mu.Unlock()
    // 異步執(zhí)行調(diào)用
    go g.doCall(c, key, fn)
    // 返回通道
    return ch
}

DoChan 方法的執(zhí)行流程如下所示：

1、創(chuàng)建一個(gè)大小為 1 的緩沖通道。

2、獲取鎖：通過 g.mu.Lock() 加鎖，確保對(duì)內(nèi)部的 g.m（一個(gè) map，用于跟蹤 key 的調(diào)用狀態(tài)）和 c.dups（對(duì)于該 key 的重復(fù)調(diào)用次數(shù)）以及 c.chans（通道切片）的訪問是并發(fā)安全的。

3、初始化 map：如果 g.m == nil，意味著是第一次調(diào)用 Do 方法且沒有調(diào)用過 DoChan 方法，所以初始化 g.m。

4、檢查是否有正在進(jìn)行的調(diào)用：通過 c, ok := g.m[key]; ok 檢查是否有一個(gè)對(duì)于該 key 的調(diào)用正在進(jìn)行，如果 ok 為 true，則說明有一個(gè)對(duì)于該 key 的調(diào)用正在進(jìn)行：

增加重復(fù)調(diào)用次數(shù) c.dups，表示來了一個(gè)新的 goroutine 在等待這個(gè)調(diào)用結(jié)果。
將新創(chuàng)建的通道追加到當(dāng)前 call 的通道切片里。
釋放鎖 g.mu.Unlock()，因?yàn)椴辉傩枰薷墓蚕碣Y源。
返回新創(chuàng)建的通道。

5、初始化并異步執(zhí)行新的調(diào)用：如果沒有一個(gè)對(duì)于該 key 的調(diào)用正在進(jìn)行，則：

創(chuàng)建一個(gè)新的 call 實(shí)例，并關(guān)聯(lián)新創(chuàng)建的通道。
c.wg 等待組計(jì)數(shù)加 1，標(biāo)記新操作的開始，后續(xù)有相同調(diào)用的請(qǐng)求將會(huì)等待該操作的完成并共享結(jié)果。
在 g.m 中注冊(cè) key 和新創(chuàng)建的 call 實(shí)例的映射 g.m[key] = c。
釋放鎖。
異步調(diào)用 g.doCall(c, key, fn) 執(zhí)行實(shí)際的函數(shù)調(diào)用。
返回新創(chuàng)建的通道。

DoChan 與 Do 方法的區(qū)別在于同步共享結(jié)果的方式：

Do 方法：

如果有其他請(qǐng)求正在進(jìn)行（對(duì)同一個(gè)key），它會(huì)使用 sync.WaitGroup 等待這個(gè)請(qǐng)求完成以共享結(jié)果。
如果是針對(duì)給定 key 的新請(qǐng)求，它將直接啟動(dòng) doCall 來執(zhí)行函數(shù)調(diào)用，等待執(zhí)行完成且 call 實(shí)例的更新，然后返回結(jié)果。

DoChan 方法：為每個(gè)調(diào)用創(chuàng)建一個(gè)新的通道，將其加入到對(duì)應(yīng) key 的 call 實(shí)例的通道切片里，然后返回一個(gè)通道。這樣，等 g.doCall 正常異步調(diào)用完成后，會(huì)向各個(gè)通道發(fā)送結(jié)果。

Forget

Forget 方法用于從 g.m 移除特定 key 的調(diào)用。

func (g *Group) Forget(key string) {
    // 加鎖
    g.mu.Lock()
    // 移除特定的 key
    delete(g.m, key)
    // 釋放鎖
    g.mu.Unlock()
}

該方法在刪除特定 key 前執(zhí)行加鎖操作，保護(hù)并發(fā)環(huán)境下 map 的讀寫操作，避免并發(fā)沖突。

小結(jié)

本文對(duì) Go singleflight 的源碼進(jìn)行剖析，該包的主要作用是用于防止重復(fù)的請(qǐng)求，它確保給定的 key，函數(shù)在同一時(shí)間內(nèi)只執(zhí)行一次，多個(gè)請(qǐng)求共享同一結(jié)果。singleflight 能實(shí)現(xiàn)這種效果，關(guān)鍵點(diǎn)在于：

將多個(gè)相同請(qǐng)求合并成一個(gè)請(qǐng)求，確保函數(shù)只執(zhí)行一次：singleflight 為了解決這個(gè)問題，引入了互斥鎖 sync.Mutex 和 map。

互斥鎖用于保護(hù)在并發(fā)環(huán)境下 map 的讀寫操作，避免并發(fā)沖突。

map 則負(fù)責(zé)將每一個(gè)唯一的 key 映射到 call 實(shí)例上，該實(shí)例包含了fn 函數(shù)的返回值和可能的錯(cuò)誤等。

遇到一個(gè)尚未在 map 中記錄的 key 請(qǐng)求時(shí)，創(chuàng)建并執(zhí)行一個(gè)新的 call 實(shí)例。
如果 map 中已存在該 key 對(duì)應(yīng)的 call 實(shí)例，表明有一個(gè)相同的請(qǐng)求正在執(zhí)行或已完成，此時(shí)僅需等待此 call 完成并直接其共享結(jié)果。

結(jié)果共享機(jī)制：singleflight 通過阻塞式和非阻塞式兩種方式，實(shí)現(xiàn)了結(jié)果的共享。

阻塞式機(jī)制：當(dāng)多個(gè)請(qǐng)求通過 Do 方法進(jìn)行相同的調(diào)用時(shí)，它們處于等待狀態(tài)（里面借助了 sync.WaitGroup 來實(shí)現(xiàn)阻塞的效果），直到首個(gè)請(qǐng)求的 fn 函數(shù)的執(zhí)行完畢。此后，等待的請(qǐng)求會(huì)接收到已完成的請(qǐng)求結(jié)果。

非阻塞式機(jī)制：相比于阻塞等待，當(dāng)請(qǐng)求通過 DoChan 方法發(fā)起時(shí)，每個(gè)請(qǐng)求會(huì)立即獲得一個(gè)專屬的通道。這些請(qǐng)求可以繼續(xù)執(zhí)行其他操作，直到它們準(zhǔn)備好從各自的通道接收結(jié)果。在接收結(jié)果時(shí)，如果結(jié)果尚未發(fā)送過來，也會(huì)暫時(shí)處于阻塞狀態(tài)。

除了以上兩個(gè)關(guān)鍵點(diǎn)，還需要考慮錯(cuò)誤的處理，singleflight 通過使用雙重 defer 的機(jī)制，用于辨別函數(shù)正常執(zhí)行完成、函數(shù)里發(fā)生了 panic 以及函數(shù)里調(diào)用了 runtime.Goexit() 函數(shù) 三種情況，每種情況采取不同的處理機(jī)制。

到此這篇關(guān)于源碼剖析Golang中singleflight的應(yīng)用的文章就介紹到這了,更多相關(guān)Go singleflight內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: