美文网首页
经常用Redis,这些坑你知道吗?

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

作者: jeanron | 来源:发表于2020-12-23 22:19 被阅读0次

    文章转载自二马读书,作者二马读书

    作者简介:曾任职于阿里巴巴,每日优鲜等互联网公司,任技术总监,15年电商互联网经历。

    近些年,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实现的加锁和释放锁的代码:

    @Componentpublic 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就过期失效了。这样会产生什么问题呢?我们还是以拼团为例加以说明,先看看下面这张图:

    如上图,用户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相同才释放锁。

    改进后的代码如下:

    @Componentpublic 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 等。

    OK,就分享到这。如果感觉本文对您有帮助,有劳点下在看,分享知识是美德哦🙏

    相关文章

      网友评论

          本文标题:经常用Redis,这些坑你知道吗?

          本文链接:https://www.haomeiwen.com/subject/jcxpnktx.html