经常用Redis,这些坑你知道吗?

时间:2021-09-18 05:48:21

近些年,Redis凭借在性能、稳定性和高可扩展性上的卓越表现,基本上已经成了互联网行业缓存中间件的标配,甚至很多传统行业也在使用Redis。那么我们在使用Redis等缓存中间件时,要注意哪些问题呢?本文咱们就来聊聊,我们使用缓存中间件过程中曾经遇到的坑!

缓存穿透

先看一个常见的缓存使用方式。请求来了,先查缓存,缓存有值就直接返回;缓存没值,查数据库,然后把数据库的值存到缓存,再返回。

假如缓存没查到某个值,查数据库也没这个值,也就是说要查的值根本不存在,这样就会导致每次对这个值的查询请求都会穿透到数据库。这就是所谓的“缓存穿透”。

如何避免缓存穿透?

如果从数据库中没查到值,可以在缓存中记录一个空值,来避免“缓存穿透”。并且要给这个空值设置一个较短的过期时间。

比如说,我们经常会把用户信息缓存到Redis。如果调用方传了一个不存在的UserID,在缓存中就查不到这个用户信息,然后去DB也查不到。这样就会导致,每次根据这个UserID查用户信息,都会穿透到数据库,给数据库造成了压力。为了避免缓存穿透,当数据库查不到时,我们可以在缓存中记录一条空数据,比如userID做为key,空json做为值,如果程序获得这个空json,就按用户不存在处理。再给这个key设置一个很短的过期时间,比如30秒。

缓存雪崩

我们经常会遇到需要初始化缓存的情况。比如说用户系统重构,表结构发生了变化,缓存信息也要变,上线前需要初始化缓存,将用户信息批量存入缓存。假如我们给这些用户信息设置相同的过期时间,到过期时间点所有用户信息的缓存记录就会同时集中失效,导致大量请求瞬间打到数据库,数据库很可能会被搞挂。这种缓存集中失效,导致大量请求同时穿透到数据库的情况,就是所谓的“雪崩效应”。

所以,当我们向缓存初始化数据时,要保证每个缓存记录过期时间的离散性。可以采用一个较大的固定值加上一个较小的随机值。比如过期时间可以是:10小时 + 0到3600秒的随机值。

缓存并发

当系统并发很高,缓存数据尤其是热点数据过期后,可能会出现多个请求同时访问数据库并设置缓存的情况,不但给数据库带来压力,而且会有缓存频繁更新的问题。

我们可以通过加锁来避免缓存并发问题。如果从缓存查不到数据,对查询数据加分布式锁,然后查数据库并把数据库查询结果放入缓存。其他线程等待锁释放后,直接从缓存取值。

比如,电商系统会缓存商品SKU价格,一些热点商品的并发访问会非常高。当缓存过期失效后,访问请求从缓存查不到记录,此时可以用商品SKU ID为Key加分布式锁,然后从数据库查询价格并把价格放入缓存,最后解锁。解锁后其他请求就可以从缓存直接取值了。从而避免了数据库的压力。

分布式锁

以我们之前做过的5人拼团为例。如果有用户参加团购,我们需要先校验参团人数是否达到了上限5人。如果没达到5人,用户才可以参团。伪代码如下:

//根据拼团ID获取目前参团成员数量 

int numOfMembers = pinTuanService.getNumOfMembersById(pinTuanID); 

if(numOfMembers < 5) { 

  pinTuanService.pintuan();//执行,加入拼团,生单等逻辑 

}  

高并发场景下,上面的代码会有很严重的问题。如果某个团当前的参团人数是4,这时有两个用户同时参团,用户A和用户B的请求同时进入上面的代码块,A和B的请求同时执行到第2行代码,获取的numOfMembers都是4,表达式 numOfMembers < 5 成立,所以两个用户都能执行到第4行代码,就是说A用户和B用户都能成功参加拼团。于是,参团人数就超过了5人的上限。所以我们就需要加锁来避免这个问题。synchronized行吗?不行。因为我们的服务是多节点部署的,所以要加分布式锁。代码如下:

boolean aquired = distributedLock.aquireLock(pinTuanID, 3000); 

if(aquired == true) { 

  try{ 

    //根据拼团ID获取目前参团成员数量 

    int numOfMembers = pinTuanService.getNumOfMembersById(pinTuanID); 

    if(numOfMembers < 5) { 

      pinTuanService.pintuan();//执行,加入拼团,生单等逻辑 

    }  

  } finally { 

    distributedLock.releaseLock(pinTuanID); 

  } 

这样就好多啦!接下来我们看看基于Redis分布式锁的实现,以及特别要注意的问题。一般我们会基于setnx实现Redis分布式锁。setnx命令可以检查key是否存在,如果key不存在,就在Redis中创建一个键值对(操作成功),如果key已经存在就放弃执行(操作失败)。

先看一段基于Springboot实现的加锁和释放锁的代码:

@Component 

public class DistributedLock { 

 

 @Autowired 

 private StringRedisTemplate redisTemplate; 

  

 /** 

 * 加锁 

 * lockKey,redis的key 

 * expireTime,过期时间,单位是毫秒 

 * 注:setIfAbsent方法就使用了redis的setnx 

 */ 

  public boolean aquireLock(String lockKey, long expireTime) { 

   long waitTime = 0; 

   boolean success = redisTemplate.opsForValue().setIfAbsent(lockKey, "distributedLock"

                     expireTime, TimeUnit.MILLISECONDS); 

   if(success == true){ 

      return success;    

   } else { 

     //如果加锁失败,循环重试加锁 

     while(success != true && waitTime < 5000L ) { 

       success = redisTemplate.opsForValue().setIfAbsent(lockKey, "distributedLock"

                       expireTime, TimeUnit.MILLISECONDS); 

       sleep 100毫秒;                 

       waitTime += 100L; 

     } 

   } 

    

   return success; 

 } 

  

 /** 

 * 释放锁 

 * lockKey,redis的key 

 */ 

 public void releaseLock(String lockKey) { 

   redisTemplate.delete(lockKey); 

 }  

  

上面的代码。乍一看,好像没什么问题!加锁失败有循环重试加锁,过期时间设置了,而且也保证了创建Key-Value键值对和设置过期时间的原子性,这样当程序没有正常释放锁时,也能保证过期后锁自动释放(注意:redis较老的版本不支持 setnx 和设置过期时间的原子操作,不过可以利用Lua脚本来保证原子性)。

我们再仔细思考一下,一般场景我们会对Key设置一个很短的过期时间,当一次操作因为网络等原因耗费了较长时间,操作还没完成key就过期失效了。这样会产生什么问题呢?我们还是以拼团为例加以说明,先看看下面这张图:

经常用Redis,这些坑你知道吗?

如上图,用户A和用户B同时参加同一团,团ID为 001,我们以团ID作为分布式锁的Key,"distributedLock" 作为固定的Value,过期时间是5秒。A先获取分布式锁,但是由于网络等原因A的拼团操作在5秒内没完成,这时Key过期并从Redis清除掉,A的分布式锁失效。此时用户B拿到分布式锁,Key也同样是团ID 001。在用户B的拼团逻辑执行完之前,用户A的逻辑先执行完了,紧接着A就把锁给释放了。不过A的锁早已经过期失效了,B持有锁的Key和A又完全一样,所以此时A释放的其实是B的锁。这样一来整个拼团还是有可能会超员。怎么解决呢?

我们可以把分布式锁的Value设成可以区分的值,比如拼团的场景Value可以设置为userID,在释放锁的时候根据key和value来判断当前的锁是不是自己的,只有Redis中userID和自己的userID相同才释放锁。

改进后的代码如下:

@Component 

public class DistributedLock { 

 

 @Autowired 

 private StringRedisTemplate redisTemplate; 

  

 /** 

 * 加锁 

 * lockKey,redis的key 

 * expireTime,过期时间,单位是毫秒 

 * 注:setIfAbsent方法就使用了redis的setnx 

 */ 

  public boolean aquireLock(String lockKey, String userID, long expireTime) { 

   long waitTime = 0; 

   boolean success = redisTemplate.opsForValue().setIfAbsent(lockKey, userID, 

                     expireTime, TimeUnit.MILLISECONDS); 

   if(success == true){ 

      return success;    

   } else { 

     //如果加锁失败,循环重试加锁 

     while(success != true && waitTime < 5000L ) { 

       success = redisTemplate.opsForValue().setIfAbsent(lockKey, userID, 

                       expireTime, TimeUnit.MILLISECONDS); 

       sleep 100毫秒;                 

       waitTime += 100L; 

     } 

   } 

    

   return success; 

 } 

  

 /** 

 * 释放锁 

 * lockKey,redis的key 

 */ 

 public void releaseLock(String lockKey, String userID) { 

   String userIDFromRedis = redisTemplate.get(lockKey); 

   if( userID.equals(userIDFromRedis) ) { 

     redisTemplate.delete(lockKey); 

   } 

 }  

  

还有一种场景需要考虑。当Redis master发生故障,主备切换时往往会造成数据丢失,包括分布式锁的Key-Value 也可能丢失。这样就会导致操作还没执行完,锁就被其他请求拿到了。Redis官方提供了Redlock算法,以及相应的开源实现 Redisson。用到分布式锁的场景,大家可以直接使用 Redisson,非常方便。如果系统对可靠性要求很高,如需用到分布式锁,建议使用 Zookeeper,etcd 等。

原文地址:https://mp.weixin.qq.com/s/MJ_awR4AKirVrt1wCWi18g