应用级缓存示例
多级缓存api封装 我们的业务数据如商品类目、店铺、商品基本信息都可以进行适当的本地缓存,以提升性能。对于多实例的情况时不仅会使用本地缓存,还会使用分布式缓存,因此需要进行适当的api封装以简化缓存操作。 1.本地缓存初始化 p blic class localcacheinitservice extends baseservice { @override p blicvoid afterpropertiesset throws exception { 商品类目缓存 cachestring, object categorycache = cacheb ilder.newb ilder .softval es .maxim msize .expireafterwrite(switches.category.getexpiresinseconds 2, timeunit.seconds) .b ild； addcache(cachekeys.category_key, categorycache)； } privatevoid addcache(string key, cache？, ？ cache) { localcacheservice.addcache(key,cache)； } } 本地缓存过期时间使用分布式缓存过期时间的一半,防止本地缓存数据缓存时间太长造成多实例间的数据不一致。 另外,将缓存key前缀与本地缓存关联,从而匹配缓存key前缀就可以找到相关联的本地缓存。 2.写缓存api封装 先写本地缓存,如果需要写分布式缓存,则通过异步更新分布式缓存。 p blic void set(final string key, final object val e, final intremotecacheexpiresinseconds) throws r ntimeexception { if (val e== n ll) { ret rn； } 复制值对象 本地缓存是引用,分布式缓存需要序列化 如果不复制的话,则假设之后数据改了将造成本地缓存与分布式缓存不一致 final object finalval e = copy(val e)； 如果配置了写本地缓存,则根据key获得相关的本地缓存,然后写入 if (writelocalcache) { cache localcache = getlocalcache(key)； if(localcache ！= n ll) { localcache.p t(key, finalval e)； } } 如果配置了不写分布式缓存,则直接返回 if (！writeremotecache) { ret rn； } 异步更新分布式缓存 asynctaskexec tor.exec te( - { try { rediscache.set(key,jsonutils.tojson(finalval e), remotecacheexpiresinseconds)； } catch(exception e) { log.error(" pdateredis cache error, key : {}", key, e)； } })； } 此处使用了异步更新,目的是让用户请求尽快返回。而因为有本地缓存,所以即使分布式缓存更新比较慢又产生了回源,也可以在本地缓存命中。 读缓存api封装 先读本地缓存,本地缓存不命中的再批量查询分布式缓存,在查询分布式缓存时通过分区批量查询。 private map innermget(liststring keys, listclass types) throwsexception { mapstring, object res lt = maps.newhashmap； liststring misskeys = lists.newarraylist； listclass misstypes = lists.newarraylist； 如果配置了读本地缓存,则先读本地缓存 if(readlocalcache) { for(int i = 0； i  keys.size； i) { string key = keys.get(i)； class type = types.get(i)； cache localcache = getlocalcache(key)； if(localcache ！= n ll) { object val e = localcache.getifpresent(key)； res lt.p t(key, val e)； if (val e == n ll) { misskeys.add(key)； misstypes.add(type)； } } else { misskeys.add(key)； misstypes.add(type)； } } } 如果配置了不读分布式缓存,则返回 if(！readremotecache) { ret rnres lt； } finalmapstring, string missres lt = maps.newhashmap； 对key分区,不要一次性批量调用太大 final listliststringkeyspage = lists.partition(misskeys, 10)； listf t remapstring, string pagef t res = lists.newarraylist； try { 批量获取分布式缓存数据 for(final liststringpartitionkeys : keyspage) { pagef t res.add(asynctaskexec tor.s bmit( - rediscache.mget(partitionkeys； } for(f t remapstring,string f t re : pagef t res) { missres lt.p tall(f t re.get(3000, timeunit.milliseconds))； } } catch(exception e) { pagef t res.foreach(f t re - f t re.cancel(tr e))； throw e； } 合并res lt和missres lt,此处实现省略 ret rn res lt； } 此处将批量读缓存进行了分区,防止乱用批量获取api。 首先,定义null对象。 private static final string null_string =new string； 当db没有数据时,写入null对象到缓存 查询db string val e = loaddb； 如果db没有数据,则将其封装为null_string并放入缓存 if(val e == n ll) { val e = null_string； } mycache.p t(id, val e)； 读取数据时,如果发现null对象,则返回n ll,而不是回源到db val e = s itcache.getifpresent(id)； db没有数据,返回n ll if(val e == null_string) { ret rn n ll； } 通过这种方式可以防止当key对应的数据在db不存在时频繁查询db的情况。 强制获取最新数据 在实际应用中,我们经常需要强制更新数据,此时就不能使用缓存数据了,可以通过配置threadlocal开关来决定是否强制刷新缓存(refresh方法要配合cacheloader一起使用)。 if(forceupdater.isforceupdatemyinfo) { mycache.refresh(sk id)； } string res lt = mycache.get(sk id)； if(res lt == null_string) { ret rn n ll； } 失败统计 private loadingcachestring, atomicinteger failedcache = cacheb ilder.newb ilder .softval es .maxim msize .b ild(new cacheloaderstring, atomicinteger { @override p blic atomicintegerload(string sk id) throws exception { ret rn new atomicinteger(0)； } })； 当失败时,通过failedcache.getunchecked(id).incrementandget增加失败次数；当成功时,使用failedcache.invalidate(id)失效缓存。通过这种方式可以控制失败重试次数,而且又是内存敏感缓存。当内存不足时,可以清理该缓存腾出一些空间。 延迟报警 private static loadingcachestring, integer alarmcache = cacheb ilder.newb ilder .softval es .maxim msize.expireafteraccess(1, timeunit.hours) .b ild(new cacheloaderstring, integer { @override p blic integer load(string key) throws exception { ret rn 0； } })； 报警代码 integer co nt = 0； if(redis ！= n ll) { stringco ntstr = objects.firstnonn ll(redis.opsforval e.get(key), "0")； co nt =integer.val eof(co ntstr)； } else { co nt = alarmcache.get(key)； } if(co nt % 5 == 0) { 5次报一次 报警 } co nt = co nt + 1； if(redis ！= n ll) { redis.opsforval e.set(key,string.val eof(co nt), 1, timeunit. hours)； } else { alarmcache.p t(key,co nt)； } 如果一出问题就报警,则存在报警量非常多或者假报警,因此,可以考虑n久报警了m次,才真正报警。此时,也可以使用cache来统计。本示例还加入了redis分布式缓存记录支持。 笔者使用jmh 1.14进行基准性能测试,比如测试写。 @benchmark @warm p(iterations = 10, time = 10, timeunit =timeunit.seconds) @meas rement(iterations = 10, time = 10, timeunit= timeunit.seconds) @benchmarkmode(mode.thro ghp t) @o tp ttimeunit(timeunit.seconds) @fork(1) p blic void test_1_write { co nterwriter= co nterwriter + 1； mycache.p t("key"+ co nterwriter, "val e" + co nterwriter)； } 使用jmh时首先进行jvm预热,然后进行度量,产生测试结果(本文使用吞吐量)。建议读者按照需求进行基准性能测试来选择适合自己的缓存框架。