集群系统下多级缓存实时更新系统设计

总篇197篇 2023年第10篇1.前言在C端项目实践中，对于接口性能的要求是重中之重。为了提升接口TP99等 各项性能指标，在项目中大量使用分布式缓存和本地应用缓存提升接口响应速 度,缓存的应用必然要面对缓存更新的问题，而目前缓存更新方案主要有几种：通过启用定时任务定时刷新缓存，会存在时效性不高，性能浪费的缺点；启用缓存过期策略更新缓存，同样时效性不高，缓存更新不及时，缓存过期降级后降低接口性能；监听数据库变动主动通知缓存更新，需要监听上游数据库更新，跨业务情况下需要多方配合才能实现；本系统采用第三种更新方式，设计整套系统方案，对于自有维护数据能够方便实现毫秒级主动更新，能够实现更好的提高缓存更新时效性。2.系统架构2.1整体设计借助之家DTS实时数据流服务将数据库的数据变更日志数据实时发送到Kafka，如MySQL的binlog日志，缓存中心订阅kafka消息监听数据库变动；搭建缓存调度中心，负责接收缓存更新执行器注册，缓存更新调度，缓存key血缘依赖关系维护；通过单点调度方式调度业务系统完成分布式缓存更新，同时支持调度故障转移保证调度成功率；通过广播调度方式调度业务系统完成集群的本地缓存更新；提供缓存更新SDK，注解非侵入方式简单易用；缓存中心技术实现上参考了开源项目XXL-JOB，对其做深度二次开发，快速实现功能开发上线；2.2模块设计?1. 缓存执行器及更新处理单元注册流程说明：业务应用启动时，向缓存中心注册执行器和缓存key及key的依赖血缘关系，以及更新时的调度方式信息；缓存中心收到注册请求后判断执行器是否已经创建，没有创建则创建执行器并记录执行器IP地址，已经创建则只更新执行器IP地址；并存储缓存目标key和源key；（目标key：源key 为一对多关系）；注册逻辑实现：//执行器注册public?class?CacheSpringExecutor?extends?CacheExecutor?implementsSmartInitializingSingleton{?private?List?cacheBloodParams?=?null;?@Overridepublic?void?afterSingletonsInstantiated()?{//向spring容器中注册缓存更新handlerinitCacheUpdateHandlerRepository(applicationContext);try?{if(!CollectionUtils.isEmpty(cacheBloodParams)){//向远程缓存中心注册缓存key和key依赖血缘关系等信息super.start(cacheBloodParams);}}?catch?(Exception?e)?{throw?new?RuntimeException(e);}}private?void?initCacheUpdateHandlerRepository(ApplicationContext?applicationContext)?{?String[]?beanDefinitionNames?=?applicationContext.getBeanNamesForType(Object.class,?false,?true);?cacheBloodParams?=?new?ArrayList();?for?(String?beanDefinitionName?:?beanDefinitionNames)?{Object?bean?=?applicationContext.getBean(beanDefinitionName);?//获取所有标注了CacheUpdate注解的方法?Map?annotatedMethods?=?MethodIntrospector.selectMethods(bean.getClass(),(MethodIntrospector.MetadataLookup)?method?->?AnnotatedElementUtils.findMergedAnnotation(method,?CacheUpdate.class));if?(annotatedMethods?==?null?||?annotatedMethods.isEmpty())?{continue;}annotatedMethods.forEach((k,?v)?->?{if(!check(v)){throw?new?RuntimeException("cache-center?method-cacheHandler?param?invalid,?for["?+?bean.getClass()?+?"#"?+?"]?.");?}//初始化key依赖血缘关系参数对象CacheBloodParam?param?=?initCacheBloodParam(v);cacheBloodParams.add(param);k.setAccessible(true);//将处理缓存方法包装成MethodCacheHandler对象注册到本地handlerRepository容器中registHandler(v.targetKey(),?new?MethodCacheHandler(bean,?k,?null,?null));});}}}?2.2.2 更新消息消费及调度流程说明：缓存中心监听kafka 数据变更 消息（如mysql的binlog消息），从数据变更消息中提取发生变化数据的主键、变化类型（新增、修改、删除）信息，封装成数据变更对象存入调度队列；缓存中心调度线程异步消费调度队列信息，并根据缓存key血缘关系,通过source_key获取target_key对应执行器ip地址，根据业务服务注册的路由规则进行调度分发；业务系统接收调度请求，将变更信息放入处理队列，根据变更目标缓存key本地获取处理handler异步进行更新处理；变更消息体：{"sourceKey":"test_db.test_table",?//源key?"bizIds":["1","2"],?//源业务id?"operateType":"1"?//操作类型?1-add?2-update?3-delete}调度分发实现：public?class?CacheTrigger{/**?**?@param?targetKeyId?目标key?id?*?@param?changeInfo数据变更信息?*/public?static?void?trigger(Long?targetKeyId,CacheChangeInfo?changeInfo)?{//获取目标keyCacheTargetInfo?info?=?CacheAdminConfig.getAdminConfig().getCacheTargetInfoDao().loadById(targetKeyId);if?(null?==?info)?{return;}//获取执行器信息CacheGroup?group?=?CacheAdminConfig.getAdminConfig().getXxlJobGroupDao().loadByAppName(info.getAppName());if(null==group){logger.error("执行器不存在，调度结束，changeInfo:{}",?GsonTool.toJson(changeInfo));return;}//执行器ip列表String?addressList?=?group.getAddressList();if?(addressList?!=?null?&?addressList.trim().length()?>?0)?{group.setAddressType(1);group.setAddressList(addressList.trim());}//匹配路由策略ExecutorRouteStrategyEnum?routeStrategy?=?ExecutorRouteStrategyEnum.match(info.getRouteStrategy());if(ExecutorRouteStrategyEnum.SHARDING_BROADCAST.equals(routeStrategy)){//广播group.getRegistryList().forEach(adress->?processTrigger(info,?changeInfo,?adress));}else?{TriggerParam?triggerParam?=?new?TriggerParam();triggerParam.setId(info.getId().intValue());//其它路由方式,获取执行器ip地址ReturnT?routeRes?=?routeStrategy.getRouter().route(triggerParam,?group.getRegistryList());if?(routeRes.getCode()?==?ReturnT.SUCCESS_CODE)?{Stringaddress?=?routeRes.getContent();//调度业务服务processTrigger(info,changeInfo,address);}else?{logger.error("路由地址解析错误，msg:{}",routeRes.getMsg());}}}}2.2.3 缓存更新SDK使用项目中引入SDK包依赖项目配置文件中配置缓存中心地址业务开发人员编写更新方法，方法上添加相应注解demo:public?class?Demo{@CacheUpdate(targetKey?=?"ne:test:cache",?targetKeyDesc?=?"测试",?sourceKey?=?{"aa.test_1",?"aa.test_2"},roteType?=?RouteStrategyEnum.SHARDING_BROADCAST)public?void?cacheUpdate()?{CacheChangeParam?changeParam?=?XxlCacheHelper.getChangeParam();List?bizIds?=?changeParam.getBizIds();?//更新的业务idInteger?operateType?=?changeParam.getOperateType();?//1-add?2-update?3-deletelogger.info(GsonTool.toJson(changeParam));logger.info("测试");}}注解@CacheUpdate 说明:字段说明targetKey目标key，需要更新的缓存keytargetKeyDesc目标key描述，非必填（建议填写，方便缓存中心管理台查看）sourceKey目标key依赖的源key，源key更新会调度目标key,如果源为数据库,key：databaseName.tableNamerouteType调度使用策略，非必填，默认 为故障转移，需要所有实例都更新采用广播策略（本地缓存更新使用）3. 应用效果在之家选车新能源创新工具，VR选配器开发中的应用:3.1业务逻辑?业务人员在VR选配后台录入车型配件信息；后台提供给H5页面组件列表信息，接口数据做本地缓存，提升接口响应；选配器API微服务引入缓存更新SDK,编写缓存更新方法;业务人员后台更改配置，数据库数据发生变动，API缓存得到实时进行更新；3.2时效性测试数据库更新：业务应用日志：测试结果：数据库变动更新同步到缓存更新完成耗时约900ms4.总结通过建立独立的缓存中心把数据的变动发现，通知更新独立为单独的底层基础服务，由原来的低效的定时更新或者失效降级更新变成主动更新，缓存更新的时效性由天级，小时级变为秒级，极大的提升了业务系统的运行效率。通过提供简单易用低耦合的缓存更新SDK工具，也同时提高了业务开发同学的整体开发效率。作者简介?甄成国?服务端看选技术团队-基础架构组。C端及中台产研中心-服务端研发部-服务端看选技术团队-基础架构组。2021年加入汽车之家。负责车家号后端服务开发。阅读更多：大数据 + VR 全景技术重塑“二手车买车场景”服务模块化从SPserver到BRPC