distributeTemplate分布式配置以及上下文的同步维护是怎样的

这篇文章将为大家详细讲解有关distributeTemplate分布式配置以及上下文的同步维护是怎样的，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。

我们实现分布式 也是按照 这样一步一步来的，首先是公用的资源对象，第一个必须的公用资源 对象是配置，这配置交给用户外部填写 手动配置 并能维持同步 更新 所以需要一个配置对象 维护在内存里面，需要事件驱动监听配置文件的变化情况

ok下面来 看看代码是怎么做的 ，首先 配置 有多重配置方式 ini conf prop xml 各种方式本质上 我们分布式就是要各台主机 自己所在的节点 都能知道我在网络上情况，或者所可以找到像zookeeper 只要知道或者找到了 才能进行以后的通讯同步

我们为了能够以后支持多重配置 ，所以先把配置定义为一个接口

public interface IConfig {     public static final String DEFAULTFILE="distribute.conf"; /**  *   * 从文件读取配置初始化  * @param file  * 添加（修改）人：zhuyuping  */ public void readConfigFormFile(String file); /**  * 从用户提供的json字符串中初始化  * @param json  * 添加（修改）人：zhuyuping  */ public void readConfigFormString(String json); /**  * 获得系统内存中维护的内存表  *   * 添加（修改）人：zhuyuping  */ public Table<InetSocketAddress, String, Object> getConfig(); /**  * 摄入context上下文  * @param context  * 添加（修改）人：zhuyuping  */ public void setContext(Context context); /**  * 获得上下文context  * @return  * 添加（修改）人：zhuyuping  */ public Context getContext(); public BiMap<String, InetSocketAddress> getAlias(); }

主要 是3个方法 ，一个是读配置文件到内存对象中，拧一个就是内存对象写到文件中，对于分布式还需要一个同步

有时候 我们可能需要历史版本 能够容错还原  所以可能需要一个版本号 记录当前版本，然后保持有可能多个节点上配置文件更改后，发送请求时候能够保持顺序的更新，这后面我会处理  加入一个分布式先进先出的队列的，这里暂时未加 后面 加了会补上的，还有上下文context 对象，有人会说  你为什么需要上下文对象去保存变量啊，好多框架 都有上下文这个对象，也许是拦截器模式 也许是门面模式 等等其他模式，但是  这里我都不是，其实上下文只是为了本地节点的各个功能代码段之间的一个桥梁，有句话叫做 两情若是久长时 又岂在朝朝暮暮  我们有鹊桥，他就是上下文最重要的本质 就是维护本地节点上下文贯穿 如果上下文 里面保存着门面 那么他就有门面的功能 便于用户随时获取门面对象  进行操作，ok我们来看看 context是怎么定义的

/**  *   *        *       * @author zhuyuping         * @version 1.0       * @created 2014-7-9   * @function:上下文接口  他只是存储用户上面类所有过程中的变量不是config配置而且分布式中不会同步的只会在单点上有效 切记  、、后期如果想支持xml 配置或者其他配置 可以添加策略模式   */ public interface Context {       public final static String defaultConfig="distribute.conf";//默认配置名           public void putValue(String key,Object value);           public Object getValue(String key);           public void setCurrHost(String host,int port);           public InetSocketAddress getCurrHost();     /**      * 获得默认配置文件      * @return      * 添加（修改）人：zhuyuping      */     public String getDefaultFc(); //  /** //   * 设置默认属性文件的名称 //   * @param pfile //   * 添加（修改）人：zhuyuping //   */ //  public void setDefaultFc(String pfile); //  /** //   * 注入template 门面 便于后面直接通过上下文来使用  如果要整合spring ApplicationContextAware //   * @param distributedTemplate //   * 添加（修改）人：zhuyuping //   */ //  public void setTemplate(DistributedOperations distributedTemplate); //  //  public DistributedOperations getTemplate();             }

这里其实 就是一个map 保存属性key value 而常用的就取出作为一个方法了

这个context 因为后面我们给用户一个继承的 这样便于 用户实现自己的上下文 或交给其他框架上下文 以及整合所以我们实现了一个抽象的 默认实现

/**  *   *        *       * @author zhuyuping         * @version 1.0       * @created 2014-7-9 下午5:58:37   * @function:抽象的上下文 主要是 管理context的资源 还有就是提供用户自定义 整合spring使用该类 //这里后期需整合策略 实现  */ public abstract class AbstractContext implements Context{     //?也可以使用LocalThread 也可以     private Map<String,Object> context=Maps.newConcurrentMap();           private InetSocketAddress currHost;//当前主机 比如192.168.0.1 8888           private String dfConfig;//默认读取的配置文件 当用户 提供就修改 没有提供就默认                  public AbstractContext(String dfConfig) {         super();                   this.dfConfig = dfConfig;         //当前classes 下的文件         //currentPort         this.currHost=new InetSocketAddress(ConfigFactory.load(dfConfig).getString("client.currentHost"),ConfigFactory.load(dfConfig).getInt("client.currentPort"));     }                   @Override     public InetSocketAddress getCurrHost() {                   return currHost;     }               @Override     public void setCurrHost(String host,int port) {                   this.currHost=new InetSocketAddress(host, port);               }           @Override     public String getDefaultFc() {                   return dfConfig!=null?dfConfig:defaultConfig;     }           public AbstractContext() {         super();         this.dfConfig=defaultConfig;           this.currHost=new InetSocketAddress(ConfigFactory.load(defaultConfig).getString("client.currentHost"),ConfigFactory.load(defaultConfig).getInt("client.currentPort"));               }           @Override     public void putValue(String key, Object value) {                   context.put(key, value);               }       @Override     public Object getValue(String key) {         return context.get(key);     }                   }

ok 很简单的 维护者

然后回到刚才的配置 ，首先 我们配置文件 需要从文件读取 到配置对象中 ，这是为了用户更改 或者初始化时候 吧配置文件初始化到配置内存对象中 然后这个内存对象将会在同步 配置文件 更改 网络通讯时候用到，在对于全局的所有节点的沦陷时候 单纯的context只维护本节点桥梁信息的 已经不够用了 因为他不会同步的，这就是为什么需要他的原因，我这里采用的配置文件时conf 也就是configLoad方式，后面我会逐步添加更多的支持方式 无非是xml 等读取问题，这并不重要 思想才是重要的

/**  *   *        *       * @author zhuyuping         * @version 1.0       * @created 2014-7-9 下午4:07:00  如果后期需要支持xml 其他 只需要使用策略模式   * @function:基类config 主要维持 配置基本信息 以及配置信息的同步 备份 同时内存中维持一张内存表table  */ public abstract class AbstractConfig implements IConfig{     /**      * 当前的配置表 行 为主机别名 其中一定有一列为版本号 AotmicLong 以及配置的相关字段  值为相关的对象        */ protected Table<InetSocketAddress, String, Object> config=HashBasedTable.create();//table信息  table 信息 这里无需要用到 哪个ConcurrentHashMap<K, V> 因为这个只会加载读取 加载  //不会修改，因为这个table当用户真的更新config后 会同步并同时刷到更新到文件中的 ，而且每次用户提供查询的配置 //是不会更新到文件里面的  protected AtomicLong version=new AtomicLong(0);//初始化版本为0;//判断当前的版本 是佛在config 里面是否存在 protected BiMap<String,InetSocketAddress> alias=HashBiMap.create(); protected Context context;            public BiMap<String, InetSocketAddress> getAlias() { return alias; } /**      * context 需要提供当前主机 以及       * @param context      */ public AbstractConfig(Context context) { super(); this.context = context; wrapConfig(ConfigFactory.load(context.getDefaultFc())); } @Override public void setContext(Context context) { this.context=context; } @Override public Context getContext() { return context; } @Override public void readConfigFormFile(String file) { Config config=TypeSafeConfigLoadUtils.loadFromFile(new File(file)); wrapConfig(config); } @Override public void readConfigFormString(String json) { Config config=TypeSafeConfigLoadUtils.loadFromString(json); wrapConfig(config); } /**  * 对config进行初始化 table  * @param config  * 添加（修改）人：zhuyuping  */ protected abstract void wrapConfig(Config config); /**  * 把table 从内存中读取从新写入到配置文件中   * @param config  * 添加（修改）人：zhuyuping  */ protected abstract String wrapTable(Table<String, String, Object> config); /**  * 只保留最近的5个版本 可以回滚 更新***的  *   * 添加（修改）人：zhuyuping  */ public void updateVersion(Long version){    } /**  * 版本数更新  *  更新完后 需要   * 添加（修改）人：zhuyuping  */ public void addVersion(){ Long v=version.getAndIncrement(); //TODO 需要通知所有节点 我要修改版本了 如果同时有几个人也这样 那么接受该节点下次更新的版本号，     //在回调函数中 更新配置 随后同步 只保留5个版本  }     @Override public Table<InetSocketAddress, String, Object> getConfig() { return config; } /**      *       * 提交对文件配置做出的修改      * 添加（修改）人：zhuyuping      */     protected abstract void commit(); /**  *   * 提交对配置的修改 如果一个人在一个节点上 更改了配置 需要核对版本 并从新更新本地的配置文件  * 添加（修改）人：zhuyuping  */ protected abstract void sync();       }

这里 我为了好维护 就直接使用guava的table ，其实你可以用List< map> 实现，这里 重要是获取所有主机列表的方法

然后就是配置文件读取后 写入context 对象 当然 上面说的读取配置到内存对象 ，内存对象写入到配置文件是基础

然后看看怎么写入的 我为了以后支持xml 所以这读取方式 写入方式 交给后面的子类实现类去实现 这样实现xml只要实现这2个方法即可

/**  *   *        *       * @author zhuyuping         * @version 1.0       * @created 2014-7-9 下午10:38:05   * @function:默认的配置  允许用户实现自定义的配置规则 只需要继承 AbstractConfig  */ public class SimpleDistributeConfig extends AbstractConfig{     public SimpleDistributeConfig(Context context) {         super(context);               }       @Override     protected void wrapConfig(Config configObj) {         //得到所有的节点         List<? extends ConfigObject> nodes=configObj.getObjectList("server.hosts");         int i=0;         for (ConfigObject node : nodes) {                  i++;            //如果后期添加其他mysql 等支持 这里就需要添加判断                        //Integer localport=node.containsKey("localPort")?Integer.parseInt(node.get("localPort").render()):LOCALPORT;//Integer.parseInt(node.get("localPort").render());            Integer remotePort=Integer.parseInt(node.get("remotePort").render());            String remoteIp=node.get("remoteHost").unwrapped().toString();//远程主机的ip            //开始初始化配置table            String name=node.containsKey("name")?node.get("name").unwrapped().toString():remoteIp;//主机别名            InetSocketAddress remoteHost=new InetSocketAddress(remoteIp, remotePort);            super.alias.put(name, remoteHost);            super.config.put(remoteHost,"version", super.version.incrementAndGet());            super.config.put(remoteHost, "remoteHost", remoteHost);            //super.config.put(remoteHost, "localPort", localport);            super.config.put(remoteHost, "remotePort", remotePort);            super.config.put(remoteHost, "name", name);                                    if(node.containsKey("file")){                HashMap fcs=(HashMap) node.get("file").unwrapped();                String syncPath=fcs.get("syncPath").toString();//文件同步的路径               // System.out.println("SimpleDistributeConfig.wrapConfig() "+syncPath);                super.config.put(remoteHost, "file", syncPath);//以后配置多的时候 分装成一个bean存入            }                                   }         String chost=configObj.getString("client.currentHost");         int port=configObj.getInt("client.currentPort");         super.context.setCurrHost(chost, port);                   //config.root().containsKey(key)         //config.getString(path);     }                 @Override     protected String wrapTable(Table<String, String, Object> table) {         StringBuilder sb=new StringBuilder("server{");                   sb.append("\n\t").append("hosts=[");         Set<String> rows=table.rowKeySet();         int size=rows.size();         int i=0;         for (String row : rows) {             i++;             Map<String,Object> map=table.row(row);             if(!map.containsKey("remoteHost")) continue;             sb.append("\t {");             Object remoteHost=map.get("remoteHost");               Object remotePort=map.get("remotePort");             //Object localPort=map.get("localPort");             Object name=map.get("name");             sb.append("name=").append(name).append("\n\t");             //sb.append("localPort=").append(localPort).append("\n\t");             sb.append("remoteHost=").append(remoteHost).append("\n\t");             sb.append("remotePort=").append(remotePort).append("\n\t");             if(map.containsKey("file")){                 sb.append("file{").append("\n\t").append("syncPath=").append(map.get("syncPath")).append("}");             }                          sb.append("\t }");             if(i!=size){                 sb.append(",");             }         }        sb.append("]").append("\n\t").append("}");          //继续 保存client                sb.append("\n\t").append("client{").append("\n\t").append("currentHost=").append(context.getCurrHost().getHostString()).append("\n\t");        sb.append("\n\t").append("currentPort=").append(context.getCurrHost().getPort()).append("\n\t");        sb.append("}");                 return sb.toString();               }       @Override     protected void commit() {                         }       @Override     protected void sync() {                         }       public static void main(String[] args) {                               }   }

ok 基本的简单的 配置 以及 上下文 这些用来同步的 用来做桥梁的都已经做好了 下面就是怎么监听配置文件的更改，有人说怎么监听一个配置文件更改啊，其实 文件有个属性 ***更改的时间，只要监听这个就可以 初级版本没有加上更改，这个后面可以随时加上 而且为了以后好更改 只写了相关方式没有 加上

import java.io.File;   public interface FileListener {     public void fileChanged(File file); }

然后用定时器 timeer 或者线程池定时线程去轮训他 ，

这里 注意弱应用 ，其实 对于这些时间，我建议用LinkTransferQueue，他是FIFO 先进先出的无阻晒的队列 然后队列 加上若引用，保证时间的一些顺序，

然后 通知了接口 在接口里面我们在做相应的更改配置 同步配置 的相关操作，这一章基本思想就是

 同步 资源 配置 上下文为桥梁 事件驱动为引擎

还有支出的是 timer excutors.newSchulerXXXX  还有很多方式实现轮训的方式  这种方式 也可以实现心跳线

同时 告诉大家的事 Apache、 camel| 里面的事件驱动文件部分 核心代码就是上面 这一小部分代码没有贴完整 ，但是代码已经托管到  http://git.oschina.net/zhuyuping/distributeTemplate

关于distributeTemplate分布式配置以及上下文的同步维护是怎样的就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。