ZookeeperNamingService extends FailbackNamingService implements NamingService { @Override public List<ServiceInstance> lookup(SubscribeInfo subscribeInfo) { String path = getSubscribePath(subscribeInfo); List<ServiceInstance> instances = new ArrayList<ServiceInstance>(); try { List<String> childList = client.getChildren().forPath(path); for (String child : childList) { String childPath = path + "/" + child; try { String childData = new String(client.getData().forPath(childPath)); Endpoint endpoint = GsonUtils.fromJson(childData, Endpoint.class); ServiceInstance instance = new ServiceInstance(endpoint); if (subscribeInfo != null && StringUtils.isNoneBlank(subscribeInfo.getServiceId())) { instance.setServiceName(subscribeInfo.getServiceId()); } instances.add(instance); } catch (Exception getDataFailedException) { log.warn("get child data failed, path:{}, ex:", childPath, getDataFailedException); } } log.info("lookup {} instances from {}", instances.size(), url); } catch (Exception ex) { log.warn("lookup end point list failed from {}, msg={}", url, ex.getMessage()); if (!subscribeInfo.isIgnoreFailOfNamingService()) { throw new RpcException("lookup end point list failed from zookeeper failed", ex); } } return instances; } ZookeeperNamingService(BrpcURL url); @Override List<ServiceInstance> lookup(SubscribeInfo subscribeInfo); @Override void doSubscribe(SubscribeInfo subscribeInfo, final NotifyListener listener); @Override void doUnsubscribe(SubscribeInfo subscribeInfo); @Override void doRegister(RegisterInfo registerInfo); @Override void doUnregister(RegisterInfo registerInfo); @Override void destroy(); String getSubscribePath(SubscribeInfo subscribeInfo); String getParentRegisterPath(RegisterInfo registerInfo); String getRegisterPath(RegisterInfo registerInfo); String getRegisterPathData(RegisterInfo registerInfo);  }
@Test public void testLookup() throws Exception { setUp(); SubscribeInfo subscribeInfo = createSubscribeInfo(true); List<ServiceInstance> instances = namingService.lookup(subscribeInfo); Assert.assertTrue(instances.size() == 0); RegisterInfo registerInfo = createRegisterInfo("127.0.0.1", 8012); namingService.register(registerInfo); instances = namingService.lookup(subscribeInfo); Assert.assertTrue(instances.size() == 1); Assert.assertTrue(instances.get(0).getIp().equals("127.0.0.1")); Assert.assertTrue(instances.get(0).getPort() == 8012); namingService.unregister(registerInfo); tearDown(); }
@Test @Ignore public void testSubscribeWhenZookeeperDownAndUp() throws Exception { namingUrl = new BrpcURL("zookeeper: namingService = new ZookeeperNamingService(namingUrl); final List<ServiceInstance> adds = new ArrayList<ServiceInstance>(); final List<ServiceInstance> deletes = new ArrayList<ServiceInstance>(); SubscribeInfo subscribeInfo = createSubscribeInfo(false); namingService.subscribe(subscribeInfo, new NotifyListener() { @Override public void notify(Collection<ServiceInstance> addList, Collection<ServiceInstance> deleteList) { System.out.println("receive new subscribe info time:" + System.currentTimeMillis()); System.out.println("add size:" + addList.size()); for (ServiceInstance instance : addList) { System.out.println(instance); } adds.addAll(addList); System.out.println("delete size:" + deleteList.size()); for (ServiceInstance instance : deleteList) { System.out.println(instance); } deletes.addAll(deleteList); } }); RegisterInfo registerInfo = createRegisterInfo("127.0.0.1", 8014); namingService.register(registerInfo); System.out.println("register time=" + System.currentTimeMillis()); Thread.sleep(1000); Assert.assertTrue(adds.size() == 1); Assert.assertTrue(deletes.size() == 0); Assert.assertTrue(adds.get(0).getIp().equals("127.0.0.1")); Assert.assertTrue(adds.get(0).getPort() == 8014); adds.clear(); deletes.clear(); Thread.sleep(30 * 1000); List<ServiceInstance> instances = namingService.lookup(subscribeInfo); Assert.assertTrue(instances.size() == 1); Assert.assertTrue(instances.get(0).getIp().equals("127.0.0.1")); Assert.assertTrue(instances.get(0).getPort() == 8014); namingService.unregister(registerInfo); System.out.println("unregister time=" + System.currentTimeMillis()); Thread.sleep(1000); Assert.assertTrue(adds.size() == 0); Assert.assertTrue(deletes.size() == 1); Assert.assertTrue(deletes.get(0).getIp().equals("127.0.0.1")); Assert.assertTrue(deletes.get(0).getPort() == 8014); namingService.unsubscribe(subscribeInfo); }
@Test @Ignore public void testRegisterWhenZookeeperDownAndUp() throws Exception { namingUrl = new BrpcURL("zookeeper: namingService = new ZookeeperNamingService(namingUrl); RegisterInfo registerInfo = createRegisterInfo("127.0.0.1", 8015); namingService.register(registerInfo); SubscribeInfo subscribeInfo = createSubscribeInfo(false); List<ServiceInstance> instances = namingService.lookup(subscribeInfo); Assert.assertTrue(instances.size() == 1); Assert.assertTrue(instances.get(0).getIp().equals("127.0.0.1")); Assert.assertTrue(instances.get(0).getPort() == 8015); Thread.sleep(30 * 1000); instances = namingService.lookup(subscribeInfo); Assert.assertTrue(instances.size() == 1); System.out.println(instances.get(0)); Assert.assertTrue(instances.get(0).getIp().equals("127.0.0.1")); Assert.assertTrue(instances.get(0).getPort() == 8015); namingService.unregister(registerInfo); }