一种面向业务配置基于JSF广播定时生效的工具

技术分享 2年前 (2023-01-21) 0 999+

关注

作者：京东物流王北永姚再毅李振

1 背景

目前，ducc实现了实时近乎所有配置动态生效的场景，但是配置是否实时生效，不能直观展示每个机器上jvm内对象对应的参数是否已变更为准确的值，大部分时候需要查看日志确认是否生效。

2 技术依赖

1)Jsf:京东RPC框架，用作机器之间的通讯工具

2)redis/redisson：redis，用作配置信息的存储

3)ZK/Curator: Zookeeper，用作配置信息的存储和redis二选一

3)clover：定时任务集群，用作任务延迟或周期性执行

3 实现原理

1）接入方：

各个接入系统通过接入管理模块获取token，并指定所在系统发布的的服务器ip,用作后续的ip鉴权。当系统启动时，自动在各个系统生成接口提供方，并注册到JSF注册中心。别名需各个系统唯一不重复。鉴权为统一服务端做IP鉴权。

2）统一配置服务端：

提供按不同接入方、不同系统、不同环境的配置界面。业务人员可设定自动生效时间或者立即生效时间。如果是立刻生效，则通过JSF广播或者指定机器生效配置。如果是定时生效，则新增定时器并指定生效规则，达到时间后触发广播通知。

整个接入方和统一配置服务端的架构如下图

4 实现步骤

1）重写JSF类ConsumerConfig类方法refer，将其中的轮训调用客户端改为广播调用客户端BroadCastClient。

this.client= new BroadCastClient(this); this.proxyInvoker = new ClientProxyInvoker(this); ProtocolFactory.check(Constants.ProtocolType.valueOf(this.getProtocol()), Constants.CodecType.valueOf(this.getSerialization())); this.proxyIns = (T) ProxyFactory.buildProxy(this.getProxy(), this.getProxyClass(), this.proxyInvoker);

2)广播调用客户端方法分别获取当前注册中心有心跳的服务提供者和已失去连接的机器列表。对统一配置来讲，要么同时失败，要么同时成功，判断如果存在不正常的服务提供方，则不同步。只有全部提供方存在才可以开始广播配置信息

 ConcurrentHashMap<Provider, ClientTransport>  concurrentHashMap = this.connectionHolder.getAliveConnections();         ConcurrentHashMap<Provider, ClientTransportConfig>  deadConcurrentHashMap = this.connectionHolder.getDeadConnections();         if(deadConcurrentHashMap!=null && deadConcurrentHashMap.size()>0){             log.warn("当前别名{}存在不正常服务提供方数量{}，请关注！",msg.getAlias(),deadConcurrentHashMap.size());             throw new RpcException(String.format("当前别名%s存在不正常服务提供方数量%s，请关注！",msg.getAlias(),deadConcurrentHashMap.size()));         }         if(concurrentHashMap.isEmpty()){             log.info("当前别名{}不存在正常服务提供方",msg.getAlias());             throw new RpcException(String.format("当前别名%s不存在正常服务提供方",msg.getAlias()));         }         Iterator aliveConnections = concurrentHashMap.entrySet().iterator();         log.info("当前别名{}存在正常服务提供方数量{}",msg.getAlias(),concurrentHashMap.size());         while (aliveConnections.hasNext()) {             Entry<Provider, ClientTransport> entry = (Entry) aliveConnections.next();             Provider provider = (Provider) entry.getKey();             log.info("当前连接ip={}、port={}、datacenterCode={}",provider.getIp(),provider.getPort(),provider.getDatacenterCode());             ClientTransport connection = (ClientTransport) entry.getValue();             if (connection != null && connection.isOpen()) {                 try {                     result = super.sendMsg0(new Connection(provider, connection), msg);                 } catch (RpcException rpc) {                     exception = rpc;                     log.warn(rpc.getMessage(), rpc);                 } catch (Throwable e) {                     exception = new RpcException(e.getMessage(), e);                     log.warn(e.getMessage(), e);                 }             }         }

3)服务配置端，当业务人员配置及时生效或者任务达到时，则根据配置，生成服务调用方，通过统一刷新接口将配置同步刷新到对应的接入系统中,如下图为操作界面，当增删改查时，会将属性增量同步。

服务端在上面操作增删改时，通过以下方式获取服务调用方

 public static ExcuteAction createJsfConsumer(String alias, String token) {         RegistryConfig jsfRegistry = new RegistryConfig();         jsfRegistry.setIndex("i.jsf.jd.com");         BroadCastConsumerConfig consumerConfig = new BroadCastConsumerConfig<>();         Map<String, String> parameters = new HashMap<>();         parameters.put(".token",token);         consumerConfig.setParameters(parameters);         consumerConfig.setInterfaceId(RefreshRemoteService.class.getName());         consumerConfig.setRegistry(jsfRegistry);         consumerConfig.setProtocol("jsf");         consumerConfig.setAlias(alias);         consumerConfig.setRetries(2);         return  new ExcuteAction(consumerConfig);     }

通过以上的配置的客户端，调用服务提供方方法refreshRemoteService#refresh，将配置信息进行同步到各个接入系统

public void call(Map<String,Object> propertiesValue){         try{             RefreshRemoteService refreshRemoteService = (RefreshRemoteService)consumerConfig.refer();             if(refreshRemoteService!=null){                 refreshRemoteService.refresh(propertiesValue);             }         }catch (Exception e){             log.error(e.getMessage());             throw new EasyConfigException(e);         }finally {             consumerConfig.unrefer(); ;         }     }

4）接入方启动时，需要根据自己配置，将存在redis或者zk的配置一次加载到实例变量中。并注册刷新接口到JSF注册中心。

其中注册刷新接口到JSF注册中心代码如下

 @Bean(name = "refreshPorpertiesService")     public ProviderConfig createJsfProvider() throws Exception {         ProviderConfig providerConfig = new ProviderConfig();         providerConfig.setId("refreshPorpertiesService");         providerConfig.setInterfaceId(RefreshRemoteService.class.getName());         providerConfig.setRef(new RefreshRemoteServiceDelage(applicationContext));         providerConfig.setTimeout(30000);         providerConfig.setAlias(EasyConfigure.getAppCode()+EasyConfigure.getEnv());         providerConfig.setServer(serverConfig);         providerConfig.setRegistry(jsfRegistry);         providerConfig.setParameter("token", MD5Util.md5(EasyConfigure.getAppCode()));         providerConfig.export();         return providerConfig;     }

其中
RefreshRemoteServiceDelage类提供刷新接口的实际逻辑如下，需判断当前实例是jdk动态代理还是cglib代理

判断逻辑如下

 if(AopUtils.isJdkDynamicProxy(object)) {          object= AopUtil.getJdkDynamicProxyTargetObject(object);  } else if(AopUtils.isCglibProxy(object)){ //cglib          object= AopUtil.getCglibProxyTargetObject(object);    }

实例对象变量值根据自定义的参数转换方式转换后赋值实例变量

if(autoValue.convert()!=null && !autoValue.getClass().isInterface()){          if(!autoValue.convert().newInstance().getInClassType().isAssignableFrom(newVal.getClass()) ){                     continue;              }            newVal = autoValue.convert().newInstance().convert(newVal);             if(newVal!=null){                   if(!autoValue.convert().newInstance().getOutClassType().isAssignableFrom(newVal.getClass()) ){                             continue;                       }               field.setAccessible(true);                Object value = ReflectionUtils.getField(field,object);               log.info("change  properties{} for object {} before value {}",field.getName(),object.getClass().getName(),value);                ReflectionUtils.setField(field,object,newVal);                 log.info("change  properties{} for object {} after value {}",field.getName(),object.getClass().getName(),newVal);          }               }

5 实践

1)pom引入

<dependency>    <groupId>com.jdl</groupId>    <artifactId>easyconfig</artifactId>    <version>1.0-SNAPSHOT</version> </dependency>

2)配置存储配置(比如redis方式)

refresh:   config:     appCode: zdzq-worker-appcode     redisUrl: redis://:@127.0.0.1

3）类全局变量需要实时刷新配置，需在类统一指定注解PropertiryChangeListener，实例变量需要增加注解AutoValue并指定数据格式转换器

@PropertiryChangeListener public class ChanceServiceImpl implement ChanceService{ @AutoValue(convert = DateConvert.class,alias = "config-id") private Date  signDate; @AutoValue(convert = SpmKaApply Convert.class,alias = "config-id") private SpmKaApply  spmKaApply; }

以上convert方法自定义，支持各种复杂配置对象，举例数据转换为List如下

public  class Convert2 implements Convert<Map<String, String>, Set<String>> {     public Convert2(){}     @Override     public Set<String> convert(Map<String, String> siteInfoMap) {         ....你的对象值转换     }     @Override     public Class<?> getInClassType() {         return Map.class;     }     @Override     public Class<?> getOutClassType() {         return Set.class;     }

接入应用服务启动后，可访问/refreshUI 可查看应用在集群中为自动配置的实例，并显示当前实例中变量值参数。key为实例变量名。