Golang 并发非阻塞缓存

《The Go Programming Language》笔记

import "sync"

type Func func(key string) (interface{},error)

type result struct {
    value interface{}
    err   error
}

type entry struct {
    res result
    ready chan struct{} //closed when res is ready
}

type MemoryCache struct {
    f     Func
    mu    sync.Mutex
    cache map[string]*entry
}

func New(f Func) *MemoryCache {
    return &MemoryCache{f: f,cache: make(map[string]*entry)}
}

//获取互斥锁来保护共享变量cache map，查询map中是否存在指定key的value
//如果不存在那么分配空间插入新值，释放互斥锁。
//如果存在且其值没有写入完即其他goroutine在调用f这个慢函数，
//goroutine必须等待值ready之后才能读到key的value
//如果没有key对应的value，需要向map中插入一个没有ready的entry，
//当前正在调用的goroutine就需要负责调用慢函数更新result以及向其他goroutine
//广播（关闭ready）result已经可读了
func (mc *MemoryCache) Get(key string) (interface{},error) {
    mc.mu.Lock()
    e := mc.cache[key]
    if e == nil { //e为空
        //This is the first request for this key
        e = &entry{ready:make(chan struct{})}
        mc.cache[key] = e
        mc.mu.Unlock()

        e.res.value,e.res.err = mc.f(key) //执行耗时函数
        close(e.ready)//broadcast ready condition
    }else {
        mc.mu.Unlock()
        <-e.ready //阻塞，直到ready关闭
    }
    return e.res.value,e.res.err
}

type Func func(key string) (interface{},error)

type result struct {
    value interface{}
    err   error
}

type entry struct {
    res   result
    ready chan struct{} //closed when res is ready
}

type request struct {
    key      string
    response chan<- result
}

type MemoryCache struct {
    requests chan request
}

func New(f Func) *MemoryCache {
    mc := &MemoryCache{requests: make(chan request)}
    go mc.server(f)
    return mc
}

func (mc *MemoryCache) Get(key string) (interface{},error) {
    response := make(chan result)
    mc.requests <- request{key,response}
    res := <- response
    return res.value,res.err
}

func (mc *MemoryCache) server(f Func) {
    cache := make(map[string]*entry)
    for req := range mc.requests {
        e := cache[req.key]
        if e == nil {
            e = &entry{ready: make(chan struct{})}
            cache[req.key] = e
            go e.call(f,req.key)
        }
        go e.deliver(req.response)

    }
}

func (mc *MemoryCache) Close()  {
    close(mc.requests)
}

func (e *entry) call(f Func,key string) {
    e.res.value,e.res.err = f(key)
    close(e.ready)
}

func (e *entry) deliver(response chan<- result) {
    <-e.ready         //wait for the ready condition
    response <- e.res //Send the result to the client
}