当前位置: 首页 > 科技观察

张凯涛:京东业务数据应用级缓存示例

时间:2023-03-20 01:48:33 科技观察

1.多级缓存API封装我们的业务数据,比如商品分类、门店、商品基本信息等,都可以在本地进行适当的缓存,以提升性能。在多实例的情况下,不仅会使用本地缓存,还会使用分布式缓存,因此需要适当的API封装来简化缓存操作。1.本地缓存初始化publicclassLocalCacheInitServiceextendsBaseService{@OverridepublicvoidafterPropertiesSet()throwsException{//商品类别缓存CachecategoryCache=CacheBuilder.newBuilder().softValues().maximumSize(1000000).expireAfterWrite(Switches.CATEGORY.CATEGORY)/2,TimeUnit.SECONDS).build();addCache(CacheKeys.CATEGORY_KEY,categoryCache);}privatevoidaddCache(Stringkey,Cachecache){localCacheService.addCache(key,cache);}}本地缓存过期时间采用分布式缓存过期时间的一半,防止本地缓存数据缓存时间过长导致多个实例间数据不一致。另外缓存KEY前缀与本地缓存相关联,这样就可以通过匹配缓存KEY前缀找到关联的本地缓存。2.写缓存API包先写入本地缓存,如果需要写入分布式缓存,则异步更新分布式缓存。publicvoidset(finalStringkey,finalObjectvalue,finalintremoteCacheExpiresInSeconds)throwsRuntimeException{if(value==null){return;}//复制value对象//本地缓存是引用,分布式缓存需要序列化//如果没有复制,假设Changing后的数据会导致本地缓存和分布式缓存不一致finalObjectfinalValue=copy(value);//如果配置了写本地缓存,根据KEY获取相关的本地缓存,然后写入if(writeLocalCache){CachelocalCache=getLocalCache(key);if(localCache!=null){localCache.put(key,finalValue);}}//如果分布式缓存没有写入,直接返回if(!writeRemoteCache){return;}//异步更新分布式缓存asyncTaskExecutor.execute(()->{try{redisCache.set(key,JSONUtils.toJSON(finalValue),remoteCacheExpiresInSeconds);}catch(Exceptione){LOG.error("updaterediscacheerror,key:{}",key,e);}});}这里使用异步更新,让用户请求尽快返回。并且因为有本地缓存??,即使分布式缓存更新慢,出现回源,也可以把***缓存到本地。3、读缓存API包先读取本地缓存,如果本地缓存不完善,再批量查询分布式缓存,查询分布式缓存时通过分区批量查询。privateMapinnerMget(Listkeys,Listtypes)throwsException{Mapresult=Maps.newHashMap();ListmissKeys=Lists.newArrayList();ListmissTypes=Lists。newArrayList();//如果配置了读取本地缓存,则先读取本地缓存if(readLocalCache){for(inti=0;imissResult=Maps.newHashMap();//对于KEY分区,不要一次调用过多的batchfinalList>keysPage=Lists.partition(missKeys,10);List>>pageFutures=Lists.newArrayList();try{//批量获取分布式缓存数据for(finalListpartitionKeys:keysPage){pageFutures.add(asyncTaskExecutor.提交(()->redisCache.mget(partitionKeys)));}for(Future>future:pageFutures){missResult.putAll(future.get(3000,TimeUnit.MILLISECONDS));}}catch(Exceptione){pageFutures.forEach(future->future.cancel(true));throwe;}//合并result和missResult,此处省略returnresult;}这里对batchreadcache进行了分区,防止batch获取随意使用API2.NULLCache首先,定义NULL对象。privatestaticfinalStringNULL_STRING=newString();当DB没有数据时,将NULL对象写入缓存//查询DBStringvalue=loadDB();//如果DB没有数据,则封装为NULL_STRING放入缓存if(value==null){value=NULL_STRING;}myCache.put(id,value);读取数据时,如果发现NULL对象,则返回null,而不是返回DBvalue=suitCache.getIfPresent(id);//DB没有数据,返回nullif(value==NULL_STRING){returnnull;}这种方式,可以防止DB中不存在KEY对应的数据时频繁查询DB。3、强制获取***数据在实际应用中,我们经常需要强制更新数据。这个时候缓存的数据是不能使用的。可以配置ThreadLocal开关来决定是否强制刷新缓存(refresh方法要和CacheLoader一起使用)。if(ForceUpdater.isForceUpdateMyInfo()){myCache.refresh(skuId);}Stringresult=myCache.get(skuId);if(result==NULL_STRING){returnnull;}4.失败统计privateLoadingCachefailedCache=CacheBuilder.newBuilder().softValues().maximumSize(10000).build(newCacheLoader(){@OverridepublicAtomicIntegerload(StringskuId)throwsException{returnnewAtomicInteger(0);}});失败时,通过failedCache.getUnchecked(id).incrementAndGet()增加失败次数;成功后,使用failedCache.invalidate(id)使缓存无效。这样可以控制失败的重试次数,也是内存敏感的缓存。当内存不足时,您可以清除缓存以释放一些空间。5.延迟报警privatestaticLoadingCachealarmCache=CacheBuilder.newBuilder().softValues().maximumSize(10000).expireAfterAccess(1,TimeUnit.HOURS).build(newCacheLoader(){@OverridepublicIntegerload(Stringkey)throwsException{return0;}});//报警代码Integercount=0;if(redis!=null){StringcountStr=Objects.firstNonNull(redis.opsForValue().get(key),"0");count=Integer.valueOf(countStr);}else{count=alarmCache.get(key);}if(count%5==0){//每5次报一次//报警}countcount=count+1;if(redis!=null){redis.opsForValue().set(key,String.valueOf(count),1,TimeUnit.HOURS);}else{alarmCache.put(key,count);}如果有问题,报警存在大量的告警或误报。因此,可以认为该报警器在实际报警前已经长时间报警了M次。这时候也可以使用Cache来统计。此示例还添加了对Redis分布式缓存记录的支持。6.性能测试笔者使用JMH1.14进行基准性能测试,比如测试写作。@Benchmark@Warmup(iterations=10,time=10,timeUnit=TimeUnit.SECONDS)@Measurement(iterations=10,time=10,timeUnit=TimeUnit.SECONDS)@BenchmarkMode(Mode.Throughput)@OutputTimeUnit(TimeUnit.SECONDS)@Fork(1)publicvoidtest_1_Write(){counterWritercounterWriter=counterWriter+1;myCache.put("key"+counterWriter,"value"+counterWriter);}使用JMH时,先预热JVM,然后测量,生成测试结果(本文使用吞吐量)。建议读者根据自己的需求进行基准性能测试,选择适合自己的缓存框架。【本文为专栏作者张凯涛原创文章,作者微信公众号:凯涛博客(kaitao-1234567)】点此阅读更多本作者好文