前言

前面的一篇文章 Go singleflight：防緩存擊穿利器 詳細(xì)介紹 singleflight 包的使用，展示如何利用它來(lái)避免緩存擊穿。而本篇文章，我們來(lái)剖析 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)行讀寫操作，所以需要通過(guò)互斥鎖來(lái)確保對(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 通過(guò)維護(hù) m 字段來(lái)跟蹤每個(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)用過(guò)程中可能發(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)試和診斷問(wèn)題。

方法

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ù)，然后通過(guò) 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 語(yǔ)句的執(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)用開(kāi)始，執(zhí)行所接受的函數(shù) fn
    g.doCall(c, key, fn)
    // 返回結(jié)果
    return c.val, c.err, c.dups > 0
}

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

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

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

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

• 增加重復(fù)調(diào)用次數(shù) c.dups，表示來(lái)了一個(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)用：如果沒(méi)有一個(gè)對(duì)于該 key 的調(diào)用正在進(jìn)行，則：

• 創(chuàng)建一個(gè)新的 call 實(shí)例。
• c.wg 等待組計(jì)數(shù)加 1，標(biāo)記新操作的開(kāi)始，后續(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í)行一次。這是通過(guò) map 檢查當(dāng)前 key 是否存在對(duì)應(yīng)的 call 實(shí)例來(lái)實(shí)現(xiàn)的。如果已存在，意味著函數(shù)調(diào)用正在執(zhí)行或已完成，不需要再次執(zhí)行。
• 同一時(shí)刻，所有請(qǐng)求同一 key 的調(diào)用都能得到同一個(gè)結(jié)果。通過(guò) map 追蹤特定 key 對(duì)應(yīng)的調(diào)用結(jié)果，確保所有的 goroutine 對(duì)同一 key 發(fā)起 Do 方法調(diào)用都能共享相同的結(jié)果。
• 正確地處理并發(fā)和同步。通過(guò) sync.Mutex 保護(hù)并發(fā)環(huán)境下 map 的讀寫操作，避免并發(fā)沖突；通過(guò) 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 來(lái)區(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)起見(jiàn)，判斷當(dāng)前 key 對(duì)應(yīng)的 call 是否被覆蓋，沒(méi)有被覆蓋就從 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 {
                // 開(kāi)一個(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 沒(méi)有正常執(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 沒(méi)有正常執(zhí)行完，則發(fā)生了 panic
    if !normalReturn {
        // 設(shè)置 panic 標(biāo)志為 true
        recovered = true
    }
}

代碼剖析：

• 標(biāo)志位定義：定義 normalReturn 和 recovered 用來(lái)區(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í)行完成并且沒(méi)有發(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é)果，有的話，新開(kāi)一個(gè)協(xié)程去 panic，確保 panic 不能被恢復(fù)，這里還用到了 select{} 來(lái)阻塞當(dāng)前線程，保證 panic 之后，當(dāng)前 goroutine 的信息會(huì)出現(xiàn)在堆棧中。如果沒(méi)有通道正在等待結(jié)果，則直接 panic。
      • 如果是 errGoexit 錯(cuò)誤，說(shuō)明 fn 函數(shù)中執(zhí)行了 runtime.Goexit，這時(shí)什么都不用做。
    • 結(jié)果同步。如果沒(méi)有發(fā)生 error，就向正在等待的通道發(fā)送結(jié)果。

  • 第二個(gè) defer 在一個(gè)匿名函數(shù)里，它的目的是執(zhí)行 fn 函數(shù)和捕獲 panic。如果 fn 函數(shù)正常執(zhí)行完成，normalReturn 就會(huì)被設(shè)置為 true；在 defer 里，如果 normalReturn 為 false，則說(shuō)明可能發(fā)生了 panic，通過(guò) 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 來(lái)判斷 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：這種情況說(shuō)明在 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、獲取鎖：通過(guò) 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（通道切片） 的訪問(wèn)是并發(fā)安全的。

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

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

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

5、初始化并異步執(zhí)行新的調(diào)用：如果沒(méi)有一個(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)記新操作的開(kāi)始，后續(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 來(lái)執(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è)問(wèn)題，引入了互斥鎖 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 通過(guò)阻塞式和非阻塞式兩種方式，實(shí)現(xiàn)了結(jié)果的共享。

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

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

除了以上兩個(gè)關(guān)鍵點(diǎn)，還需要考慮錯(cuò)誤的處理，singleflight 通過(guò)使用雙重 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)文章希望大家以后多多支持腳本之家！

最新評(píng)論