我可以在 Go 中使用特定值锁定吗？

Question

在回答另一个问题时，我编写了一个sync.Map用于缓存 API 请求的小结构。

type PostManager struct {

    sync.Map

}

func (pc PostManager) Fetch(id int) Post {

    post, ok := pc.Load(id)

    if ok {

        fmt.Printf("Using cached post %v\n", id)

        return post.(Post)

    }

    fmt.Printf("Fetching post %v\n", id)

    post = pc.fetchPost(id)

    pc.Store(id, post)

    return post.(Post)

}

不幸的是，如果两个 goroutine 同时获取同一个未缓存的 Post，它们都会发出请求。

var postManager PostManager

wg.Add(3)

var firstPost Post

var secondPost Post

var secondPostAgain Post

go func() {

    // Fetches and caches 1

    firstPost = postManager.Fetch(1)

    defer wg.Done()

}()

go func() {

    // Fetches and caches 2

    secondPost = postManager.Fetch(2)

    defer wg.Done()

}()

go func() {

    // Also fetches and caches 2

    secondPostAgain = postManager.Fetch(2)

    defer wg.Done()

}()

wg.Wait()

我需要确保当同时提取相同 ID 时，只允许一个实际发出请求。另一个必须等待并将使用缓存的 Post。但也不要锁定不同 ID 的提取。

在上面的示例中，我希望只有一个调用pc.fetchPost(1)andpc.fetchPost(2)并且它们应该是同时的。

Answer 1

golang.org/x/sync/singleflight包正是为此目的而编写的。请注意，所有缓存访问都应该发生在传递给 Do 的回调函数中。在您在评论中链接到的代码中，您在外部进行查找；这有点违背了目的。此外，您必须使用指向 singleflight.Group 的指针。这就是你的数据竞赛的来源，请去 vet 指出：./foo.go:41:10：fetchPost 按值传递锁：命令行参数。PostManager 包含 golang.org/x/sync/singleflight.Group 包含 sync.Mutex这是我的写法（操场上的完整示例：https: //play.golang.org/p/2hE721uA88S）：import (    "strconv"    "sync"    "golang.org/x/sync/singleflight")type PostManager struct {    sf    *singleflight.Group    cache *sync.Map}func (pc *PostManager) Fetch(id int) Post {    x, _, _ := pc.sf.Do(strconv.Itoa(id), func() (interface{}, error) {        post, ok := pc.cache.Load(id)        if !ok {            post = pc.fetchPost(id)            pc.cache.Store(id, post)        }        return post, nil    })    return x.(Post)}

Answer 2

如果提取已经在进行中，看起来可以使用第二张地图等待。type PostManager struct {&nbsp; &nbsp; sync.Map&nbsp; &nbsp; q sync.Map}func (pc *PostManager) Fetch(id int) Post {&nbsp; &nbsp; post, ok := pc.Load(id)&nbsp; &nbsp; if ok {&nbsp; &nbsp; &nbsp; &nbsp; fmt.Printf("Using cached post %v\n", id)&nbsp; &nbsp; &nbsp; &nbsp; return post.(Post)&nbsp; &nbsp; }&nbsp; &nbsp; fmt.Printf("Fetching post %v\n", id)&nbsp; &nbsp; if c, loaded := pc.q.LoadOrStore(id, make(chan struct{})); !loaded {&nbsp; &nbsp; &nbsp; &nbsp; post = pc.fetchPost(id)&nbsp; &nbsp; &nbsp; &nbsp; pc.Store(id, post)&nbsp; &nbsp; &nbsp; &nbsp; close(c.(chan struct{}))&nbsp; &nbsp; } else {&nbsp; &nbsp; &nbsp; &nbsp; <-c.(chan struct{})&nbsp; &nbsp; &nbsp; &nbsp; post,_ = pc.Load(id)&nbsp; &nbsp; }&nbsp; &nbsp; return post.(Post)}或者，更复杂一点，使用相同的地图；-)func (pc *PostManager) Fetch(id int) Post {&nbsp; &nbsp; p, ok := pc.Load(id)&nbsp; &nbsp; if !ok {&nbsp; &nbsp; &nbsp; &nbsp; fmt.Printf("Fetching post %v\n", id)&nbsp; &nbsp; &nbsp; &nbsp; if p, ok = pc.LoadOrStore(id, make(chan struct{})); !ok {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; fetched = pc.fetchPost(id)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; pc.Store(id, fetched)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; close(p.(chan struct{}))&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return fetched&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }&nbsp; &nbsp; if cached, ok := p.(Post); ok {&nbsp; &nbsp; &nbsp; &nbsp; fmt.Printf("Using cached post %v\n", id)&nbsp; &nbsp; &nbsp; &nbsp; return cached&nbsp; &nbsp; }&nbsp; &nbsp; fmt.Printf("Wating for cached post %v\n", id)&nbsp; &nbsp; <-p.(chan struct{})&nbsp; &nbsp; return pc.Fetch(id)}

Answer 3

您可以使用两张地图来做到这一点，一张保留缓存的值，另一张保留正在获取的值。您还需要将锁保持更长时间，这样就不需要保持同步地图，常规地图就可以了。像这样的东西应该可以工作（未经测试）：type PostManager struct {&nbsp; &nbsp; sync.Mutex&nbsp; &nbsp; cached map[int]Post&nbsp; &nbsp; loading map[int]chan struct{}}您需要处理以下加载失败的情况：// Need to pass pointer pcfunc (pc *PostManager) Fetch(id int) Post {&nbsp; &nbsp; pc.Lock()&nbsp; &nbsp; post, ok:=pc.cached[id]&nbsp; &nbsp; if ok {&nbsp; &nbsp; &nbsp; &nbsp; pc.Unlock()&nbsp; &nbsp; &nbsp; &nbsp; return post&nbsp; &nbsp; }&nbsp; &nbsp; // See if it is being loaded&nbsp; &nbsp; loading, ok:=pc.loading[id]&nbsp; &nbsp; if ok {&nbsp; &nbsp; &nbsp; &nbsp;// Wait for the loading to complete&nbsp; &nbsp; &nbsp; &nbsp;pc.Unlock()&nbsp; &nbsp; &nbsp; &nbsp;<-loading&nbsp; &nbsp; &nbsp; &nbsp;// Reload&nbsp; &nbsp; &nbsp; &nbsp;pc.Lock()&nbsp; &nbsp; &nbsp; &nbsp;post,ok:=pc.cached[id]&nbsp; &nbsp; &nbsp; &nbsp;// Maybe you need to handle the case where loading failed?&nbsp; &nbsp; &nbsp; &nbsp;pc.Unlock()&nbsp; &nbsp; &nbsp; &nbsp;return post&nbsp; &nbsp; }&nbsp; &nbsp; // load it&nbsp; &nbsp; loading=make(chan struct{})&nbsp; &nbsp; pc.loading[id]=loading&nbsp; &nbsp; pc.Unlock()&nbsp; &nbsp; post = pc.fetchPost(id)&nbsp; &nbsp; pc.Lock()&nbsp; &nbsp; pc.cached[id]=post&nbsp; &nbsp; delete(pc.loading,id)&nbsp; &nbsp; pc.Unlock()&nbsp; &nbsp; close(loading)&nbsp; &nbsp; return post}