如何同步慢速计算并缓存它？

今天golang学习网给大家带来了《如何同步慢速计算并缓存它？》，其中涉及到的知识点包括等等，无论你是小白还是老手，都适合看一看哦~有好的建议也欢迎大家在评论留言，若是看完有所收获，也希望大家能多多点赞支持呀！一起加油学习~

在 golang 后端中，我想为多个客户端提供一个值，我们称之为分数。分数随时间变化，其计算速度较慢。计算不依赖于先前的结果。当没有客户时我根本不想计算它。因此，计算应该仅根据请求进行。但还有一个事实——分数在 5 秒内不能改变。所以我尝试了不同的方法，但一切都有其缺点：

1. 在没有客户的情况下进行昂贵的计算：

var score interface{}

// run in a separate goroutine
func calculatescoreperiodically() {
    for{
        select{
        case <-time.after(5*time.second):
            score = calculatescoreexpensiveandslow()
        }
    }
}

func servescore(w http.responsewriter, r* http.request) {
    b, _ := json.marshal(score)
    w.write(b)
}

1. 在很长的计算时间内阻止所有客户端（但实际上可能只是向他们提供旧数据）。并且您不能将 if 移出互斥体，因为这样多个客户端可能会同时进入计算块，并且不会在 5 秒间隔内进行计算，而是按顺序进行计算：

var (
    score interface{}
    mutex sync.Mutex
    updatedAt time.Time
)

func getCachedScore() float64 {
    mutex.Lock()
    defer mutex.Unlock()
    currentTime := time.Now()
    if currentTime.Sub(updatedAt) < 5*time.Second {
        return score
    }
    updatedAt = currentTime
    score = calculateScoreExpensiveAndSlow()
    return score
}

func serveScore(w http.ResponseWriter, r* http.Request) {
    b, _ := json.Marshal(getCachedScore())
    w.Write(b)
}

如何解决以上两个缺点？

正确答案

可能有多种解决方案。一个简单的解决方案是指定一个 goroutine 进行计算，您可以通过在通道上发送值来向其发出需要重新计算的信号。发送可能是非阻塞的，因此如果正在进行计算，则不会发生任何事情。

这是一个可重用的缓存实现：

type cache struct {
    mu      sync.rwmutex
    value   interface{}
    updated time.time

    calcch     chan struct{}
    expiration time.duration
}

func newcache(calc func() interface{}, expiration time.duration) *cache {
    c := &cache{
        value:   calc(),
        updated: time.now(),
        calcch:  make(chan struct{}),
    }

    go func() {
        for range c.calcch {
            v := calc()

            c.mu.lock()
            c.value, c.updated = v, time.now()
            c.mu.unlock()
        }
    }()

    return c
}

func (c *cache) get() (value interface{}, updated time.time) {
    c.mu.rlock()
    value, updated = c.value, c.updated
    c.mu.runlock()

    if time.since(updated) > c.expiration {
        // trigger a new calculation (will happen in another goroutine).
        // do non-blocking send, if a calculation is in progress,
        // this will have no effect
        select {
        case c.calcch <- struct{}{}:
        default:
        }
    }

    return
}

func (c *cache) stop() {
    close(c.calcch)
}

注意：cache.stop()是停止后台goroutine。调用cache.stop()后，不得再调用cache.get()。

将其用于您的案例：

func getcachedscore() interface{} {
    // ...
}

var scorecache = newcache(getcachedscore, 5*time.second)

func servescore(w http.responsewriter, r* http.request) {
    score, _ := scorecache.get()
    b, _ := json.marshal(score)
    w.write(b)
}

这是我实现的，与 icza 的答案相关，但还有更多功能：

package common

import (
    "context"
    "sync/atomic"
    "time"
)

type (
    CachedUpdater func() interface{}
    ChanStruct    chan struct{}
)

type Cached struct {
    value        atomic.Value  // holds the cached value's interface{}
    updatedAt    atomic.Value  // holds time.Time, time when last update sequence was started at
    updatePeriod time.Duration // controls minimal anount of time between updates
    needUpdate   ChanStruct
}

//cachedUpdater is a user-provided function with long expensive calculation, that gets current state
func MakeCached(ctx context.Context, updatePeriod time.Duration, cachedUpdater CachedUpdater) *Cached {
    v := &Cached{
        updatePeriod: updatePeriod,
        needUpdate:   make(ChanStruct),
    }
    //v.updatedAt.Store(time.Time{}) // "was never updated", but time should never be nil interface
    v.doUpdate(time.Now(), cachedUpdater)
    go v.updaterController(ctx, cachedUpdater)
    return v
}

//client will get cached value immediately, and optionally may trigger an update, if value is outdated
func (v *Cached) Get() interface{} {
    if v.IsExpired(time.Now()) {
        v.RequestUpdate()
    }
    return v.value.Load()
}

//updateController goroutine can be terminated both by cancelling context, provided to maker, or by closing chan
func (v *Cached) Stop() {
    close(v.needUpdate)
}

//returns true if value is outdated and updater function was likely not called yet
func (v *Cached) IsExpired(currentTime time.Time) bool {
    updatedAt := v.updatedAt.Load().(time.Time)
    return currentTime.Sub(updatedAt) > v.updatePeriod
}

//requests updaterController to perform update, using non-blocking send to unbuffered chan. controller can decide not to update in case if it has recently updated value
func (v *Cached) RequestUpdate() bool {
    select {
    case v.needUpdate <- struct{}{}:
        return true
    default:
        return false
    }
}

func (v *Cached) updaterController(ctx context.Context, cachedUpdater CachedUpdater) {
    for {
        select {
        case <-ctx.Done():
            return
        case _, ok := <-v.needUpdate:
            if !ok {
                return
            }
            currentTime := time.Now()
            if !v.IsExpired(currentTime) {
                continue
            }
            v.doUpdate(currentTime, cachedUpdater)
        }
    }
}

func (v *Cached) doUpdate(currentTime time.Time, cachedUpdater CachedUpdater) {
    v.updatedAt.Store(currentTime)
    v.value.Store(cachedUpdater())
}

今天关于《如何同步慢速计算并缓存它？》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！