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

Question

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

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

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)
}

在很长的计算期内阻止所有客户端（但实际上可能只是向它们提供旧数据）。并且你不能将 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)
}

如何解决以上两个缺点？

PS。我认为这是一个普遍的问题，也是一种模式 - 它有一个特殊的名称吗？

Answer 1

可能有多种解法。一个简单的解决方案是指定一个用于计算的 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)
}

Answer 2

这是我实现的，与 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())
}

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

How to syncronize slow calculation and cache it?

multithreading

synchronization

go