缓存击穿
缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力。
我们知道，使用缓存，如果获取不到，才会去数据库里获取。但是如果是热点 key，访问量非常的大，数据库在重建缓存的时候，会出现很多线程同时重建的情况。因为高并发导致的大量热点的 key 在重建还没完成的时候，不断被重建缓存的过程，由于大量线程都去做重建缓存工作，导致服务器拖慢的情况。
解决方案
对于缓存击穿问题，首先要找到问题的关键，它带来的危害就是大量的数据库访问，我们就是要避免这种情况的发生。
其实对于缓存没有但数据库可以查到的数据只需要一此查询和再次缓存就可以解决后续查找访问数据库，这种大量访问同时查询数据库我们需要只让一次访问到数据库，或者说让它们以此进入，这就需要借助锁的机制。
对于锁的选择我们需要考虑生产环境，因为一般只有大型项目才需要缓存击穿问题，而且这种项目大多是分布式的，所以下面介绍的两个锁都是基于分布式的。
下面的两个锁都是基于redis实现的，一个是我们自己依靠redis实现的锁，一个是redis官方提供的分布式锁redLock。
自己实现的redis锁
基本的思路就是，因为每一个需要查询的数据的id是唯一的，通过redis的两个命令setnx和setex设置以id为key的string数据，通过返回值判断是否设置成功，如果设置成功代表拿到了锁，如果没有则代表有其他的进程拿到此锁。就需要阻塞住，等待并再次获取锁。
锁的接口
public interface Lock {
void lock();
void lockInterruptibly() throws InterruptedException;
boolean tryLock();
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
void unlock() throws Exception;
Condition newCondition();
}
锁的实现类
@Component
public class RedisLock implements Lock {
@Autowired
private JedisPool jedisPool;
private static final String key="lock";
private ThreadLocal<String> threadLocal=new ThreadLocal<>();
@Override
public void lock() {
boolean b = tryLock(); //尝试加锁
if(b){
return;
}
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
lock();
}
@Override
public void lockInterruptibly() throws InterruptedException {
}
@Override
public boolean tryLock() {
SetParams setParams=new SetParams();
setParams.ex(1);
setParams.nx();
String s = UUID.randomUUID().toString();
Jedis resource = jedisPool.getResource();
String lock = resource.set(key, s,setParams);
resource.close();
if("OK".equals(lock)){
threadLocal.set(s);
return true;
}
return false;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return false;
}
@Override
public void unlock() throws Exception{
String script="if redis.call(\"get\",KEYS[1])==ARGV[1] then\n" +
" return redis.call(\"del\",KEYS[1])\n" +
"else\n" +
" return 0\n" +
"end";
Jedis resource = jedisPool.getResource();
Object eval = resource.eval(script, Arrays.asList(key), Arrays.asList(threadLocal.get()));
if(Integer.valueOf(eval.toString())==0){
resource.close();
throw new Exception("解锁失败");
}
resource.close();
}
@Override
public Condition newCondition() {
return null;
}
}
测试
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes=SpringConfig.class)
public class Dome {
private int count=100;
// Lock lock=new ReentrantLock();
@Autowired
@Qualifier("redisLock")
private Lock lock;
@Test
public void testRedis(){
Jedis jedis=new Jedis("192.168.0.104",6379);
// SetParams setParams=new SetParams();
// setParams.ex(60);
// setParams.nx();
String s = UUID.randomUUID().toString();
// String lock = jedis.set("lock", s,setParams);
jedis.setnx("lock", s);
jedis.expire("lock",60);
System.out.println(lock);
}
@Test
public void Test() throws InterruptedException {
TicketsRunBle ticketsRunBle=new TicketsRunBle();
Thread thread1=new Thread(ticketsRunBle,"窗口1");
Thread thread2=new Thread(ticketsRunBle,"窗口2");
Thread thread3=new Thread(ticketsRunBle,"窗口3");
Thread thread4=new Thread(ticketsRunBle,"窗口4");
Thread thread5=new Thread(ticketsRunBle,"窗口5");
thread1.start();
thread2.start();
thread3.start();
thread4.start();
thread5.start();
Thread.currentThread().join();
}
public class TicketsRunBle implements Runnable{
@Override
public void run() {
while (count>0){
lock.lock(); //10miao
try {
if(count>0){
System.out.println(Thread.currentThread().getName()+"售出第"+count--+"张票");
} //11
}catch (Exception e){
e.printStackTrace();
}finally {
try {
lock.unlock();
} catch (Exception e) {
e.printStackTrace();
}
}
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
这里还需要注意的是因为我们获取锁和解锁都不只是单个命令，但这些过程必须保证原子性，否则就有可能造成锁失效，所以对于获取锁的两个命令我采用了setParams实现的，setParams是jedis3.0之后提供的对象，它可以保证我们使用多个命令也可以当成单个命令来看。
而对于解锁的操作则涉及到了一次判断，因为解锁的时候会出现两种情况
这个时候因为过期时间的原因，可能解锁的操作发生时已经解锁了，如果我们此时将redis中的值删除就可能出现问题，对于其他进程来说就会出现不应该解锁的情况下解锁了，锁自然就失效了，我们当然不想这种情况发生，所以我们在手动删除锁的值的时候会判断一下当前的锁是否是自身拥有的。这就需要我们每一个进程生成一个uuid，并且存在以id为key的数据种，在删除值得时候做一下判断，如果当前锁已经不是自己的了就说明锁已经过期了，也就不需要再删除值了。
需要注意的是锁的过期时间是必须存在的，因为如果进程加锁了，但突然宕掉了，又没有过期时间，就会将这个数据阻塞住，带来的影响可想而知。
当然这个解锁的过程也必须是原子性的，但是这显然包含了一个判断，一条或者多条命令也是解决不了问题的，所以我们需要借助lua脚本实现解锁的操作，因为发送的lua脚本是可以当作一条命令的。
RedLock锁
这里使用的是和spring整合的redis
引入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/org.redisson/redisson -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.7.4</version>
</dependency>
redis配置
@Configuration
public class RedisConfig {
@Bean
public JedisPoolConfig jedisPoolConfig(){
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
jedisPoolConfig.setMaxTotal(1500); //最大连接数
jedisPoolConfig.setMaxIdle(1500); //最大空闲连接数
jedisPoolConfig.setMinIdle(500); //最小空闲连接数
jedisPoolConfig.setMaxWaitMillis(20000); //获取连接时最大等待时间
jedisPoolConfig.setTestOnBorrow(true); //获取连接时检查是否可用
jedisPoolConfig.setTestOnReturn(true); //返回连接时检查是否可用
jedisPoolConfig.setTestWhileIdle(true); //是否开启空闲资源监测
jedisPoolConfig.setTimeBetweenEvictionRunsMillis(300000); //-1不检测 单位为毫秒 空闲资源监测周期
jedisPoolConfig.setMinEvictableIdleTimeMillis(30*60*1000);//资源池中资源最小空闲时间 单位为毫秒 达到此值后空闲资源将被移除
jedisPoolConfig.setNumTestsPerEvictionRun(300); //做空闲监测时，每次采集的样本数 -1代表对所有连接做监测
return jedisPoolConfig;
}
@Bean
public Redisson redisson(){
Config config=new Config();
// config.useSingleServer().setAddress("redis://192.168.0.104:6379").setDatabase(0);
config.useClusterServers().addNodeAddress("redis://192.168.0.104:7000").addNodeAddress("redis://192.168.0.104:7001")
.addNodeAddress("redis://192.168.0.104:7002").addNodeAddress("redis://192.168.0.104:7003")
.addNodeAddress("redis://192.168.0.104:7004").addNodeAddress("redis://192.168.0.104:7005");
Redisson redisson = (Redisson) Redisson.create(config);
return redisson;
}
// @Bean
// public JedisPool jedisPool(JedisPoolConfig jedisPoolConfig){
// return new JedisPool(jedisPoolConfig,"192.168.0.104",6379);
// }
@Bean
public JedisConnectionFactory jedisConnectionFactory(JedisPoolConfig jedisPoolConfig){
JedisClientConfiguration jedisClientConfiguration = JedisClientConfiguration.builder().usePooling().poolConfig(jedisPoolConfig).and().readTimeout(Duration.ofMillis(2000)).build();
RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();
redisStandaloneConfiguration.setDatabase(0);
redisStandaloneConfiguration.setPort(6379);
redisStandaloneConfiguration.setHostName("192.168.0.104");
return new JedisConnectionFactory(redisStandaloneConfiguration, jedisClientConfiguration);
}
@Bean
public RedisTemplate<String,Object> redisTemplate(RedisConnectionFactory jedisConnectionFactory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
// 配置连接工厂
template.setConnectionFactory(jedisConnectionFactory);
//使用Jackson2JsonRedisSerializer来序列化和反序列化redis的value值（默认使用JDK的序列化方式）
Jackson2JsonRedisSerializer jacksonSeial = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
// 指定要序列化的域，field,get和set,以及修饰符范围，ANY是都有包括private和public
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
// 指定序列化输入的类型，类必须是非final修饰的，final修饰的类，比如String,Integer等会跑出异常
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
om.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
jacksonSeial.setObjectMapper(om);
// 值采用json序列化
template.setValueSerializer(jacksonSeial);
//使用StringRedisSerializer来序列化和反序列化redis的key值
template.setKeySerializer(new StringRedisSerializer());
// 设置hash key 和value序列化模式
template.setHashKeySerializer(new StringRedisSerializer());
template.setHashValueSerializer(jacksonSeial);
template.afterPropertiesSet();
return template;
}
/**
* 对hash类型的数据操作
*
* @param redisTemplate
* @return
*/
@Bean
public HashOperations<String, String, Object> hashOperations(RedisTemplate<String, Object> redisTemplate) {
return redisTemplate.opsForHash();
}
/**
* 对redis字符串类型数据操作
*
* @param redisTemplate
* @return
*/
@Bean
public ValueOperations<String, Object> valueOperations(RedisTemplate<String, Object> redisTemplate) {
return redisTemplate.opsForValue();
}
/**
* 对链表类型的数据操作
*
* @param redisTemplate
* @return
*/
@Bean
public ListOperations<String, Object> listOperations(RedisTemplate<String, Object> redisTemplate) {
return redisTemplate.opsForList();
}
/**
* 对无序集合类型的数据操作
*
* @param redisTemplate
* @return
*/
@Bean
public SetOperations<String, Object> setOperations(RedisTemplate<String, Object> redisTemplate) {
return redisTemplate.opsForSet();
}
/**
* 对有序集合类型的数据操作
*
* @param redisTemplate
* @return
*/
@Bean
public ZSetOperations<String, Object> zSetOperations(RedisTemplate<String, Object> redisTemplate) {
return redisTemplate.opsForZSet();
}
}
redis初始化
@Component
public class RedisDataInit {
@Autowired
OrderService orderService;
@Autowired
RedisBloomFilter redisBloomFilter;
@PostConstruct
public void init(){
List<Order> orders = orderService.selectOrderyAll();
for (Order order : orders) {
redisBloomFilter.put(String.valueOf(order.getId()));
}
}
}
缓存查询模板类
@Component
public class CacheTemplate<T> {
@Autowired
private ValueOperations<String,Object> valueOperations;
@Autowired
RedisBloomFilter bloomFilter;
@Autowired
private RedisLock redisLock;
@Autowired
private Redisson redisson;
@Autowired
OrderMapper orderMapper;
public R findCache(String key, long expire, TimeUnit unit, CacheLoadble<T> cacheLoadble){
//查询缓存
Object redisObj = valueOperations.get(String.valueOf(key));
if(redisObj!=null){
return new R().setCode(200).setData(redisObj).setMsg("OK");
}
synchronized (this){
redisObj = valueOperations.get(String.valueOf(key));
if(redisObj!=null){
return new R().setCode(200).setData(redisObj).setMsg("OK");
}
T load = cacheLoadble.load();
valueOperations.set(String.valueOf(key), load,expire, unit); //加入缓存
return new R().setCode(200).setData(load).setMsg("OK");
}
}
public R redisFindCache(String key, long expire, TimeUnit unit, CacheLoadble<T> cacheLoadble,boolean b){
//判断是否走过滤器
if(b){
//先走过滤器
boolean bloomExist = bloomFilter.isExist(String.valueOf(key));
if(!bloomExist){
return new R().setCode(600).setData(null).setMsg("查询无果");
}
}
//查询缓存
Object redisObj = valueOperations.get(String.valueOf(key));
//命中缓存
if(redisObj != null) {
//正常返回数据
return new R().setCode(200).setData(redisObj).setMsg("OK");
}
//缓存不命中
// if(redisLock.tryLock(key)){
RLock lock0 = redisson.getLock("{taibai0}:"+key);
RLock lock1 = redisson.getLock("{taibai1}:" + key);
RLock lock2 = redisson.getLock("{taibai2}:" + key);
RedissonMultiLock lock = new RedissonMultiLock(lock0,lock1, lock2);
try {
// redisLock.lock(key);
lock.lock();
//查询缓存
redisObj = valueOperations.get(String.valueOf(key));
//命中缓存
if(redisObj != null) {
//正常返回数据
return new R().setCode(200).setData(redisObj).setMsg("OK");
}
// T load = cacheLoadble.load();//查询数据库
Order load = orderMapper.selectOrderById(Integer.valueOf(key));
if (load != null) {
valueOperations.set(key, load,expire, unit); //加入缓存
return new R().setCode(200).setData(load).setMsg("OK");
}
return new R().setCode(500).setData(new NullValueResultDO()).setMsg("查询无果");
}finally {
// redisLock.unlock(key);
lock.unlock();
}
// }
// else {
// return redisFindCache(key,expire,unit,cacheLoadble,false);
// }
}
}
缓存查询业务类
@Service
public class OrderServiceImpl implements OrderService{
@Autowired
OrderMapper orderMapper;
@Autowired
ValueOperations valueOperations;
@Autowired
CacheTemplate cacheTemplate;
@Autowired
RedisBloomFilter bloomFilter;
@Autowired
private RedisLock redisLock;
@Override
public Integer insertOrder(Order order) {
Integer integer = orderMapper.insertOrder(order);
return integer;
}
@Override
public R selectOrderById(Integer id) {
return cacheTemplate.redisFindCache(String.valueOf(id), 10, TimeUnit.MINUTES, new CacheLoadble<Order>() {
@Override
public Order load() {
return orderMapper.selectOrderById(id);
}
},true);
}
// @Override
// public R selectOrderById(Integer id) {
// if(!bloomFilter.isExist(String.valueOf(id))){
// return new R().setCode(600).setData(null).setMsg("查询无果");
// }
// //查询缓存
// Object redisObj = valueOperations.get(String.valueOf(id));
// //命中缓存
// if(redisObj != null) {
// //正常返回数据
// return new R().setCode(200).setData(redisObj).setMsg("OK");
// }
// try {
// if(redisLock.tryLock(String.valueOf(id))){
// //查询缓存
// redisObj = valueOperations.get(String.valueOf(id));
// //命中缓存
// if(redisObj != null) {
// //正常返回数据
// return new R().setCode(200).setData(redisObj).setMsg("OK");
// }
// Order order = orderMapper.selectOrderById(id);
// if (order != null) {
// valueOperations.set(String.valueOf(id), order); //加入缓存
// return new R().setCode(200).setData(order).setMsg("OK");
// }
// }else{
// selectOrderById(id);
// }
// }finally {
// redisLock.unlock(String.valueOf(id));
// }
//
//
// return new R().setCode(500).setData(new NullValueResultDO()).setMsg("查询无果");
// }
public R synchronizedSelectOrderById(Integer id) {
return cacheTemplate.findCache(String.valueOf(id), 10, TimeUnit.MINUTES, new CacheLoadble<Order>() {
@Override
public Order load() {
return orderMapper.selectOrderById(id);
}
});
}
@Override
public List<Order> selectOrderyAll() {
return orderMapper.selectOrderyAll();
}
}
RedLock优势
redis提供的RedLock锁主要有两大优势：
- 可以解决单点故障，因为这个锁也是基于分布式实现的，锁的数据会分别存放多个redis服务器中，每一次解锁和获取锁必须有超过半数的redis服务器同意才能成功完成。
- 解决锁过期时间问题，在上面我们自己实现的redis锁中，我们对于锁的过期还需要做出判断，因为我们不能很准确设置锁的过期时间，对于生产环境下更是难以判断。redLock对于锁的过期提供了一种看门狗机制，它相当于一个进程，会帮助我们监控拿到锁的进程，如果锁过期了但业务逻辑还没执行完，这个进程会主动延长锁的过期时间，知道我们主动释放锁。而且如果我们的业务进程突然宕掉了，它也会监控到，自然就不会延续锁的过期时间。
缓存更新策略
如果在使用缓存的时候，如果缓存中的数据发生了改变，我们就需要更新缓存和数据库中的数据，那么究竟是先更新数据库还是先更新缓存呢？我们该选择怎么样的缓存更新策略呢？
先更新数据库再更新缓存
这种策略带来的问题是如果出现缓存更新失败，那么缓存的就是旧数据，这样就算数据库中的数据是新数据也会出现数据不一致的问题。
而且如果存在网络问题，线程B后更新的数据库，但是却先更新了缓存，结果线程A的更新的数据存在了缓存中，同样也造成了数据不一致的问题。
对于上面这种数据不一致的问题我们可以不更新缓存，而是直接将缓存删掉，这样再查询的时候再从数据库中查出来数据放在缓存中，这样自然就不会出现缓存不一致的问题，这种方式还节省了更新缓存带来的性能消耗，而且对于那种经常更新却不常查询的数据来说，这种方式是最能提高性能的。
先删除缓存再更新数据库
先删除缓存再更新数据库，如果数据库更新失败了，缓存中是空的，数据库里的是旧数据，这样也不会造成数据不一致的问题。
但这种方式也会造成其他问题，如果一个进程首先删除了缓存，正要更新数据库，此时另一个进程也来查询数据，首先查缓存，没有数据接着会查询数据库，第一个进程这时还没有更新数据，所以会查到旧数据，然后放到缓存中，那么等第一个进程更新完数据库，又会出现数据不一致的问题。
对于这种问题这里提供两种解决方案，分别是延迟双删和串行化。
延迟双删
在更新完数据库后再删除一次缓存的数据，保证数据的一致性。
串行化
串行化保证了两个进程间执行操作的顺序，也就不存在缓存不一致的情况，不过很影响性能，一般会采取上面延迟双删的策略。