缓存穿透
缓存穿透，是指查询一个数据库一定不存在的数据。正常的使用缓存流程大致是，数据查询先进行缓存查询，如果key不存在或者key已经过期，再对数据库进行查询，并把查询到的对象，放进缓存。如果数据库查询对象为空，则不放进缓存。
本篇讨论缓存击穿的其中一个表现：
对于一些设置了过期时间的key，如果这些key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑另外一个问题：缓存被“击穿”的问题。
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操作并不是原子操作（一个是保存数据操作，一个是设置缓存时间操作，是两个请求），在并发环境下可能会有问题，该如何解决呢，欢迎大家留言
