企业官网建设流程全解析

缓存击穿指的是热点 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;@RequiredArgsConstructor@ServicepublicclassDistributedLockCacheService{privatefinalRedissonClientredissonClient;publicStringget(Stringkey){// get cacheRBucket<String>cache=redissonClient.getBucket(key);Stringdata=cache.get();if(data!=null){returndata;}RLocklock=redissonClient.getLock("lock:"+key);booleanlocked;try{// 加分布式锁locked=lock.tryLock(3L,TimeUnit.SECONDS);}catch(InterruptedExceptione){Thread.currentThread().interrupt();thrownewRuntimeException("获取锁被中断",e);}if(!locked){thrownewRuntimeException("获取锁超时");}try{// double checkdata=cache.get();if(data!=null){returndata;}data=getFromDatabase(key);// set cache、expirecache.set(data,30,TimeUnit.MINUTES);returndata;}finally{lock.unlock();}}privateStringgetFromDatabase(Stringkey){return"xx";}}

逻辑过期，异步更新缓存

缓存不设置过期时间，而是在数据中添加一个逻辑过期字段。

查询缓存，如果缓存为空，必须使用分布式锁方案填充缓存，不能使用异步更新缓存方案，因为可能多个线程都发现缓存为空，会顺序多次重复更新缓存。另一个原因是异步更新缓存方案只能返回旧数据，这里只能返回null，用户体验不好。

如果缓存不为空，则查看缓存的逻辑过期时间是否已过期，如果已过期则需要提交异步任务更新缓存。最后返回旧数据。

适用于对数据一致性要求不高的场景。

packageexample;importlombok.RequiredArgsConstructor;importorg.redisson.api.RBucket;importorg.redisson.api.RLock;importorg.redisson.api.RedissonClient;importorg.springframework.stereotype.Service;importjava.util.concurrent.ThreadPoolExecutor;@RequiredArgsConstructor@ServicepublicclassLogicalExpireCacheService{privatefinalRedissonClientredissonClient;privatefinalDistributedLockCacheServicedistributedLockCacheService;privatefinalThreadPoolExecutorthreadPoolExecutor;publicStringget(Stringkey){// get cacheRBucket<String>cache=redissonClient.getBucket(key);StringcacheData=cache.get();if(cacheData==null){// 缓存为空，使用分布式锁方案填充缓存returndistributedLockCacheService.get(key);}else{// 0: value, 1 expireAtString[]parsedCacheData=parseCacheData(cacheData);// 缓存已过期if(isExpired(parsedCacheData[1])){// 异步更新缓存updateCacheAsync(cache,key);}// 返回旧数据returnparsedCacheData[0];}}privatevoidupdateCacheAsync(RBucket<String>cache,Stringkey){// 提交异步任务更新缓存threadPoolExecutor.execute(()->{RLocklock=redissonClient.getLock("lock:"+key);// 加分布式锁，注意这里加锁不需要 timeout 阻塞等待，加锁失败说明其他线程正在更新缓存if(lock.tryLock()){try{StringcacheData=cache.get();String[]parsedCacheData=parseCacheData(cacheData);// 加锁后再次检查是否过期，防止重复更新if(isExpired(parsedCacheData[1])){Stringdata=getFromDatabase(key);// set cachecache.set(buildCacheData(data,System.currentTimeMillis()));}}finally{// 解锁lock.unlock();}}});}privateString[]parseCacheData(StringcacheData){intindex=cacheData.lastIndexOf(":");returnnewString[]{cacheData.substring(0,index),cacheData.substring(index+1)};}privatebooleanisExpired(StringexpireAt){// 缓存已过期returnLong.parseLong(expireAt)<System.currentTimeMillis();}privateStringbuildCacheData(Stringdata,LongexpireAt){returndata+":"+expireAt;}privateStringgetFromDatabase(Stringkey){return"xx";}}

以下是我之前的错误实现，加锁和解锁不在同一个线程，导致永远无法解锁，千万不要这样写：

privatevoidupdateCacheAsync(RBucket<String>cache,Stringkey){RLocklock=redissonClient.getLock("lock:"+key);// 加分布式锁，注意这里加锁不需要 timeout 阻塞等待，加锁失败说明其他线程正在更新缓存if(lock.tryLock()){// 提交异步任务更新缓存threadPoolExecutor.execute(()->{try{Stringdata=getFromDatabase(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;@RequiredArgsConstructor@ServicepublicclassLocalLockCacheService{privatefinalRedissonClientredissonClient;privatefinalConcurrentHashMap<String,Object>locks=newConcurrentHashMap<>();publicStringget(Stringkey){// get cacheRBucket<String>cache=redissonClient.getBucket(key);Stringdata=cache.get();if(data!=null){returndata;}// 类似 redis 的 set nxObjectlock=locks.computeIfAbsent(key,k->newObject());synchronized(lock){// double checkdata=cache.get();if(data!=null){returndata;}data=getFromDatabase(key);// set cache、expirecache.set(data,30,TimeUnit.MINUTES);// 删除锁对象，防止 locks 导致 OOMlocks.remove(key,lock);}returndata;}privateStringgetFromDatabase(Stringkey){return"xx";}}