背景：

在当下的互联网行业的实际开发中，在大流量、分布式架构，高并发等等复杂的情况下，维护运行一套：性能不错的、一致性不错的缓存系统，是一些初入职场的后端同学经常要面对的现状。

本文旨在分析这其中出现的常见问题，帮助梳理理解，以及提供一个切实可行的解决方案供大家参考。

在大流量的背景了，为了追求性能，业界最通用的还是Write-Behind缓存模式，本文的提供的解决方案也是参照此模式的思路。

实战情况：

目前这套缓存系统，正运行在的产品，拥有近千万的月活，每日同时最大在线人数CCU超过10万。

下面是去掉敏感数据之后，某一个区的流量概况：

QPS Total：约20～30k

同时段Redis集群流量：700k左右

同时段Mysql集群流量：5k

系统架构（技术栈）

本缓存系统会使用以下三种基础组件

1. Cache：Redis（其他KV缓存数据库都可）
2. DB：Mysql（TiDB等持久化数据库都可）
3. MQ：Kafka（RMQ等消息队列都可）

流程

读过程：

此图为Redis=Nil，即没有读到缓存的流程，后文将简称为 Load流程

写过程：

1. 先写缓存，如果失败则返回错误
2. 写Kafka
3. 消费Kafka，更新Mysql

NOTE1：本文的重点不是讨论如何确保异步的成功，这里可以简单地认为一定会成功，且并发安全

（PS：下篇文章就详解 && 提供一个异步落库实战模板，您的点赞收藏关注（催更）是我更新的动力

NOTE2：这里和普通的模式不一样，在写完Mysql之后不会再改动缓存，认为此时缓存已经是最新的数据。

目的

维护缓存和持久化数据库的最终一致，且缓存数据的即时性是很高的。即如果缓存中有数据，则一定是最新的数据，不会落后于持久化数据库的数据。

持久化数据库中的数据库是相对落后的，这取决于写完缓存后到真正写到Mysql的时间长短。

问题与解决

讲到这里有一些聪明的同学应该已经能看出来问题了，对的，我们的缓存系统听上去 并发不安全。

有哪些不安全的情况呢？不妨枚举并发的情况（假设Redis为Nil，此时读操作将进行Load流程）

1. 读请求和读请求并发，因为都是同一份数据，相互覆盖也不会有问题
   • ✅因此可以并发。
2. 写请求和写请求的并发，除非是类似 incr 的操作，否则一般不能相互覆盖，因此不能并发。
   • ✅一般来说 分布式互斥锁（用Redis实现即可）是简单有效的办法。当一个写操作正在进行时，其他写操作无法进行。
3. 读请求和写请求的并发，如果并发了，读请求的Set操作将进行错误的覆盖（除非是类型SetNx的操作

如图：蓝色写请求在更新完Redis数据之后，被红色读请求读到的落后数据覆盖

解决方案：

鲁迅曾经说过：没有什么并发问题是加锁解决不了的，如果一把不够，那就再来一把！

请看时序图：

图中 Lock0 为第一把锁，即写请求之间的互斥锁

Lock1 为第二把锁，即读写请求之间的互斥锁，在加上锁之后可以看出来，蓝色框和红色框只能按顺序执行，不可交替执行。

枚举此时的情况，只有两种：

1. 时间顺序为先读请求， 再进行写请求：那么读请求 读到了一个相对落后的数据，且Load，但是很快蓝色的写请求会更新Redis为最新的数据，用户再请求一次就能获取最新的数据。

2. 时间顺序为先写请求，再进行读请求：那么写请求 会先写最新的数据到Redis，读请求会读到此数据并返回。

所以，在一般的并发情况下，最终一致性是没有问题的。

NOTE：一个容易踩坑的并发场景需要注意：

当我们进行Redis Key数据迁移的时候，即涉及到两个Redis Key的双写双读时，因为服务机器数量一般都大于1，在滚动发布时，需要考虑一下新版本和老版本之间的并发情况。

比如:

1. 用户首先发送读请求，打到新代码上，会进行Load将V1数据写进Redis Key1。
2. 再进行一次写请求，打到旧代码上，此时只会更新Redis Key2。
3. 此时出现Redis Key 1 和 2 的数据不一致。如果新代码直接使用Redis Key1 则等于用户的写请求被淹没了。

解决方法有三种：

1. 停服发布。
2. 代码进行双读的时候，比较两个数据，取最新的版本。
3. 发布一个中间态版本：在旧代码的基础上，进行双写Redis Key1和Redis Key2。即便写请求打到运行旧代码的机器执行，也会对两个key都进行更新

性能分析：

1. 正常情况下，用户的写请求很少会并发，因为业务场景大部分情况下更新一份数据之后会有动画播放、弹窗提示，且前端也不允许用户对同一个按钮连续点击发送大量相同的无意义请求。
2. 将缓存时间设置得长一点，即有缓存的情况下是不需要加锁的，真正需要加锁的也就是用户刚来的第一次读请求
3. 实际生产环境，这套系统基本能够Cover住 最终一致性 的绝大部分业务需求

建议（闲聊）

根据实际经验，Redis集群能扛住很大的流量，但是公司也是有预算的。不是所有问题都能靠简单的堆机器来解决。

Redis的注意点可以从两个方向来聊聊：

Redis Ops

最值得关注的一点是：设计好用的、用户纬度的 Hash Redis Key

可能的现状：

屎山🪨：一个项目可能由若干个人在若干时间段维护，很容易出现一个功能新建一个Redis Key实现

日积月累，一个请求出现十几次甚至几十次Redis请求， 假设流量峰值时接口10W的QPS，那Redis集群要承载的请求量就会被放大到100W+

解决思路（todo：贴个Hash Key读取的代码模板）

实际上很多redis key的Value就是简单的字符串，完全可以用一个或几个Hash做到。如果使用上了完美的Hash Key，那么理想状态下：

用户请求到达，只需要做少数几次的Redis.HGetAll，就能拿到所有的用户信息，之后也仅需少数几次的HMSet请求即可完成更新数据。

这样一个读请求的Redis Ops一般都会<5次，一个复杂的写请求也不会超过10次。

缺点：

本质上是用空间换次数，占用的Redis内存可能会稍大一些，因此用户的一部分数据的更新，会更新整个key的过期时间。

另外就是避免将不同过期时间的内容放在一起，比如一个一小时过期的数据 和 一个7天过期的数据，这种情况最好是拆开两个Hash Key，一个为小时级过期，一个为周级过期。

Redis内存

注意以下几点：

1. 首先是 Redis Key Name，特别是用户维度的Key，能短则短（注意可读性），一个key只节省了几个字节，但是几百万用户每人都有一个Key，就能节约很大的空间。
2. 同上，一些Hash Key的Member Name，或者用Json序列化的结构体也可以如此优化
3. （谨慎使用）用 PB 代替 Json 来序列化数据，然后进行存储
   • 缺点是牺牲可读性，线上查问题的时候，无法直接读到value，。
4. 静态的资源、模板不要放在Redis中存储，比如 A 给 B 发了一条打招呼的信息，假设模板为：“Hi， 您的好友{friend_name}给您打了一个招呼。”
   • 那么在Redis中只需要存储这个模板ID，和A的Name即可。B在读取信息的时候可以从本地缓存中根据模板ID获取具体内容再进行替换。

伪代码

分布式锁

type RedisLock struct {
    ReleaseScript *redis.Script
    // redis.NewScript(`if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`)

    Val           string

    Cli         *redis.Client
    Key         string
    Timeout     time.Duration
    IsWait        bool
    WaitTimeout time.Duration
}

func(r *RedisLock) Lock() (success bool, err error) {
    var timer *time.Timer
    randomVal = GetRandomSeq()
    success = false
    err = nil

    deadline := time.Now().Add(waitTimeout)
    for {
       succ, err := cache.SetNX(context.Background(), r.LockKey, randomVal, timeout).Result()
       if err != nil {
          // 补充日志
          return
       }

       if success {
          return
       }

       if !l.IsWait {
          success = false
          return
       }

       // 重试
       if timer == nil {
          timer = time.NewTimer(RETRY_INTERVAL * time.Millisecond)
       } else {
          timer.Reset(RETRY_INTERVAL * time.Millisecond)
       }
       
       // 等待
       select {
       case <-timer.C:
       }
       if !time.Now().Before(deadline) {
          break
       }
    }

    if !success {
       // 补充日志
    }
    return
}

func(r *RedisLock) Unlock() error {
    res, err := redisLockScript.Run(context.Background(), redisClient, []string{r.Key}, r.Val).Result()
    if err == redis.Nil {
       return RedisLockUnHoldErr
    }
    if err != nil {
       // 补充日志
       return err
    }

    code, ok := res.(int64)
    if !ok || code != 1 {
       return RedisLockUnHoldErr
    }

    return nil
}

加锁的写请求

func ChangeInfo(ctx context.Context, uid int64) error {
    // lock
    userLock := cache.LockUser(ctx, uid)
    if success, err := userLock.Lock(); err != nil || !success {
       err = errors.New("lock failed")
       return err
    }
    defer userLock.Unlock()

    info, err := dal.GetInfo(ctx, uid)
    ...
    // change info
    ...

    // lock 2
    dataLock := cache.LockData(ctx, uid)
    if success, err := dataLock.Lock(); err != nil || !success {
       err = errors.New("lock failed")
       return err
    }
    defer dataLock.Unlock()


    err := cache.UpdateInfo(ctx, uid)
    if err != nil {
       return err
    }
    ...
}

正常加锁的读请求

func GetInfo(ctx context.Context, uid int64) (*model.Info, error) {
    info, err := cache.GetInfo(ctx, uid)
    if err != nil && err != redis.Nil {
       // log
       return nil, err
    }

    if info != nil {
       return info, nil
    }

    // get redis nil
    // lock
    dataLock := cache.LockData(ctx, uid)
    if success, err := dataLock.Lock(); err != nil {
       err = errors.New("lock failed")
       return err
    }
    defer dataLock.Unlock()
    if !lockSuccess {
       // 等 10ms 然后获取
       time.Sleep(10 * time.Millisecond)

       // 能获取到则返回
       info, _ := cache.GetInfo(ctx, uid)
       if info != nil {
          return info, nil
       }

       // 获取不到,则报错
       return nil, nil, nil, nil, response.ErrRedisLockFailed
    }

    // lock 成功，再读一遍, 确保没有
    info, _ := cache.GetInfo(ctx, uid)
    if info != nil {
       return info, nil
    }

    // 在锁有效的时间内需要完成set cache, 否则放弃写入
    newCtx, cancel := context.WithTimeout(context.Background(), time.Duration(5) * time.Second)
    defer cancel()

    info, err = dao.GetInfoFromMysql(ctx, uid)
    if err != nil {
       return nil, err
    }

    isOK, err := cache.SetInfo(newCtx, info)
    if err != nil {
       return nil, err
    }

    return info, err
}

总结

优点：因为加锁，虽然仍然是最终一致性，但只要不出现Redis的集群挂掉，几乎可以认为是强一致性的，Mysql虽然落后于Redis，但如果有Redis信息，就不会读到Mysql。

缺点：因为加锁，实际上将并发换成了部分顺序执行，牺牲掉了部分性能。但在读多写少的场景下，这点牺牲甚至可以忽略。

最终还是要依照业务场景，进行取舍。