业务背景
在分发服务中，我们需要调用优惠券模板的相关信息。为了避免各个服务中重复实现模板查询功能，我们首先编写了一个通用的引擎层模板查询方法，以支持 C 端用户和内部应用的查询。像这种大流量的接口，肯定是需要放到缓存的。
用户常规访问优惠券模板时序图如下：
什么是缓存击穿？
缓存击穿指在高并发的系统中，一个热点数据缓存过期或者在缓存中不存在，导致大量并发请求直接访问数据库，从而给数据库造成巨大压力，甚至可能引起宕机。
1. 预热和缓存永不过期
一般来说，我们会通过预热和缓存永不过期的机制让缓存不击穿，这样即使再大的流量也可以通过缓存去抗。
- 缓存预热：热点数据预加载，指的是在活动或者大促开始前，针对已知的热点数据从数据库加载到缓存中，这样可以避免海量请求第一次访问热点数据需要从数据库读取的流程。
- 永不过期：热点数据永不过期，指的就是可以预知的热点数据，在活动开始前，设置过期时间为 -1。这样的话，就不会有缓存击穿的风险。
上面两个一般都是搭配一起使用的。等对应热点缓存的活动结束后，这些数据访问量就比较低了，可以通过后台任务的方案对指定缓存设置过期时间，这样可以有效降低 Redis 存储压力。
2. 分布式锁之双重判定锁
分布式锁的解决方案就是保证只有一个请求可以访问数据库，其它请求等待结果。这样可以避免大量的请求同时访问数据库。
但是这种的话有一个弊端，那就是获取分布式锁的请求，都会执行一遍查询数据库，并更新到缓存。理论上只有第一个加载数据库记录请求是有效的。
针对这个问题，可以通过双重判定锁的形式，在获取到分布式锁之后，再次查询一次缓存是否存在。如果缓存中存在数据，就直接返回；如果不存在，才继续执行查询数据库的操作。这样就可以避免大量请求访问数据库。
public String selectTrain(String id) {
// 查询缓存不存在，去数据库查询并放入到缓存
String cacheData = cache.get(id);
if (StrUtil.isBlank(cacheData)) {
// 为避免大量请求同时访问数据库，通过分布式锁减少数据库访问量
Lock lock = getLock(id);
lock.lock();
try {
// 获取锁后双重判定
cacheData = cache.get(id);
// 理论上只有第一个请求加载数据库是有效的，因为它加载后会把数据放到缓存
// 后面的请求再请求数据库加载缓存就没有必要了
if (StrUtil.isBlank(cacheData)) {
// 获取数据库中存在的数据
String dbData = trainMapper.selectId(id);
if (StrUtil.isNotBlank(dbData)) {
// 将查询到的数据放入缓存，下次查询就有数据了
cahce.set(id, dbData);
cacheData = dbData;
}
}
} finally {
lock.unlock();
}
}
return cacheData;
}
3. 高并发极端情况
很多同学认为到这里就结束了，但这恰恰只是开始，真正难得是接下来要讲的。
我举个场景，有一万个请求同一时间访问触发了缓存击穿，如果用双重判定锁，逻辑是这样的：
- 第一个请求加锁、查询缓存是否存在、查询数据库、放入缓存、解锁，假设我们用了50毫秒；
- 第二个请求拿到锁查询缓存、解锁用了1毫秒；
- 那最后一个请求需要等待10049毫秒后才能返回，用户等待时间过长，极端情况下可能会触发应用的内存溢出。
内存泄露是指本来无用的对象却继续占用 内存，没有在恰当的时机释放占用的内存。 不使用的内存，却没有被释放，这个就叫 做内存泄露。 比较典型的场景是:每一个请求进来，或 者每一次操作处理，都分配了内存，却有一部 分不能回收，那么随着处理的请求越来越多 内存泄露也就越来越严重。 与内存溢出的关系:如果存在严重的内存 泄露问题，随着时间的推移，则必然会引起内 存溢出。内存泄露一般是资源管理问题和程序 BUG，内存溢出则是内存空间不足和内存泄露 的最终结果。
3.1 尝试获取锁 tryLock
像上面这种场景，类似于秒杀的架构，我们要做的就是不让用户请求在服务端阻塞过长时间。那就可以使用尝试获取锁 tryLock API，它的语义是如果拿锁失败直接返回，而不是阻塞等待直到获取锁。
public String selectTrain(String id) {
// 查询缓存不存在，去数据库查询并放入到缓存
String cacheData = cache.get(id);
if (StrUtil.isBlank(cacheData)) {
// 为避免大量请求同时访问数据库，通过分布式锁减少数据库访问量
Lock lock = getLock(id);
// 尝试获取锁，获取失败直接返回用户请求，并提醒用户稍后再试
if (!lock.tryLock()) {
throw new RuntimeException("当前访问人数过多，请稍候再试...");
}
try {
// 获取数据库中存在的数据
String dbData = trainMapper.selectId(id);
if (StrUtil.isNotBlank(dbData)) {
// 将查询到的数据放入缓存，下次查询就有数据了
cahce.set(id, dbData);
cacheData = dbData;
}
} finally {
lock.unlock();
}
}
return cacheData;
}
通过这种方式我们可以快速失败，告诉用户网络异常请稍后再试，等用户再尝试刷新的时候，其实获取锁的线程已经把数据放到了缓存。
因为这种方案对用户操作体验不友好，所以也只是适用于部分场景。在实际开发中，需要灵活变更。
项目实战
我们省去了简单的分布式过程，上面已经有简单的示例，直接用较常见的双重判定锁。
@Override
public CouponTemplateQueryRespDTO findCouponTemplate(CouponTemplateQueryReqDTO requestParam) {
// 查询 Redis 缓存中是否存在优惠券模板信息
String couponTemplateCacheKey = String.format(EngineRedisConstant.COUPON_TEMPLATE_KEY, requestParam.getCouponTemplateId());
Map<Object, Object> couponTemplateCacheMap = stringRedisTemplate.opsForHash().entries(couponTemplateCacheKey);
// 如果存在直接返回，不存在需要通过双重判定锁的形式读取数据库中的记录
if (MapUtil.isEmpty(couponTemplateCacheMap)) {
// 获取优惠券模板分布式锁
RLock lock = redissonClient.getLock(String.format(EngineRedisConstant.LOCK_COUPON_TEMPLATE_KEY, requestParam.getCouponTemplateId()));
lock.lock();
try {
// 通过双重判定锁优化大量请求无意义查询数据库
couponTemplateCacheMap = stringRedisTemplate.opsForHash().entries(couponTemplateCacheKey);
if (MapUtil.isEmpty(couponTemplateCacheMap)) {
LambdaQueryWrapper<CouponTemplateDO> queryWrapper = Wrappers.lambdaQuery(CouponTemplateDO.class)
.eq(CouponTemplateDO::getShopNumber, Long.parseLong(requestParam.getShopNumber()))
.eq(CouponTemplateDO::getId, Long.parseLong(requestParam.getCouponTemplateId()))
.eq(CouponTemplateDO::getStatus, CouponTemplateStatusEnum.ACTIVE.getStatus());
CouponTemplateDO couponTemplateDO = couponTemplateMapper.selectOne(queryWrapper);
// 优惠券模板不存在或者已过期直接抛出异常
if (couponTemplateDO == null) {
throw new ClientException("优惠券模板不存在或已过期");
}
// 通过将数据库的记录序列化成 JSON 字符串放入 Redis 缓存
CouponTemplateQueryRespDTO actualRespDTO = BeanUtil.toBean(couponTemplateDO, CouponTemplateQueryRespDTO.class);
Map<String, Object> cacheTargetMap = BeanUtil.beanToMap(actualRespDTO, false, true);
Map<String, String> actualCacheTargetMap = cacheTargetMap.entrySet().stream()
.collect(Collectors.toMap(
Map.Entry::getKey,
entry -> entry.getValue() != null ? entry.getValue().toString() : ""
));
// 通过 LUA 脚本执行设置 Hash 数据以及设置过期时间
String luaScript = "redis.call('HMSET', KEYS[1], unpack(ARGV, 1, #ARGV - 1)) " +
"redis.call('EXPIREAT', KEYS[1], ARGV[#ARGV])";
List<String> keys = Collections.singletonList(couponTemplateCacheKey);
List<String> args = new ArrayList<>(actualCacheTargetMap.size() * 2 + 1);
actualCacheTargetMap.forEach((key, value) -> {
args.add(key);
args.add(value);
});
// 优惠券活动过期时间转换为秒级别的 Unix 时间戳
args.add(String.valueOf(couponTemplateDO.getValidEndTime().getTime() / 1000));
// 执行 LUA 脚本
stringRedisTemplate.execute(
new DefaultRedisScript<>(luaScript, Long.class),
keys,
args.toArray()
);
couponTemplateCacheMap = cacheTargetMap.entrySet()
.stream()
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
} finally {
lock.unlock();
}
}
return BeanUtil.mapToBean(couponTemplateCacheMap, CouponTemplateQueryRespDTO.class, false, CopyOptions.create());
}
优惠券模板缓存查询时序图如下所示：
