防止缓存击穿的几种方案每个 key 分别加分布式锁逻辑过期异步更新缓存永不过期监听数据库 binlog 更新缓存使用本地锁而不是分布式锁缓存击穿指的是热点 key 过期导致大量请求都直接访问数据库数据库的压力剧增甚至直接导致数据库宕机。每个 key 分别加分布式锁先查缓存如果缓存存在则直接返回不存在则加分布式锁保证一个 key 只有一个线程处理加锁后再次检查缓存是否存在存在则直接返回不存在则从数据库查询数据然后填充缓存并返回最后解锁。加锁后再次检查缓存是否存在是为了防止其他等待线程拿到锁后重复查询数据库导致数据库的压力剧增。适用于对数据一致性要求较高的场景。packageexample;importlombok.RequiredArgsConstructor;importorg.redisson.api.RBucket;importorg.redisson.api.RLock;importorg.redisson.api.RedissonClient;importorg.springframework.stereotype.Service;importjava.util.concurrent.TimeUnit;RequiredArgsConstructorServicepublicclassDistributedLockCacheService{privatefinalRedissonClientredissonClient;publicStringget(Stringkey){// get cacheRBucketStringcacheredissonClient.getBucket(key);Stringdatacache.get();if(data!null){returndata;}RLocklockredissonClient.getLock(lock:key);booleanlocked;try{// 加分布式锁lockedlock.tryLock(3L,TimeUnit.SECONDS);}catch(InterruptedExceptione){Thread.currentThread().interrupt();thrownewRuntimeException(获取锁被中断,e);}if(!locked){thrownewRuntimeException(获取锁超时);}try{// double checkdatacache.get();if(data!null){returndata;}datagetFromDatabase(key);// set cache、expirecache.set(data,30,TimeUnit.MINUTES);returndata;}finally{lock.unlock();}}privateStringgetFromDatabase(Stringkey){returnxx;}}逻辑过期异步更新缓存缓存不设置过期时间而是在数据中添加一个逻辑过期字段。查询缓存如果缓存为空必须使用分布式锁方案填充缓存不能使用异步更新缓存方案因为可能多个线程都发现缓存为空会顺序多次重复更新缓存。另一个原因是异步更新缓存方案只能返回旧数据这里只能返回null用户体验不好。如果缓存不为空则查看缓存的逻辑过期时间是否已过期如果已过期则需要提交异步任务更新缓存。最后返回旧数据。适用于对数据一致性要求不高的场景。packageexample;importlombok.RequiredArgsConstructor;importorg.redisson.api.RBucket;importorg.redisson.api.RLock;importorg.redisson.api.RedissonClient;importorg.springframework.stereotype.Service;importjava.util.concurrent.ThreadPoolExecutor;RequiredArgsConstructorServicepublicclassLogicalExpireCacheService{privatefinalRedissonClientredissonClient;privatefinalDistributedLockCacheServicedistributedLockCacheService;privatefinalThreadPoolExecutorthreadPoolExecutor;publicStringget(Stringkey){// get cacheRBucketStringcacheredissonClient.getBucket(key);StringcacheDatacache.get();if(cacheDatanull){// 缓存为空使用分布式锁方案填充缓存returndistributedLockCacheService.get(key);}else{// 0: value, 1 expireAtString[]parsedCacheDataparseCacheData(cacheData);// 缓存已过期if(isExpired(parsedCacheData[1])){// 异步更新缓存updateCacheAsync(cache,key);}// 返回旧数据returnparsedCacheData[0];}}privatevoidupdateCacheAsync(RBucketStringcache,Stringkey){// 提交异步任务更新缓存threadPoolExecutor.execute(()-{RLocklockredissonClient.getLock(lock:key);// 加分布式锁注意这里加锁不需要 timeout 阻塞等待加锁失败说明其他线程正在更新缓存if(lock.tryLock()){try{StringcacheDatacache.get();String[]parsedCacheDataparseCacheData(cacheData);// 加锁后再次检查是否过期防止重复更新if(isExpired(parsedCacheData[1])){StringdatagetFromDatabase(key);// set cachecache.set(buildCacheData(data,System.currentTimeMillis()));}}finally{// 解锁lock.unlock();}}});}privateString[]parseCacheData(StringcacheData){intindexcacheData.lastIndexOf(:);returnnewString[]{cacheData.substring(0,index),cacheData.substring(index1)};}privatebooleanisExpired(StringexpireAt){// 缓存已过期returnLong.parseLong(expireAt)System.currentTimeMillis();}privateStringbuildCacheData(Stringdata,LongexpireAt){returndata:expireAt;}privateStringgetFromDatabase(Stringkey){returnxx;}}以下是我之前的错误实现加锁和解锁不在同一个线程导致永远无法解锁千万不要这样写privatevoidupdateCacheAsync(RBucketStringcache,Stringkey){RLocklockredissonClient.getLock(lock:key);// 加分布式锁注意这里加锁不需要 timeout 阻塞等待加锁失败说明其他线程正在更新缓存if(lock.tryLock()){// 提交异步任务更新缓存threadPoolExecutor.execute(()-{try{StringdatagetFromDatabase(key);// set cachecache.set(buildCacheData(data,System.currentTimeMillis()));}finally{// 解锁lock.unlock();}});}}永不过期监听数据库 binlog 更新缓存缓存不设置过期时间如果缓存为空使用分布式锁方案兜底监听数据库 binlog 实时更新缓存。数据一致性很高。使用本地锁而不是分布式锁对于机器数较少、并发不高的场景我个人认为可以使用本地锁代替分布式锁能够减少分布式锁的性能损耗代价是 key 不在缓存中时重复填充缓存的次数最多等于机器数。packageexample;importlombok.RequiredArgsConstructor;importorg.redisson.api.RBucket;importorg.redisson.api.RedissonClient;importorg.springframework.stereotype.Service;importjava.util.concurrent.ConcurrentHashMap;importjava.util.concurrent.TimeUnit;RequiredArgsConstructorServicepublicclassLocalLockCacheService{privatefinalRedissonClientredissonClient;privatefinalConcurrentHashMapString,ObjectlocksnewConcurrentHashMap();publicStringget(Stringkey){// get cacheRBucketStringcacheredissonClient.getBucket(key);Stringdatacache.get();if(data!null){returndata;}// 类似 redis 的 set nxObjectlocklocks.computeIfAbsent(key,k-newObject());synchronized(lock){// double checkdatacache.get();if(data!null){returndata;}datagetFromDatabase(key);// set cache、expirecache.set(data,30,TimeUnit.MINUTES);// 删除锁对象防止 locks 导致 OOMlocks.remove(key,lock);}returndata;}privateStringgetFromDatabase(Stringkey){returnxx;}}