源码剖析Golang中singleflight的应用

singleflight 是 Go 语言标准库 sync 包中的一个组件，它的作用是避免重复的 expensive 函数调用。当多个 goroutine 同时请求相同的数据时，singleflight 会确保只有一个 goroutine 执行调用，而其他的 goroutine 会等待并重用结果。这可以减少对资源的重复请求，提高程序的性能和效率。

singleflight 的工作原理是通过一个 map 来存储正在进行的调用和对应的结果。当一个新的请求到来时，singleflight 会检查是否已经有 goroutine 正在处理相同的请求。如果有，它会将请求加入到等待队列中；如果没有，它会创建一个新的 goroutine 来执行请求，并将其加入到正在执行的调用 map 中。

源码剖析

singleflight 的核心是一个 Group 结构体，它包含两个 map：m 和 wg。m 用于存储调用的键（key）和对应的结果，而 wg 用于存储等待的 goroutine。

type Group struct {
    m   map[interface{}]*call
    wg sync.WaitGroup
    mu  sync.Mutex
}

• m 是一个 map，键是一个空接口类型的数据，可以存储任何类型的数据。值是指向 call 结构体的指针，call 结构体包含了 goroutine 的结果和等待该结果的 goroutine 列表。

• wg 是一个 WaitGroup，用于等待所有的 goroutine 完成。

• mu 是一个互斥锁，用于同步对 m 和 wg 的访问。

下面是 singleflight 的几个关键方法：

1. Do：这是 singleflight 的主要方法，用于执行或重用现有的调用。

func (g *Group) Do(key interface{}, fn func() (interface{}, error)) (interface{}, error) {
    g.mu.Lock()
    if g.m == nil {
        g.m = make(map[interface{}]*call)
    }
    c, exists := g.m[key]
    g.mu.Unlock()

    if exists {
        // 如果调用已经存在，等待结果
        g.wg.Add(1)
        c.wg.Wait()
        return c.result, c.err
    }

    // 如果调用不存在，创建一个新的调用
    c = &call{
        wg: new(sync.WaitGroup),
    }
    g.wg.Add(1) // 增加等待的 goroutine 数量
    g.mu.Lock()
    g.m[key] = c
    g.mu.Unlock()

    // 执行调用
    c.result, c.err = fn()
    c.wg.Done() // 完成调用

    // 通知所有等待的 goroutine
    g.mu.Lock()
    for _, w := range g.m[key].wg {
        w.Done()
    }
    g.mu.Unlock()

    return c.result, c.err
}

在 Do 方法中，首先会检查是否已经有一个 goroutine 正在处理相同的请求。如果有，它会等待该 goroutine 完成并返回结果。如果没有，它会创建一个新的 goroutine 来执行调用，并存储结果供其他 goroutine 使用。

2. Forget：这个方法用于从 singleflight 中移除一个调用，通常是在调用不再需要时使用。

func (g *Group) Forget(key interface{}) {
    g.mu.Lock()
    delete(g.m, key)
    g.mu.Unlock()
}

3. Wait：这个方法用于等待所有正在进行的调用完成。

func (g *Group) Wait() {
    g.wg.Wait()
}

应用示例

singleflight 可以用于任何需要避免重复工作的场景。例如，一个缓存库可能会使用 singleflight 来确保对同一个键的多个并发请求只执行一次获取操作。

var group = &sync.Group{}

func fetch(key string) ([]byte, error) {
    // 假设这是一个昂贵的操作，我们不想重复执行
    result, err := someExpensiveOperation(key)
    return result, err
}

func concurrentFetch(key string, group *sync.Group) {
    group.Go(func() error {
        // 使用 singleflight 来避免重复的 expensive 调用
        _, err := group.Do(key, func() (interface{}, error) {
            return fetch(key)
        })
        return err
    })
}

func main() {
    keys := []string{"key1", "key2", "key3"}
    group = &sync.Group{}

    // 并发地获取所有的键
    for _, key := range keys {
        concurrentFetch(key, group)
    }

    // 等待所有的并发调用完成
    if err := group.Wait(); err != nil {
        log.Fatal(err)
    }
}

在这个示例中，我们使用 singleflight 来确保对于每个 key 的多个并发请求，fetch 函数只被执行一次。其他的 goroutine 会等待并重用第一次调用的结果。

总结

singleflight 是一个简单但强大的工具，它可以帮助你避免不必要的重复工作，提高程序的性能。通过使用 singleflight，你可以确保昂贵的操作只被执行一次，即使面对大量的并发请求。在使用 singleflight 时，需要注意正确地管理 goroutine 的生命周期，确保所有的资源都被正确地释放。