分布式学习之zookeeper实战篇服务注册与发现2

　　前面我们使用zk的原生客户端实现了简单的服务注册与发现
　　先来分析下前面的程序缺点有哪些 原生客户端的监听都是一次性的 没有负载均衡（说白了就是前面写的就不支持同一个服务注册多个）
　　我们来一个一个解决上面的问题。首先我们这次不采用原生客户端而采用zkClinet来处理改写之前的程序。在此基础上 先解决只能监听一次 的问题。
　　pom.xml文件自然要增加 zkClinet的依赖      com.101tec     zkclient     0.10 
　　配置文件 applciation..yml - 这里将原来写死在类里面的部分数据抽取到了配置文件中，如节点目录、zk地址、项目名称 spring:   config:     name: product server:   port: 8081 zk:   address: 127.0.0.1:2181   root: /clinetServices
　　product
　　首先使用zkClinet来首先 注册器 @Component public class RegisterClinet {      @Value(＂${zk.root}＂)     private String zkRoot;      @Value(＂${spring.config.name}＂)     private String appName;      @Value(＂${zk.address}＂)     private String address;      @Value(＂${server.port}＂)     private String serverPort;      // 注册     public void register(){         ZkClient zkClient = new ZkClient(address, 5000);         // 解决乱码问题         zkClient.setZkSerializer(new ZkCodeSerializer());          // 创建根目录         if (!zkClient.exists(zkRoot)) {             zkClient.createPersistent(zkRoot);         }          String ip = ＂＂;         try {             InetAddress ip4 = Inet4Address.getLocalHost();             ip = ip4.getHostAddress();         } catch (UnknownHostException e) {             System.out.println(＂--------------------------＂);         }          String value = ip + ＂:＂ + serverPort;         zkClient.createEphemeralSequential(zkRoot + ＂/＂ + appName, value);      }      // 解决中文乱码     class ZkCodeSerializer implements ZkSerializer {         @Override         public byte[] serialize(Object object) throws ZkMarshallingError {             return ((String)object).getBytes(Charset.forName(＂UTF-8＂));         }          @Override         public Object deserialize(byte[] bytes) throws ZkMarshallingError {             return new String(bytes, Charset.forName(＂UTF-8＂));         }     } }
　　初始化的地方自然也要改写一下 public class InitListener implements ServletContextListener {     // 前面写过，这里就把变动点写一下     /*         String hostAddress = InetAddress.getLocalHost().getHostAddress();         ServiceRegister.register(hostAddress,port);     */     registerClinet.register();     // ... }
　　cusotmer
　　消费者这边的获取服务列表及持续监听 @Component public class InitZkClinetListener  {      @Value(＂${zk.root}＂)     private String zkRoot;      @Value(＂${zk.address}＂)     private String address;      @PostConstruct     public void contextInitialized() {          ZkClient zkClient = new ZkClient(address, 5000);         // 解决乱码问题         zkClient.setZkSerializer(new ZkCodeSerializer());         try {             // 第一次直接调用刷新缓存列表             refreshServiceList();             // 监听变化             reWatch(zkClient);          } catch (Exception e) {             System.out.println(＂---------------＂);         }      }      /**      * 刷新注册列表      */     private void refreshServiceList() {         ZkClient zkClient = new ZkClient(address, 5000);         // 解决乱码问题         zkClient.setZkSerializer(new ZkCodeSerializer());          List serverList = new ArrayList();         List children = zkClient.getChildren(zkRoot);         for (String child: children) {             String sAddr = zkClient.readData(zkRoot + ＂/＂ + child);             serverList.add(sAddr);         }         ServiceCache.SERVICE_LIST = serverList;     }      /**      * 持续监听      * @param zkClient      */     private void reWatch(ZkClient zkClient){          zkClient.subscribeChildChanges(zkRoot, new IZkChildListener() {             // 监测子节点的操作事件             @Override             public void handleChildChange(String s, List list) throws Exception {                 refreshServiceList();                 reWatch(zkClient);             }         });     }      // 解决中文乱码     class ZkCodeSerializer implements ZkSerializer {         @Override         public byte[] serialize(Object object) throws ZkMarshallingError {             return ((String)object).getBytes(Charset.forName(＂UTF-8＂));         }          @Override         public Object deserialize(byte[] bytes) throws ZkMarshallingError {             return new String(bytes, Charset.forName(＂UTF-8＂));         }     } }
　　消费者不在使用ServletContextListener，所以从启动类中注释掉之前写的注入bean @SpringBootApplication public class SpackManagerZkCustomerApplication {      public static void main( String[] args ) {         SpringApplication.run(SpackManagerZkCustomerApplication.class);     }   /*   @Bean     public ServletListenerRegistrationBean servletListenerRegistrationBean() {         ServletListenerRegistrationBean servletListenerRegistrationBean = new ServletListenerRegistrationBean();         servletListenerRegistrationBean.setListener(new InitListener());         return servletListenerRegistrationBean;     }*/      @Bean     public RestTemplate restTemplate() {         return new RestTemplate();     } }
　　ok,启动测试。测试过程： 启动product 启动customer 调用customer测试类，debug查看列表（当然也可以写一个方法直接查列表） 停止product(直接停止应用不会立即删除临时节点，需要稍微等一会、这里可以使用zk客户端查看zk对应的临时节点是否还在) 调用customer测试类，debug查看列表 启动product 调用customer测试类，debug查看列表
　　当然也可以直接打断点的方式测试.我这里就不把测试截图贴上了.
　　接下来我们来解决负载问题（我们就简单的采用随机负载）
　　首先我们先来处理解决缓存名称问题，很简单。在product注册的时候将服务名称也写进去。代码如下 String value = appName + ＂#＂ + ip + ＂:＂ + serverPort; zkClient.createEphemeralSequential(zkRoot + ＂/＂ + appName, value);
　　在customer解析的时候按照规则进行拆分解析即可 private void refreshServiceList() {     ZkClient zkClient = new ZkClient(address, 5000);     // 解决乱码问题     zkClient.setZkSerializer(new ZkCodeSerializer());      Map> service_map = new HashMap<>();     List children = zkClient.getChildren(zkRoot);     for (String child: children) {         String sAddr = zkClient.readData(zkRoot + ＂/＂ + child);         String key = sAddr.split(＂#＂)[0];         Set set = service_map.get(key) == null ? new HashSet<>() : service_map.get(key);         set.add(sAddr.split(＂#＂)[1]);         service_map.put(key, set);     }     ServiceCache.SERVICE_MAP = service_map; }
　　负载的处理 public class ServiceCache {      // zk  service列表缓存     public volatile static List SERVICE_LIST;      public volatile static Map> SERVICE_MAP;      // 实现随机访问     public static String loadBalance(String servieName){         Set set = SERVICE_MAP.get(servieName);         if(!CollectionUtils.isEmpty(set)) {             int index = new Random().nextInt(SERVICE_LIST.size());             return SERVICE_LIST.get(index);         }         return ＂＂;     } }
　　这样就完成了整体的二次优化。后面继续优化