用户查询优惠卷功能之缓存常见问题解决
缓存击穿场景
当某个热点数据在缓存中过期时，如果此时有大量并发请求同时访问这个数据，由于缓存中不存在，所有请求都会直接访问数据库，导致数据库负载急剧增加。
接下来，我们将根据系统不同并发程度来探讨不同解决方案。
缓存预热和永不过期方案
- 缓存预热：热点数据预加载，指的是在活动或者大促开始前，针对已知的热点数据从数据库加载到缓存中，这样可以避免海量请求第一次访问热点数据需要从数据库读取的流程。
- 永不过期：热点数据永不过期，指的就是可以预知的热点数据，在活动开始前，设置过期时间为 -1。这样的话，就不会有缓存击穿的风险。
上面两个一般都是搭配一起使用的。等对应热点缓存的活动结束后，这些数据访问量就比较低了，可以通过后台任务的方案对指定缓存设置过期时间，这样可以有效降低 Redis 存储压力。
这种方案适合在并发不高且热点数据已知的情况下使用。
DCL 方案
这就是我们之前提到过的双重判定锁方案，这种方案对数据库是比较友好的：
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;
}这种方案较上面的方案来说具备更高的适应性，是绝大多数场景下的解决方案。
tryLock + 分布式锁分片
这种方案就是在极高并发下才可能使用，假设我们在使用上面的方案的基础上，存在如下场景：
假设有一万个请求同一时间访问触发了缓存击穿，如果用双重判定锁，逻辑是这样的：
- 第一个请求加锁、查询缓存是否存在、查询数据库、放入缓存、解锁，假设我们用了50毫秒；
- 第二个请求拿到锁查询缓存、解锁用了1毫秒；
- 那最后一个请求需要等待10049毫秒后才能返回，用户等待时间过长，极端情况下可能会触发应用的内存溢出。
考虑到我们做的是一个偏向于秒杀的场景，在 lock()和 tryLock()两个 API 之间的抉择问题上，建议遵循以下原则：
- 如果是类似于秒杀那种用户占便宜或者真正海量的场景，应该使用 tryLock() 方式。用户占便宜场景返回失败用户会尝试刷新，海量访问不使用 tryLock()是不行的，容易堵塞用户请求。
- 如果是类似于做 SaaS 的场景，用 tryLock()的话肯定就不行了，用户体验不佳，应当使用 lock()。
所以此处我们选择的是 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;
}通过这种方式我们可以快速失败，告诉用户网络异常请稍后再试，等用户再尝试刷新的时候，其实获取锁的线程已经把数据放到了缓存。
因为这种方案对用户操作体验不友好，所以也只是适用于部分场景。在实际开发中，需要灵活变更。
上面的方式对性能的限制比较大，所以接下来我们介绍的是性能略有提升的分布式锁分片方案。
相当于就是增加了分布式锁的数量，这个增加的数量直接决定了性能的提升程度： 
注意到，实际上对数据库进行查询的只有一个线程，因为 DCL 的存在，后面的获得了锁的线程因为在查询数据库之前再次检查缓存中的数据，发现数据已经存在（由第一个线程存放），所以不必访问数据库。
public String selectTrain(String id， String userId) {
// 查询缓存不存在，去数据库查询并放入到缓存
String cacheData = cache.get(id);
if (StrUtil.isBlank(cacheData)) {
// 假设设置10把分布式锁，那么就通过唯一标识（这里取用户ID）进行取模获取分片下标
int idx = Math.abs(userId.hashCode()) % 10;
// 为避免大量请求同时访问数据库，通过分布式锁减少数据库访问量
Lock lock = getLock(id + idx);
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;
}前面也说了，这种方案针对的是极端的高并发场景，基本上用不到，但 tryLock 的思想还是很重要的。
业务实践
原有代码：
@Override
public CouponTemplateQueryRespDTO findCouponTemplate(CouponTemplateQueryReqDTO requestParam) {
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);
return BeanUtil.toBean(couponTemplateDO, CouponTemplateQueryRespDTO.class);
}引入 DCL：
public class CouponTemplateServiceImpl extends ServiceImpl<CouponTemplateMapper, CouponTemplateDO> implements CouponTemplateService {
private final CouponTemplateMapper couponTemplateMapper;
private final StringRedisTemplate stringRedisTemplate;
private final RedissonClient redissonClient;
@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());
}
}整体业务逻辑如下：
性能压测
引入 DCL 前：
引入后： 
Comments NOTHING