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

京东云开发者
• 阅读 81

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

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广播或者指定机器生效配置。如果是定时生效,则新增定时器并指定生效规则,达到时间后触发广播通知。

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

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

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

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

 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为实例变量名。

6 总结

1、支持jdk动态代理的实例对象和cglib代理对象的参数动态配置

2、支持定时刷新配置

3、直接查看和验证应用集群中实例变量是否一致

点赞
收藏
评论区
推荐文章
blmius blmius
1年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
Wesley13 Wesley13
1年前
FLV文件格式
1.        FLV文件对齐方式FLV文件以大端对齐方式存放多字节整型。如存放数字无符号16位的数字300(0x012C),那么在FLV文件中存放的顺序是:|0x01|0x2C|。如果是无符号32位数字300(0x0000012C),那么在FLV文件中的存放顺序是:|0x00|0x00|0x00|0x01|0x2C。2.  
Easter79 Easter79
1年前
Twitter的分布式自增ID算法snowflake (Java版)
概述分布式系统中,有一些需要使用全局唯一ID的场景,这种时候为了防止ID冲突可以使用36位的UUID,但是UUID有一些缺点,首先他相对比较长,另外UUID一般是无序的。有些时候我们希望能使用一种简单一些的ID,并且希望ID能够按照时间有序生成。而twitter的snowflake解决了这种需求,最初Twitter把存储系统从MySQL迁移
Stella981 Stella981
1年前
ELK学习笔记之配置logstash消费kafka多个topic并分别生成索引
0x00 filebeat配置多个topicfilebeat.prospectors:input_type:logencoding:GB2312fields_under_root:truefields:添加字段
Easter79 Easter79
1年前
SpringBoot整合Redis乱码原因及解决方案
问题描述:springboot使用springdataredis存储数据时乱码rediskey/value出现\\xAC\\xED\\x00\\x05t\\x00\\x05问题分析:查看RedisTemplate类!(https://oscimg.oschina.net/oscnet/0a85565fa
Wesley13 Wesley13
1年前
Java日期时间API系列36
  十二时辰,古代劳动人民把一昼夜划分成十二个时段,每一个时段叫一个时辰。二十四小时和十二时辰对照表:时辰时间24时制子时深夜11:00凌晨01:0023:0001:00丑时上午01:00上午03:0001:0003:00寅时上午03:00上午0
Wesley13 Wesley13
1年前
PHP中的NOW()函数
是否有一个PHP函数以与MySQL函数NOW()相同的格式返回日期和时间?我知道如何使用date()做到这一点,但是我问是否有一个仅用于此的函数。例如,返回:2009120100:00:001楼使用此功能:functiongetDatetimeNow(){
Stella981 Stella981
1年前
Jenkins 插件开发之旅:两天内从 idea 到发布(上篇)
本文首发于:Jenkins中文社区(https://www.oschina.net/action/GoToLink?urlhttp%3A%2F%2Fjenkinszh.cn)!huashan(https://oscimg.oschina.net/oscnet/f499d5b4f76f20cf0bce2a00af236d10265.jpg)
Wesley13 Wesley13
1年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
helloworld_34035044 helloworld_34035044
6个月前
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为