投稿
  1. 首页
  2. python

缓存击穿常用解决方案

code • 2022-4-26 • python • 阅读 40

缓存击穿常用解决方案,第1张

目录
  • 1 抢红包排行查询
  • 2 击穿现象分析
  • 3 击穿解决方案
    • 3.1 定时器
    • 3.2 多级缓存
    • 3.3 分布式锁
    • 3.4 队列术

1 抢红包排行查询

上面案例我们实现了某个用户抢红包的信息查询，接下来我们实现公示抢到红包并且按照红包金额大小排序查询出前
100名用户信息，这块数据并发量将更大，我们需要做缓存处理。

Controller

Service 这里做了缓存 *** 作，缓存1分钟，1分钟过后，会再次查询数据库

测试结果如下：

2 击穿现象分析


我们先来了解下缓存击穿，缓存在某个时间点过期的时候，恰好在这个时间点对这个Key有大量的并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。
上面查询红包排名就存在击穿现象，比如10万用户请求，此时缓存刚好过期，10万用户同时到达了第②个步骤，而且此时Redis中都判断没有数据，那么此时就都查询数据库，给数据库带来巨大的压力，甚至是宕机。

3 击穿解决方案

针对缓存击穿现象，可以有多重解决方案。我们这里给大家讲解一下实用的5种方案。

3.1 定时器

后台定义一个job(定时任务)专门主动更新缓存数据.比如,一个缓存中的数据过期时间是1分钟,那么job每隔25秒刷新数据(将从数据库中查到的数据更新到缓存中)，或者缓存不过期，直接写定时任务定时更新即可。定时器需要择优选择，比如可以用 elastic-job , xxl-job 。
这种方案比较容易理解，但会增加系统复杂度。比较适合那些 key 相对固定,cache 粒度较大的业务，key 比较分散的则不太适合，实现起来也比较复杂。

3.2 多级缓存


采用多级缓存也可以有效防止击穿现象，首先通过程序将缓存存入到Redis缓存，且永不过期，用户查询的时候，先查询Nginx缓存，如果Nginx缓存没有，则查询Redis缓存，并将Redis缓存存入到Nginx一级缓存中，并设置更新时间。这种方案不仅可以提升查询速度，同时又能防止击穿问题，并且提升了程序的抗压能力。

3.3 分布式锁

解决上面超卖问题，我们可以采用分布式锁来控制，分布式锁的原理很简单。
分布式锁主要是实现在分布式场景下保证数据的最终一致性。在单进程的系统中，存在多个线程可以同时改变某个变量（可变共享变量）时，就需要对变量或代码块做同步(lock—synchronized)，使其在修改这种变量时能够线性执行消除并发修改变量。但分布式系统是多部署、多进程的，开发语言提供的并发处理API在此场景下就无能为力了。
目前市面上分布式锁常见的实现方式有三种：

1.基于数据库实现分布式锁；（性能效率比较低，不推荐）
2.基于缓存（Redis等）实现分布式锁； （推荐）
3.基于Zookeeper实现分布式锁；（推荐）

大部分网站使用的分布式锁是基于缓存的，有更好的性能，而缓存一般是以集群方式部署，保证了高可用性。而Redis分布式锁官方推荐使用redisson。

Redission分布式锁说明：
1、redission获取锁释放锁的使用和JDK里面的lock很相似，底层的实现采用了类似lock的处理方式

2、redisson 依赖redis，因此使用redisson 锁需要服务端安装redis，而且redisson 支持单机和集群两种模式下的锁的实现

3、redisson 在多线程或者说是分布式环境下实现机制，其实是通过设置key的方式进行实现，也就是说多个线程为了抢占同一个锁，其实就是争抢设置key。

基于Redisson分布式锁实现

步骤：
1、引入依赖包
2、定义锁的 *** 作方法
a.获取锁的方法
b.解锁
3、创建链接Redis集群的配置文件，在里面做相关配置
4、RedissonClient，实现锁的 *** 作->获取锁和解锁
5、创建工厂对象Redisson的工厂

  1. 引入依赖
<dependency>
  <groupId>org.redissongroupId>
  <artifactId>redisson-spring-boot-starterartifactId>
  <version>3.11.0version>
dependency>
  1. 锁 *** 作方法实现
    要想用到分布式锁，我们就必须要实现获取锁和释放锁，获取锁和释放锁可以编写一个 DistributedLocker 接口，
    代码如下：
public interface DistributedLocker {
  /***
  * lock(), 拿不到lock就不罢休，不然线程就一直block
  * @param lockKey
  * @return
  */
  RLock lock(String lockKey);
  /***
  * timeout为加锁时间，单位为秒
  * @param lockKey
  * @param timeout
  * @return
  */
  RLock lock(String lockKey, long timeout);
  /***
  * timeout为加锁时间，时间单位由unit确定
  * @param lockKey
  * @param unit
  * @param timeout
  * @return
  */
  RLock lock(String lockKey, TimeUnit unit, long timeout);
  /***
  * tryLock()，马上返回，拿到lock就返回true，不然返回false。
  * 带时间限制的tryLock()，拿不到lock，就等一段时间，超时返回false.
* @param lockKey
  * @param unit
  * @param waitTime
  * @param leaseTime 租赁时间，如果超过指定时间还没解锁，就强制解锁
  * @return
  */
  boolean tryLock(String lockKey, TimeUnit unit, long waitTime, long leaseTime);
  /***
  * 解锁
  * @param lockKey
  */
  void unlock(String lockKey);
  /***
  * 解锁
  * @param lock
  */
  void unlock(RLock lock);
}

实现上面接口中对应的锁管理方法,编写一个锁管理类 RedissonDistributedLocker ，代码如下：

@Component
public class RedissonDistributedLocker implements DistributedLocker {
  @Autowired
  private RedissonClient redissonClient;
  /***
  * lock(), 拿不到lock就不罢休，不然线程就一直block
  * @param lockKey
  * @return
  */
  @Override
  public RLock lock(String lockKey) {
    RLock lock = redissonClient.getLock(lockKey);
    lock.lock();
    return lock;
 }
  /***
  * timeout为加锁时间，单位为秒
  * @param lockKey
  * @param timeout
  * @return
  */
  @Override
  public RLock lock(String lockKey, long timeout) {
    RLock lock = redissonClient.getLock(lockKey);
    lock.lock(timeout, TimeUnit.SECONDS);
    return lock;
 }
 /***
  * timeout为加锁时间，时间单位由unit确定
  * @param lockKey
  * @param unit
  * @param timeout
  * @return
  */
  @Override
  public RLock lock(String lockKey, TimeUnit unit, long timeout) {
    RLock lock = redissonClient.getLock(lockKey);
    lock.lock(timeout, unit);
    return lock;
 }
  /***
  * tryLock()，马上返回，拿到lock就返回true，不然返回false。
  * 带时间限制的tryLock()，拿不到lock，就等一段时间，超时返回false.
  * @param lockKey
  * @param unit
  * @param waitTime
  * @param leaseTime
  * @return
  */
  @Override
  public boolean tryLock(String lockKey, TimeUnit unit, long waitTime, long leaseTime) {
    RLock lock = redissonClient.getLock(lockKey);
    try {
      return lock.tryLock(waitTime, leaseTime, unit);
   } catch (InterruptedException e) {
      e.printStackTrace();
   }
    return false;
 }
  /***
  * 解锁
  * @param lockKey
  */
  @Override
  public void unlock(String lockKey) {
    RLock lock = redissonClient.getLock(lockKey);
    lock.unlock();
 }
  /***
  * 解锁
  * @param lock
  */
  @Override
  public void unlock(RLock lock) {
    lock.unlock();
 }
 }
  1. 配置Redis链接
    在resources下新建文件 redisson.yml ，主要用于配置redis集群节点链接配置，代码如下：
clusterServersConfig:
# 连接空闲超时，单位：毫秒 默认10000
idleConnectionTimeout: 10000
pingTimeout: 1000
# 同任何节点建立连接时的等待超时。时间单位是毫秒 默认10000
connectTimeout: 10000
# 等待节点回复命令的时间。该时间从命令发送成功时开始计时。默认3000
timeout: 3000
# 命令失败重试次数
retryAttempts: 3
# 命令重试发送时间间隔，单位：毫秒
retryInterval: 1500
# 重新连接时间间隔，单位：毫秒
reconnectionTimeout: 3000
# 执行失败最大次数
failedAttempts: 3
# 密码
#password: test1234
# 单个连接最大订阅数量
subscriptionsPerConnection: 5
clientName: null
# loadBalancer 负载均衡算法类的选择
loadBalancer: !<org.redisson.connection.balancer.RoundRobinLoadBalancer> {}
#从节点发布和订阅连接的最小空闲连接数
slaveSubscriptionConnectionMinimumIdleSize: 1
#从节点发布和订阅连接池大小 默认值50
slaveSubscriptionConnectionPoolSize: 50
# 从节点最小空闲连接数 默认值32
slaveConnectionMinimumIdleSize: 32
# 从节点连接池大小 默认64
slaveConnectionPoolSize: 64
# 主节点最小空闲连接数 默认32
masterConnectionMinimumIdleSize: 32
# 主节点连接池大小 默认64
masterConnectionPoolSize: 64
# 订阅 *** 作的负载均衡模式
subscriptionMode: SLAVE
# 只在从服务器读取
readMode: SLAVE
# 集群地址
nodeAddresses:
- "redis://192.168.211.141:7001"
- "redis://192.168.211.141:7002"
- "redis://192.168.211.141:7003"
- "redis://192.168.211.141:7004"
- "redis://192.168.211.141:7005"
- "redis://192.168.211.141:7006"
# 对Redis集群节点状态扫描的时间间隔。单位是毫秒。默认1000
scanInterval: 1000
#这个线程池数量被所有RTopic对象监听器，RRemoteService调用者和RExecutorService任务共同共享。默认2
threads: 0
#这个线程池数量是在一个Redisson实例内，被其创建的所有分布式数据类型和服务，以及底层客户端所一同共享的线程池里
保存的线程数量。默认2
nettyThreads: 0
# 编码方式 默认org.redisson.codec.JsonJacksonCodec
codec: !<org.redisson.codec.JsonJacksonCodec> {}
#传输模式
transportMode: NIO
# 分布式锁自动过期时间，防止死锁，默认30000
lockWatchdogTimeout: 30000
# 通过该参数来修改是否按订阅发布消息的接收顺序出来消息，如果选否将对消息实行并行处理，该参数只适用于订阅发布消
息的情况, 默认true
keepPubSubOrder: true
# 用来指定高性能引擎的行为。由于该变量值的选用与使用场景息息相关（NORMAL除外）我们建议对每个参数值都进行尝试。
#
#该参数仅限于Redisson PRO版本。
#performanceMode: HIGHER_THROUGHPUT

4)创建Redisson管理对象
Redisson管理对象有2个，分别为 RedissonClient 和 RedissonConnectionFactory ，我们只用在项目的
RedisConfig 中配置一下这2个对象即可，在 RedisConfig 中添加的代码如下：

/****
* Redisson客户端
* @return
* @throws IOException
*/
@Bean
public RedissonClient redisson() throws IOException {
ClassPathResource resource = new ClassPathResource("redisson.yml");
Config config = Config.fromYAML(resource.getInputStream());
RedissonClient redisson = Redisson.create(config);
return redisson;
}
/***
* Redisson工厂对象
* @param redisson
* @return
*/
@Bean
public RedissonConnectionFactory redissonConnectionFactory(RedissonClient redisson) {
return new RedissonConnectionFactory(redisson);
}
  1. 分布式锁实现
3.4 队列术


上面我们已经使用过队列术了，队列术在面对零点洪峰流量时，是相当有效，可以直接将所有流量存入到队列中，让后台不用同时处理很多请求，而是从队列中逐个消费逐个处理，上图是实现流程，由于前面抢单已经实现过该流程，所以这里不再重复讲解。

基于Nginx缓存队列术

针对一些特定 *** 作，如果请求并发量极高，我们可以采用Nginx自身的队列术，在上一章我们已经学过了Nginx的代理缓存，其中有一个属性叫 proxy_cache_lock ，该属性的意思是：当多个客户端请求一个缓存中不存在的文件（或称之为一个MISS），只有这些请求中的第一个被允许发送至服务器。其他请求在第一个请求得到满意结果之后在缓存中得到文件。如果不启用 proxy_cache_lock ，则所有在缓存中找不到文件的请求都会直接与服务器通信。
proxy_cache_lock 的作用其实和队列术及其类似，只不过发生的地方以及处理的语言不同而已。
我们正好可以使用代理缓存来处理一些查询量大的相同数据，例如查询抢红包Top100就可以用Nginx的代理缓存中

proxy_cache_lock 来实现。

后台代码移除分布式锁：


此时只有第1次从后台获取数据，当然这里会请求3次后才会从缓存拿数据，因为有个属性 proxy_cache_min_use 3属性。

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/langs/726188.html

赞 (0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
code code 管理员组
0 0
上一篇 2022-04-26
下一篇 2022-04-26

发表评论

登录后才能评论

评论列表（0条）

保存