针对随机key穿透,我们可以使用布隆过滤器

布隆过滤器

布隆过滤器数据一致性

执行流程

2.缓存击穿

缓存击穿是指:

• 大量并发查询一个热点数据,但是呢我们的热点数据在某一刻刚好过期了,这样大量的并发请求会先经过缓存,但是缓存中没有,再进入布隆过滤器bloom保存了该热点数据的ID所以会让请求去查询数据库,结果这大量请求就把数据库压垮了

风险：

• 由于缓存某一刻会过期,刚好该时刻大量并发出来,数据库瞬时压力增大，最终导致崩溃

解决办法:

• 加锁: 本地锁: 直接使用synchronize,juc.lock不适用于分布式情况,分布式下他们只能锁住当前自己的服务
• 分布式锁:

分布式锁阶段演进

• 加锁,就是"抢坑位"

• 第一阶段

• 第二阶段

• 第三阶段

• 第四阶段

• 第五阶段

• Redis原生实现分布式锁核心代码如下:

/**
* 根据skuId查询商品详情
*
* 使用Redis实现分布式锁:
*  解决大并发下,缓存击穿|穿透问题
*
* @param skuId
* @return
*/
@Override
public SkuItemTo findSkuItem(Long skuId) {
// 缓存key
String cacheKey = RedisConstants.SKU_CACHE_KEY_PREFIX + skuId;
// 查询缓存
SkuItemTo data = cacheService.getData(RedisConstants.SKU_CACHE_KEY_PREFIX + skuId, new TypeReference<SkuItemTo>() {
});
// 判断是否命中缓存
if (data == null) {
// 缓存没有,回源查询数据库.但是这个操作之前先问一下bloom是否需要回源
if (skuIdBloom.contains(skuId)) {
// bloom返回true说明数据库中有
log.info("缓存没有,bloom说有,回源");
SkuItemTo skuItemTo = null;
// 使用UUID作为锁的值,防止修改别人的锁
String value = UUID.randomUUID().toString();
// 摒弃setnx ,加锁个设置过期时间不是原子的
// 原子加锁,防止被击穿 分布式锁 设置过期时间
Boolean ifAbsent = stringRedisTemplate.opsForValue()
.setIfAbsent(RedisConstants.LOCK, value, RedisConstants.LOCK_TIMEOUT, TimeUnit.SECONDS);
if (ifAbsent) {
try {
// 设置自动过期时间,非原子的,加锁和设置过期时间不是原子的操作,所以会出现问题
// stringRedisTemplate.expire(RedisConstants.LOCK, RedisConstants.LOCK_TIMEOUT, TimeUnit.SECONDS);

// 大量请求,只有一个抢到锁
log.info(Thread.currentThread().getName() + "抢到锁,查询数据库");
skuItemTo = findSkuItemDb(skuId); // 执行回源查询数据库
// 把数据库中查询的数据缓存里存一份
cacheService.saveData(cacheKey, skuItemTo);
} finally { // 解锁前有可能出现各种问题导致解锁失败,从而出现死锁
// 释放锁,非原子,不推荐使用
// String myLock = stringRedisTemplate.opsForValue().get(RedisConstants.LOCK);

//删锁： 【对比锁值+删除(合起来保证原子性)】
String deleteScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Long executeResult = stringRedisTemplate.execute(new DefaultRedisScript<Long>(deleteScript,Long.class),
Arrays.asList(RedisConstants.LOCK), value);

// 判断是否解锁成功
if (executeResult.longValue() == 1) {
log.info("自己的锁:{},解锁成功", value);
stringRedisTemplate.delete(RedisConstants.LOCK);
} else {
log.info("别人的锁,解不了");
}
}
} else {
// 抢锁失败,自旋抢锁. 但是实际业务为我们只需要让让程序缓一秒再去查缓存就好了
try {
log.info("抢锁失败,1秒后去查询缓存");
Thread.sleep(1000);
data = cacheService.getData(RedisConstants.SKU_CACHE_KEY_PREFIX + skuId, new TypeReference<SkuItemTo>() {
});
return data;
} catch (InterruptedException e) {
}
}
return skuItemTo;
} else {
log.info("缓存没有,bloom也说没有,直接打回");
return data;
}
}
log.info("缓存中有数据,直接返回,不回源");
// 价格不缓存,有些需要变的数据,可以"现用现拿"
Result<BigDecimal> decimalResult = productFeignClient.findPriceBySkuId(skuId);
if (decimalResult.isOk()) {
BigDecimal price = decimalResult.getData();
data.setPrice(price);
}
return data;
}

• Redisson框架实现分布式锁

3.缓存雪崩

缓存雪崩是指:

• 大量key同时过期，正好百万请求进来，全部要查这些数据？一查数据库就炸了

解决办法:

• 过期时间+随机值防止大面积同时失效； 单点失效，自然会由防击穿来加锁处理

@Override
public void saveData(String key, Object data) {
if (data == null) {
// 缓存null值,防止缓存穿透.设置缓存过期时间
stringRedisTemplate.opsForValue().set(key, cacheConfig.getNullValueKey(), cacheConfig.getNullValueTimeout(), cacheConfig.getNullTimeUnit());
} else {
// 为了防止缓存同时过期,发生缓存雪崩.给每个缓存过期时间加上随机值
Double value = Math.random() * 10000000L;
long mill = 1000 * 60 * 24 * 3 + value.intValue();
stringRedisTemplate.opsForValue().set(key, JsonUtils.objectToJson(data),
mill, cacheConfig.getDataTimeUnit());
}
}