基于社区电商的Redis缓存架构-缓存数据库双写、高并发场景下优化

基于社区电商的Redis缓存架构

首先来讲一下 Feed 流的含义：

Feed 流指的是当我们进入 APP 之后，APP 要做一个 Feed 行为，即主动的在 APP 内提供各种各样的内容给我们

在电商 APP 首页，不停在首页向下拉，那么每次拉的时候，APP 就会根据你的喜好、算法来不停地展示新的内容给你看，这就是电商 APP 的 Feed 流了

那么接下来呢就基于首页 Feed 流以及社区电商 APP 中的一些业务场景，来实现一套基于 Redis 的企业级缓存架构，MySQL 为基础，RocketMQ 为辅助

那么在缓存中常见的问题有：

• 热 key 问题
• 大 key 问题
• 缓存雪崩（穿透）
• 数据库和缓存数据一致性问题

在 Redis 生产环境中，也存在一些问题，如下：

• Redis 集群部署，需要进行高并发压测
• 监控 Redis 集群：每个节点数据存储情况、接口 QPS、机器负载情况、缓存命中率
• Redis 节点故障的主从切换、Redis 集群扩容

下边我们来逐个剖析在缓存架构中常见的一些问题

首先，热 key 举个例子就是微博突然某个明星出现新闻，那么会有大量请求去访问这个数据，这个 key 就是热 key

大 key 指的是某个 key 所存储的 value 很大，value 多大 10 mb，那么如果读取这个大 key 过于频繁，就会对网络带宽造成影响，阻塞其他请求

缓存雪崩是因为大量缓存数据同时过期或者 Redis 集群故障，如果因为缓存雪崩导致 Redis 集群都崩掉了，那么此时只有数据库可以访问，我们的系统需要可以识别出来缓存故障，立马对各个接口进行限流、降级错误，来保护数据库，避免数据库崩掉，可以在 jvm 内存缓存中存储少量缓存数据，用于在 Redis 崩了之后提供降级备用数据，那么流程为：限流 --> 降级 --> jvm 缓存数据

读多写少数据缓存

那么我们先来分析一下在社区电商中，用户去分享一个内容时，如何操作缓存：

1. 用户分享内容
2. 加锁：针对用户 id 上分布式锁，避免同一用户重复请求，导致数据重复灌入
3. 将用户分享内容写入数据库
4. 将用户的个人信息在缓存写一份（用户信息在注册后一般不会变化，读多写少，因此用户信息非常适合放入缓存中），这样在后续高并发访问用户的数据时，就可以在缓存中进行查询，根据用户前缀 + 用户 id 作为 key，并设置缓存过期时间，过期时间可以设置为 2 天加上随机几小时（添加随机几小时的原因是避免同一时间大量缓存同时过期）
5. 释放锁

缓存自动延期以及缓存穿透

那么上边我们已经对用户的个人信息进行了缓存，那么某些热门的用户是经常被很多人看到的，而有些冷门用户的内容没多少人看，因此将用户信息的缓存过期时间设置为 2天+随机几小时 ，因此我们针对热门的用户信息数据，要做一个缓存自动延期，因此只要访问用户数据，那么就对该缓存数据进行一个延期，流程如下：

1. 获取用户个人信息
2. 根据 用户前缀 + 用户id 作为 key 去缓存中查询用户信息

public UserInfo getUserInfo(Integer userId) {// 读缓存String userInfoJson = redisCache.get("user_info_lock:" + userId);if (!StringUtils.isEmpty(userInfoJson)) {// 取到的是空缓存 避免一致访问数据库不存在的数据导致缓存穿透if ("{}".equals(userInfoJson)) {// 延期缓存redisCache.expire("user_info_lock:" + userId, expireSecond);return null;} else {// 延期缓存redisCache.expire("user_info_lock:" + userId, expireSecond);UserInfo userInfo = JSON.parseObject(productStr, UserInfo.class);}}// 如果缓存中没有取到数据，则在数据库中查，并放入缓存lock("user_lock_prefix" + userId); // 上分布式锁 伪代码try {// 读数据库UserInfo userInfo = userInfoService.get(userId);if(userInfo != null) {// 如果用户信息不为空，就将用户数据写入缓存redisCache.set("user_info_lock:" + userId, JSON.toJSONString(userInfo), expireSecond);} else {// 如果数据库为空，写入一个空字符串即可redisCache.set("user_info_lock:" + userId, "{}", expireSecond);}} finally {unlock("user_lock_lock:" + userId); // 解锁}
}

缓存+数据库双写不一致

在上边存储用户个人信息时，使用的是 缓存+数据库双写，这样可能造成数据不一致性，如下：

有两个线程并发，一个读线程，一个写线程，假设执行流程如下，会造成双写不一致

• 当读线程去缓存中读取数据，此时缓存中数据正好过期，那么该线程就去读数据库中的数据
• 此时写线程开始执行，修改用户信息，并且写入数据库，再写入缓存
• 此时读线程再接着执行，将之前在数据库中读取的旧数据写入缓存，覆盖了写线程更新后的数据

造成这种情况的原因是，在读的时候，加的是读锁，key 为 user_info_lock:，在写的时候，加的是写锁，key 为 user_update_lock:，那么读写就可以并发

想要解决的话，可以让读和写操作加同一把锁，让读写串行化，就可以了，如下：

让读和写都加同一把锁：user_update_lock

• 针对先读后写的情况，就不会出现双写不一致了，读的时候先加上锁user_update_lock，此时缓存假设正好过期，去数据库中读取数据，此时写线程开始执行，阻塞等待锁user_update_lock，此时读线程再去数据库读取数据放入缓存，结束后释放锁，那么写线程再拿到锁操作数据库，再将数据写入缓存，那么缓存中的数据是新数据
• 针对先写后读的情况，也不会出现双写不一致，在写的时候，加上锁 user_update_lock，那么在并发读的时候，先获取缓存内容，如果获取不到，尝试去 DB 中获取，此时就会阻塞等待锁 user_update_lock，在写线程写完之后更新了缓存，释放锁，此时读线程就拿到了锁，此时在去数据库中查数据之前再加一个 double check（双端检锁） 的操作，也就是再尝试去缓存中取一次数据，如果取到了就返回；如果没有取到，就去数据库中查询

那么完整的写和读操作流程如下图：

高并发场景下优化

在上边我们已经使用读写加同一把锁来实现缓存数据库双写一致了

但是还存在一种极端情况：某一个用户的信息并不在缓存中，但是突然火了，大量用户来访问，发现缓存中没有，那么大量用户线程就阻塞在了获取锁的这一步操作上，导致大量线程串行化的来获取锁，然后再到缓存中获取数据，下一个线程再获取锁取数据

这种情况的解决方案就是给获取锁加一个超时时间，如果在 200ms 内没有拿到锁，就算获取锁失败，这样大量用户线程获取锁失败，就会从串行再转为并发从缓存中取数据了，避免大量线程阻塞获取锁

完整流程图如下，粉色部分为优化：

代码如下：

private UserInfo getUserInfo(Long userId) {String userLockKey = "user_update_lock:" + userId;boolean lock = false;try {// 尝试加锁，并设置超时时间为 200 mslock = redisLock.tryLock("user_update_lock:", 200);} catch(InterruptedException e) {UserInfo user = getFromCache(userId);if(user != null) {return user;}log.error(e.getMessage(), e);throw new BaseBizException("查询失败");}// 如果加锁超时，就再次去缓存中查询if (!lock) {UserInfo user = getFromCache(userId);if(user != null) {return user;}// 缓存数据为空，查询用户信息失败，因为用户没有拿到锁，因此也无法取 DB 中查询throw new BaseBizException("查询失败");}// 双端检锁，如果拿到锁，再去缓存中查询try {UserInfo user = getFromCache(userId);if(user != null) {return user;}String userInfoKey = "user_info:" + userId;// 数据库中查询user = userService.getById(userId);if (Objects.isNull(user)) {redisCache.set(userInfoKey, "{}", RandomUtil.genRandomInt(30, 100));return null;}// 缓存时间设置为 2 天 + 随机几小时redisCache.set(userInfoKey, JSON.toJSONString(user), 2 * 24 * 60 * 60RandomUtil.genRandomInt(0, 10) * 60 * 60);return user;} finally {redisLock.unlock(userLockKey);}