缓存穿透问题分析压测 原

时间:2022-06-19
本文章向大家介绍缓存穿透问题分析压测 原 ，主要内容包括其使用实例、应用技巧、基本知识点总结和需要注意事项，具有一定的参考价值，需要的朋友可以参考一下。

缓存穿透

    缓存穿透，是指查询一个数据库一定不存在的数据。正常的使用缓存流程大致是，数据查询先进行缓存查询，如果key不存在或者key已经过期，再对数据库进行查询，并把查询到的对象，放进缓存。如果数据库查询对象为空，则不放进缓存。

    本篇讨论缓存击穿的其中一个表现：

    对于一些设置了过期时间的key，如果这些key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑另外一个问题：缓存被“击穿”的问题。

  • 概念：缓存在某个时间点过期的时候，恰好在这个时间点对这个Key有大量的并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。
  • 如何解决：使用mutex。简单地来说，就是在缓存失效的时候（判断拿出来的值为空），不是立即去load db，而是先使用缓存工具的某些带成功操作返回值的操作（比如Redis的SETNX或者Memcache的ADD）去set一个mutex key，当操作返回成功时，再进行load db的操作并回设缓存；否则，就重试整个get缓存的方法。类似下面的代码： 
public String get(key) {
    String value = redis.get(key);
    if (value == null) { //代表缓存值过期
        //设置3min的超时，防止del操作失败的时候，下次缓存过期一直不能load db
        if (redis.setnx(key_mutex, 1, 3 * 60) == 1) {  //代表设置成功
            value = db.get(key);
                    redis.set(key, value, expire_secs);
                    redis.del(key_mutex);
            } else {  //这个时候代表同时候的其他线程已经load db并回设到缓存了，这时候重试获取缓存值即可
                    sleep(50);
                    get(key);  //重试
            }
        } else {
            return value;      
        }
}

接下来，进行并发压力测试和优化：

首先是不使用setNX进行并发压力测试

代码如下：

package cn.chinotan.controller;

import lombok.extern.java.Log;
import org.apache.catalina.servlet4preview.http.HttpServletRequest;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

/**
 * @program: test
 * @description: redis测试
 * @author: xingcheng
 * @create: 2019-03-09 16:26
 **/
@RestController
@RequestMapping("/redis")
@Log
public class RedisController {

    @Autowired
    StringRedisTemplate redisTemplate;

    public static final String KEY = "chinotan:redis:pass";

    public static final String VALUE = "redis-pass-value";

    /**
     * 模拟耗时操作 3秒
     */
    public static final Long TIME_CONSUMING = 3 * 1000L;

    /**
     * VALUE缓存时间 5秒
     */
    public static final Long VALUE_TIME = 5 * 1000L;

    @GetMapping(value = "/pass")
    public Object hello(HttpServletRequest request) throws Exception {
        long cacheStart = System.currentTimeMillis();
        String value = redisTemplate.opsForValue().get(KEY);
        long cacheEnd = System.currentTimeMillis();
        if (StringUtils.isBlank(value)) {
            // 模拟耗时操作，从数据库获取
            long start = System.currentTimeMillis();
            TimeUnit.MILLISECONDS.sleep(TIME_CONSUMING);
            redisTemplate.opsForValue().set(KEY, VALUE, VALUE_TIME, TimeUnit.MILLISECONDS);
            long end = System.currentTimeMillis();
            log.info("从数据库中获取耗时: " + (end - start) + "ms");
            return VALUE;
        } else {
            log.info("从缓存中获取耗时：" + (cacheEnd - cacheStart) + "ms");
            return value;
        }
    }

}

很简单的一个get请求，先从缓存中获取数据，如果数据不存在，则从数据库获取，这里用

TimeUnit.MILLISECONDS.sleep(TIME_CONSUMING);

来模拟一个复杂的从数据库获取数据的操作，耗时设定为3秒钟

本次测试采用的是springBoot2.0以上进行部署，jmeter进行压力并发测试

在压力测试之前，进行springboot自带的tomcat并发数和连接数调整以及redis连接池的调整

redis的连接池调整如下：

spring:
  redis:
    database: 0
    host: 127.0.0.1
    jedis:
      pool:
        #最大连接数据库连接数
        max-active: 5000
        #最大等待连接中的数量
        max-idle: 5000
        #最大建立连接等待时间。如果超过此时间将接到异常。设为-1表示无限制。
        max-wait: -1
        #最小等待连接中的数量,设 0 为没有限制
        min-idle: 10
#    lettuce:
#      pool:
#        max-active: 5000
#        max-idle: 5000
#        max-wait: -1
#        min-idle: 10
#      shutdown-timeout: 5000ms
    password:
    port: 6379
    timeout: 5000

tomcat的调整如下：

server:
  port: 11111
  tomcat: 
      uri-encoding: UTF-8
      max-threads: 500
      max-connections: 10000

这样redis和tomcat可以支持大并发请求

设置完成后查看设置是否生效：

redis连接池不生效举例如下：

必须和配置项相同才正确

之后进行压测准备：下载jmeter，之后步骤如下

启动后控制台打印如下：

可以看到大量并发过来后，会有多次的查看操作，并没有走到缓存，缓存命中率低，缓存的意义就少很多

下面进行优化：

package cn.chinotan.controller;

import lombok.extern.java.Log;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import redis.clients.jedis.JedisCommands;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @program: test
 * @description: redis测试
 * @author: xingcheng
 * @create: 2019-03-09 16:26
 **/
@RestController
@RequestMapping("/redis")
@Log
public class RedisController {

    @Autowired
    StringRedisTemplate redisTemplate;

    public static final String KEY = "chinotan:redis:pass";

    public static final String NX_KEY = "chinotan:redis:nx";

    public static final String VALUE = "redis-pass-value";

    /**
     * 间隔时间 3秒
     */
    public static final Long NX_SLEEP_TIME = 50L;

    /**
     * 模拟耗时操作 3秒
     */
    public static final Long TIME_CONSUMING = 1 * 1000L;

    /**
     * VALUE缓存时间 5秒
     */
    public static final Long VALUE_TIME = 5 * 1000L;

    /**
     * 锁缓存时间 5分钟
     */
    public static final Long NX_TIME = 5 * 60L;

    @GetMapping(value = "/pass")
    public Object hello() throws Exception {
        long cacheStart = System.currentTimeMillis();
        String value = redisTemplate.opsForValue().get(KEY);
        long cacheEnd = System.currentTimeMillis();
        if (StringUtils.isBlank(value)) {
            long start = System.currentTimeMillis();
            if (setNX(NX_KEY, NX_KEY)) {
                // 模拟耗时操作，从数据库获取
                TimeUnit.MILLISECONDS.sleep(TIME_CONSUMING);
                redisTemplate.opsForValue().set(KEY, VALUE, VALUE_TIME, TimeUnit.MILLISECONDS);
                long end = System.currentTimeMillis();
                redisTemplate.delete(NX_KEY);
                log.info("从数据库中获取耗时: " + (end - start) + "ms");
                return VALUE;
            } else {
                TimeUnit.MILLISECONDS.sleep(NX_SLEEP_TIME);
                log.info("缓存穿透递归");
                return hello();
            }

        } else {
            log.info("从缓存中获取耗时：" + (cacheEnd - cacheStart) + "ms");
            return value;
        }
    }

    private boolean setNX(String key, String value) {
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(key, value);
        redisTemplate.expire(key, NX_TIME, TimeUnit.SECONDS);
        return aBoolean;
    }

}

通过进行setNX命令操作，这个命令在缓存存在时不会进行覆盖更新写入操作，并返回false，缓存不存在才会进行写入并返回true，通常会被用来分布式锁的设计实现

进行优化后，大量的并发请求不会打到数据库上，而是每隔50ms进行递归重试，这样只有一个请求会请求数据库，其他请求只能从缓存中取数，大大增加了缓存的命中率

下面是压测结果：

可以看到从数据库取数的操作日志只有一条，从而避免了缓存击穿的一个表现问题

下一步优化方向：

RedisTemplate提供的setNX操作并不是原子操作（一个是保存数据操作，一个是设置缓存时间操作，是两个请求），在并发环境下可能会有问题，该如何解决呢，欢迎大家留言

(adsbygoogle = window.adsbygoogle || []).push({});

随机文章
本站知识点必读
最近更新