rockscache源码分析：如何解决缓存db的最终一致性
背景
在使用的缓存策略为：
- 写：先更新db，再删除缓存
- 读：先读缓存，如果有直接返回。否则读db，然后回写缓存
当如下场景发生时，会直接导致缓存数据与数据库数据不一致：
常见的解决方案有：
- 设定稍短的过期时间兜底：在这个过期时间之内，会不一致。缺点是较短的过期时间意味着数据库的负载会更高
- 延时双删：先删除一次缓存，延时几百毫秒再删除一次。这种做法只能够进一步降低不一致的概率，但无法保证
本文介绍https://github.com/dtm-labs/rockscache（版本v0.1.1）中提供的方案，能较好地解决这个问题，让上图中的第5步写不进缓存，来达到最终一致性的效果。同时介绍如何处理缓存穿透，击穿，雪崩
核心是两个方法：
读数据时，调fetch方法：
/*
key: 缓存key
expire：缓存过期时间
fn：缓存不存在时，调fn方法拿到原始数据
*/
func (c *Client) Fetch(key string, expire time.Duration, fn func() (string, error)) (string, error) {
写数据时，调TagAsDeleted方法，将缓存置为标记删除
// key：要标记删除哪个key
func (c *Client) TagAsDeleted2(ctx context.Context, key string) error
fetch流程
缓存的数据是一个hash结构，包含3个字段：
- value: 数据本身
- lockUntil：数据锁定到期时间，当某个进程查询缓存无数据，那么先锁定缓存一小段时间，然后查询DB，然后更新缓存
- lockOwner：数据锁定者uuid
提供了弱一致性fetch和强一致性fetch两种选择
-
弱一致性：数据被标记删除后，到查db然后重新生成缓存这段期间，可以使用老数据
-
强一致性：数据被标记删除后，不能再用redis中的老数据，而需要同步等待从db读新数据
- 比弱一致性的一致性高一点
弱一致性fetch整体流程如下：
func (c *Client) weakFetch(ctx context.Context, key string, expire time.Duration, fn func() (string, error)) (string, error) {
// 生成代表自己的uuid
owner := shortuuid.New()
/**
r[0]: value
r[1]: 如果自己上锁成功，返回locked
*/
r, err := c.luaGet(ctx, key, owner)
// 锁被别人占据，且valueu不存在，休眠100ms再次执行luaGet
for err == nil && r[0] == nil && r[1].(string) != locked {
time.Sleep(c.Options.LockSleep)
r, err = c.luaGet(ctx, key, owner)
}
if err != nil {
return "", err
}
// value存在，自己没加上锁，说明value是有效的，返回
if r[1] != locked {
return r[0].(string), nil
}
// 下面都是自己加锁成功了，需要自己查数据，回填缓存
// 如果此时redis没有数据，需要同步
if r[0] == nil {
return c.fetchNew(ctx, key, expire, owner, fn)
}
// 否则可以异步
go withRecover(func() {
_, _ = c.fetchNew(ctx, key, expire, owner, fn)
})
return r[0].(string), nil
}
-
先查redis，如果没上锁或锁已过期，就锁定，并返回value和自己是否加锁成功
- 怎么判断锁已过期：lockUntil字段存在，且 lockUntil < now
- 怎么判断没上锁：lockUntil和value字段都不存在
- 怎么锁定：设置lockOwner为自己的uuid，设置lockUntil为now()+10s
用luaGet脚本完成上述操作，保证原子性
注意：
1）lua脚本里redis.call('HGET', key, field)，当field不存在时返回false。注意区分field不存在和field存在，但值为空
2） ~= 运算符表示 不等于
func (c *Client) luaGet(ctx context.Context, key string, owner string) ([]interface{}, error) {
res, err := callLua(ctx, c.rdb, ` -- luaGet
local v = redis.call('HGET', KEYS[1], 'value')
local lu = redis.call('HGET', KEYS[1], 'lockUntil')
// 锁已过期，或 没上锁
if lu ~= false and tonumber(lu) < tonumber(ARGV[1]) or lu == false and v == false then
redis.call('HSET', KEYS[1], 'lockUntil', ARGV[2])
redis.call('HSET', KEYS[1], 'lockOwner', ARGV[3])
return { v, 'LOCKED' }
end
return {v, lu}
`, []string{key}, []interface{}{now(), now() + int64(c.Options.LockExpire/time.Second), owner})
debugf("luaGet return: %v, %v", res, err)
if err != nil {
return nil, err
}
return res.([]interface{}), nil
}
-
如果数据为空，且被别人锁定，就休眠100ms再次执行步骤1
-
自己没上锁成功，但是有数据了，返回数据
- 缓存命中的情况下，会在这一步返回
-
自己上锁成功，且缓存为空，同步执行取数据流程：
-
什么情况会出现？缓存为空时
-
调
fn获取db的数据 -
获取成功，将数据写入redis：
- 如果当前没有持有锁（lockOwner不等于自己），返回。这一步校验是保证最终一致性的关键
- 否则设置值，过期时间，删除lockUntil，lockOwner字段
- （防缓存穿透）如果fn没返回err，但返回了空，且
client.EmptyExpire > 0，设置一份空缓存
-
获取失败，解锁：删除lockOwner字段，lockUntil字段设为0
- 此时会导致锁过期，别人就可以加锁了
-
-
自己上锁成功，且缓存不为空，返回数据，并异步执行取数据流程
- 什么情况会出现？在标记删除后
fetchNew方法：
func (c *Client) fetchNew(ctx context.Context, key string, expire time.Duration, owner string, fn func() (string, error)) (string, error) {
// 调fn回源查db
result, err := fn()
if err != nil {
_ = c.UnlockForUpdate(ctx, key, owner)
return "", err
}
if result == "" {
// 如果db里就没有，根据EmptyExpire决定是否要设置空缓存
if c.Options.EmptyExpire == 0 {
err = c.rdb.Del(ctx, key).Err()
return "", err
}
expire = c.Options.EmptyExpire
}
// 回填到db
err = c.luaSet(ctx, key, result, int(expire/time.Second), owner)
return result, err
}
func (c *Client) luaSet(ctx context.Context, key string, value string, expire int, owner string) error {
_, err := callLua(ctx, c.rdb, `-- luaSet
local o = redis.call('HGET', KEYS[1], 'lockOwner')
// 乐观锁：校验lockOwner有没有变化，如果校验失败直接返回
if o ~= ARGV[2] then
return
end
redis.call('HSET', KEYS[1], 'value', ARGV[1])
redis.call('HDEL', KEYS[1], 'lockUntil')
redis.call('HDEL', KEYS[1], 'lockOwner')
redis.call('EXPIRE', KEYS[1], ARGV[3])
`, []string{key}, []interface{}{value, owner, expire})
return err
}
强一致性：
原理与弱一致性的流程差别不大，仅做了很小的改变，就是当db更新后，redis里还有旧版本的数据时，不再返回该旧版本的数据，而是同步等待“取数据”的最新结果
标记删除
当更新db后，调用标记删除方法：
func (c *Client) TagAsDeleted2(ctx context.Context, key string) error {
if c.Options.DisableCacheDelete {
return nil
}
luaFn := func(con redisConn) error {
_, err := callLua(ctx, con, ` -- delete
redis.call('HSET', KEYS[1], 'lockUntil', 0)
redis.call('HDEL', KEYS[1], 'lockOwner')
redis.call('EXPIRE', KEYS[1], ARGV[1])
`, []string{key}, []interface{}{int64(c.Options.Delay / time.Second)})
return err
}
return luaFn(c.rdb)
}
-
将lockUntil设为0，删除lockOwner字段
- 这样以后别的请求进来，就会加锁成功，然后返回老数据，并异步查db然后更新redis
-
设置key 10s后过期
解决不一致
当遇到一开始提到的，缓存，db不一致场景时：
这种方案会让上面第5步写不进去：
第2步写入v2时，就将缓存标记删除：将缓存解锁，但不删除value
第3步就会上锁成功
第4步写入缓存，然后清空lockOwner字段
第5步发现lockOwner不是自己，就无法写入
为啥这种方案能防止不一致？本质上来说，是加了个乐观锁：
只有当加锁, 回查db, 回写redis这期间db没有更新时，回写redis才能成功
一旦这期间mysql数据变动了，就会触发标记删除，将锁清空。后面再来的读请求会加锁成功，然后再触发下一次查db回写缓存的操作，以此保证最终一致性
防缓存击穿
- 如果缓存中的数据不存在，那么会锁定缓存中的这条数据，避免了多个请求打到后端数据库。其他线程会每隔100ms重试请求redis
- 在弱一致性fetch下：热点数据被标记删除时，旧版本的数据还在缓存中，会被立即返回，无需等待
防缓存穿透
client有个EmptyExpire参数，如果>0，在回源查db为空时，往redis设置一份空缓存
func (c *Client) fetchNew(ctx context.Context, key string, expire time.Duration, owner string, fn func() (string, error)) (string, error) {
// 回源查db
result, err := fn()
if err != nil {
_ = c.UnlockForUpdate(ctx, key, owner)
return "", err
}
if result == "" {
// EmptyExpire为0，直接返回了
if c.Options.EmptyExpire == 0 {
err = c.rdb.Del(ctx, key).Err()
return "", err
}
// EmptyExpire不为0，下面设备空缓存
expire = c.Options.EmptyExpire
}
err = c.luaSet(ctx, key, result, int(expire/time.Second), owner)
return result, err
}
防缓存雪崩
client有RandomExpireAdjustment参数，设置过期时间时会减去这个比例的随机值：
例如本来设置600s过期，RandomExpireAdjustment = 0.1，那么最终过期时间为：
600s - 0～60s（600s * 0.1），也就是540s~600s之间的随机值
原文地址：https://blog.csdn.net/qq_39383767/article/details/142973541
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！