当前位置:联升科技 > 技术资讯 > 开发技术 >

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

2020-12-24    作者:二马读书    来源:杨建荣的学习笔记    阅读:
近些年,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就过期失效了。这样会产生什么问题呢?我们还是以拼团为例加以说明,先看看下面这张图:

如上图,用户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 等。


相关文章

我们很乐意倾听您的声音!
即刻与我们取得联络
成为日后肩并肩合作的伙伴。

行业资讯

联系我们

13387904606

地址:新余市仙女湖区仙女湖大道万商红A2栋

手机:13755589003
QQ:122322500
微信号:13755589003

江西新余网站设计_小程序制作_OA系统开发_企业ERP管理系统_app开发-新余联升网络科技有限公司 赣ICP备19013599号-1   赣公网安备 36050202000267号   

微信二维码