CuratorZookeeperClient extends AbstractZookeeperClient<CuratorZookeeperClient.CuratorWatcherImpl, CuratorZookeeperClient.CuratorWatcherImpl> { @Override protected void addTargetDataListener(String path, CuratorZookeeperClient.CuratorWatcherImpl treeCacheListener) { this.addTargetDataListener(path, treeCacheListener, null); } CuratorZookeeperClient(URL url); @Override void createPersistent(String path); @Override void createEphemeral(String path); @Override List<String> getChildren(String path); @Override boolean checkExists(String path); @Override boolean isConnected(); @Override String doGetContent(String path); @Override void doClose(); @Override CuratorZookeeperClient.CuratorWatcherImpl createTargetChildListener(String path, ChildListener listener); @Override List<String> addTargetChildListener(String path, CuratorWatcherImpl listener); @Override void removeTargetChildListener(String path, CuratorWatcherImpl listener);  }
@Test public void testAddTargetDataListener() throws Exception { String listenerPath = "/dubbo/service.name/configuration"; String path = listenerPath + "/dat/data"; String value = "vav"; curatorClient.create(path + "/d.json", value, true); String valueFromCache = curatorClient.getContent(path + "/d.json"); Assertions.assertEquals(value, valueFromCache); final AtomicInteger atomicInteger = new AtomicInteger(0); curatorClient.addTargetDataListener(listenerPath, new CuratorZookeeperClient.CuratorWatcherImpl() { @Override public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception { System.out.println("===" + event); atomicInteger.incrementAndGet(); } }); valueFromCache = curatorClient.getContent(path + "/d.json"); Assertions.assertNotNull(valueFromCache); curatorClient.getClient().setData().forPath(path + "/d.json", "sdsdf".getBytes()); curatorClient.getClient().setData().forPath(path + "/d.json", "dfsasf".getBytes()); curatorClient.delete(path + "/d.json"); curatorClient.delete(path); valueFromCache = curatorClient.getContent(path + "/d.json"); Assertions.assertNull(valueFromCache); Thread.sleep(2000L); Assertions.assertTrue(9L >= atomicInteger.get()); Assertions.assertTrue(2L <= atomicInteger.get()); }