缓存雪崩：因为缓存失效（key生存时间到期）导致所有请求都去查找数据库，导致数据库CPU和内存负载过高导致宕机。
缓存雪崩原因及解决方案：
使用缓存主要解决数据同步，并减少对数据库访问次数。因此，通常解决方案往往是使用互斥锁，让一个线程访问数据库，并将数据更新到缓存中，其他线程访问缓存中数据。如果是基于jvm锁机制的话，只能解决单机问题，也就是只让本机一个线程访问缓存，但是分布式条件下是不能使用的。所以，要基于缓存的分布式锁来实现。
以redis为例解释下实现分布式锁的原理:
获取锁：
所有线程操作一个共同的key比如:lock，如果redis中不存在key为lock的值，那么当前线程获取锁并为lock设置一个随机值。
如果lock已经存在了，说明已经有线程获取锁，该线程不能再获取了。
释放锁：
获取锁的线程操作执行完毕后，清除lock的值，这样锁就释放了。所以，对锁的操作就是通过对同一个key值的添加和删除操作。
代码：
@Service
public class RedisLock implements Lock {
@Autowired
private JedisConnectionFactory factory;
private static final String LOCK="lock";
private ThreadLocal<String> local=new ThreadLocal<String>();
//获取锁
@Override
public boolean tryLock() {
//获取Jedis的原始数据连接
Jedis jedis = (Jedis)factory.getConnection().getNativeConnection();
String uuid = UUID.randomUUID().toString();
/** 获取锁：设置一个随机值，超期时间1s
String key, String value, String nxxx, String expx, int time)
nxxx: NX:key不存在时设值 XX:key存在时设值
expx: EX|PX, expire time units: EX = seconds; PX = milliseconds
*/
String ret = jedis.set(LOCK, uuid, "NX", "PX", 1000);
if(!StringUtils.isEmpty(ret)&&ret.equals("OK")){
local.set(uuid);
return true;
}
return false;
}
/**
* 解锁
*/
@Override
public void unlock() {
String script=null;
try {
script=FileCopyUtils.copyToString(new FileReader(ResourceUtils.getFile("classpath:cn/rjx/spring/cache/unlock.c")));
} catch (IOException e) {
e.printStackTrace();
}
Jedis jedis = (Jedis)factory.getConnection().getNativeConnection();
List<String> keys=new ArrayList<String>();
keys.add(LOCK);
List<String> args=new ArrayList<String>();
args.add(local.get());
//如果redis中的 lock值和当前线程的uuid值相等，删除Key值
jedis.eval(script, keys, args);
}
}
删除键值是执行的脚本unlock.c：
if redis.call("get",KEYS[1])==ARGV[1] then
return redis.call("del",KEYS[1])
else
return 0
end操作缓存的具体流程：
1.当线程查询某一值时先查看缓存是否存在该值。
2.如果存在直接返回主缓存中的数据。
3.1如果不存在，只有一个线程获取锁并去数据库读取数据，读取后更新主缓存和备份缓存。
3.2 其他线程取备份缓存中的数据。.
代码实现：初始时，主缓存和备份缓存为空，此时可能会有线程获取的值为空，但是并不影响用户体验，用户可以再刷新一次。在要求比较高的场景里面，可以考虑先把数据写入缓存中，可以搭配定时刷新缓存的机制。
public List<Integer> queryCountByLeiMu() {
List<Integer> cacheResult = cacheService.cacheResult("101", "leimu");
if(cacheResult!=null){
logger.info("================get cache=======================");
return cacheResult;
}
if(lock.tryLock()){
logger.info("================get db=======================");
List<Integer> list=empDao.queryCountByLeiMu();
cacheService.cachePut("101", list, "leimu");//主缓存
cacheService.cachePut("beifen101", list, "beifenleimu");//备份缓存
lock.unlock();
return list;
}else{
logger.info("================get BEIFEN=======================");
//备份中拿
return cacheService.cacheResult("beifen101", "beifenleimu");
}
}数据同步问题：主缓存中key的过期时间比较短，这样保证尽可能获取新数据。
bean.xml中缓存失效时间设置：
<!-- 开启缓存注解扫描 -->
<cache:annotation-driven />
<bean id="cacheManager" class="org.springframework.data.redis.cache.RedisCacheManager">
<constructor-arg index="0" ref="redisTemplate"></constructor-arg>
<property name="expires">
<map>
<entry key="leimu" value="5"></entry>
<entry key="beifenleimu" value="100"></entry>
</map>
</property>
</bean> 测试方法模拟高并发情景：
@Autowired
LeiMuService leiMuService;
private static final int threadNum=13;
//倒计数器（发令枪） 用于制造线程并发执行
private static CountDownLatch cdl=new CountDownLatch(threadNum);
/**
* 模拟高并发条件下，数据库查询耗时比较长
* @throws InterruptedException
*/
@Test
public void test04() throws InterruptedException{
for(int i=0;i<threadNum;i++){
new Thread(new UserRequest()).start();
cdl.countDown();//threadNum每次减1，到零时同时执行cdl.await();后边代码
}
//主线程挂起，等子线程执行完以后
Thread.currentThread().join();
}
private class UserRequest implements Runnable{
@Override
public void run() {
//所有子线程在这里等待，当所有线程实例化后，同时停止等待
try {
cdl.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
//N个子线程同时调用获取类目
List<Integer> leimu = leiMuService.queryCountByLeiMu();
logger.info(Thread.currentThread().getName()+"==========================================>"+leimu.size());
}
}
}缺点：1.非阻塞，短时间不能保证数据一致性
2.锁失效时间难把握，一般为单线程处理时长的两到三倍
3.可能出现锁失效情况
4******不能在redis集群环境中使用（集群中可用redLock）
建议使用基于zookeeper的分布式锁实现方式！！.
