当前位置： > > > > 如何同步慢速计算并缓存它？
如何同步慢速计算并缓存它？
来源：stackoverflow
2024-04-19 13:18:33
0浏览
收藏
今天米云给大家带来了《如何同步慢速计算并缓存它？》，其中涉及到的知识点包括等等，无论你是小白还是老手，都适合看一看哦~有好的建议也欢迎大家在评论留言，若是看完有所收获，也希望大家能多多点赞支持呀！一起加油学习~
问题内容
在 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。我认为这是一个普遍问题，也是一个模式 – 它有一个特殊的名称吗？
正确答案
可能有多种解决方案。一个简单的解决方案是指定一个 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())
}
今天关于《如何同步慢速计算并缓存它？》的内容介绍就到此结束，如果有什么疑问或者建议，可以在米云公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！
