一、添加maven依赖

<!-- ZooKeeper -->
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.6</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>2.4.2</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.4.2</version>
<scope>provided</scope>
</dependency>
<!-- Zookeeper -->

二、类ZookeeperService.java

package cn.com.easy.zookeeper;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.collections.CollectionUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
/*** zookeeper服务类，用于创建操作zookeeper的对象* * @author nibili 2015年5月7日* */
public class ZooKeeperService {
private Logger logger = LoggerFactory.getLogger(ZooKeeperService.class);
public static final int MAX_RETRIES = 3000;
public static final int BASE_SLEEP_TIMEMS = 3000;
/** zookeeper服务器列表 */
private String zookeeperServers = "";
/** zookeeper客户端操纵对象 */
private CuratorFramework client;
/** 监听器集合(一键多值数据结构) */
private Multimap<IZookeeperWatch, Object> watchesMap = ArrayListMultimap.create();
public ZooKeeperService(String zookeeperServers) {
this.zookeeperServers = zookeeperServers;
RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIMEMS, MAX_RETRIES);
this.client = CuratorFrameworkFactory.builder().connectString(this.zookeeperServers).retryPolicy(retryPolicy).build();
client.start();
}
/*** 取消监听，* * @param zookeeperWatch*            注册监听时的对象* @auth nibili 2015年5月8日*/
public void removeNodeWatch(IZookeeperWatch zookeeperWatch) {
if (zookeeperWatch == null) {
logger.info("称除节点监听，监听器对象不能为空!");
return;
}
Collection<Object> values = watchesMap.get(zookeeperWatch);
if (CollectionUtils.isNotEmpty(values) == true) {
// 移除监听器
NodeCache cache = null;
NodeCacheListener nodeCacheListener = null;
Iterator<Object> it = values.iterator();
for (int i = 0; it.hasNext() && i < 2; i++) {
if (i == 0) {
cache = (NodeCache) it.next();
} else if (i == 1) {
nodeCacheListener = (NodeCacheListener) it.next();
} else {
break;
}
}
if (cache != null && nodeCacheListener != null) {
cache.getListenable().removeListener(nodeCacheListener);
}
} else {
logger.info("没有找到对应的监听器！");
return;
}
}
/*** 监听节点变化* * @param zookeeperWatch* @throws Exception* @auth nibili 2015年5月8日*/
public void addNodeWatch(final IZookeeperWatch zookeeperWatch) throws Exception {
// 是否是每一次触发
final AtomicBoolean isFirst = new AtomicBoolean(true);
final NodeCache cache = new NodeCache(this.client, zookeeperWatch.getWatchPath());
cache.start();
NodeCacheListener nodeCacheListener = new NodeCacheListener() {
@Override
public void nodeChanged() throws Exception {
// 节点数据
String data = new String(cache.getCurrentData().getData(), "UTF-8");
if (isFirst.get() == true) {
isFirst.set(false);
logger.debug("NodeCache loaded, data is: " + data);
zookeeperWatch.handLoad(data);
} else {
logger.debug("NodeCache changed, data is: " + data);
zookeeperWatch.handChange(data);
}
}
};
cache.getListenable().addListener(nodeCacheListener);
watchesMap.put(zookeeperWatch, cache);
watchesMap.put(zookeeperWatch, nodeCacheListener);
}
/*** 断开连接* * @auth nibili 2015年5月7日*/
public void close() {
client.close();
}
/*** 获取zookeeper操纵对象* * @param servers* @return* @auth nibili 2015年5月7日*/
public CuratorFramework getClient() {
return client;
}
/*** 获取服务器地址* * @return* @throws Exception* @auth nibili 2015年5月7日*/
public String getServers() {
return this.zookeeperServers;
}
/*** 设置节点值* * @param path* @param data* @auth nibili 2015年5月8日*/
public void setPathValue(String path, String data) {
try {
logger.debug("设置结点值,path:" + path + "，data:" + data);
this.client.setData().forPath(path, data.getBytes("UTF-8"));
} catch (Exception e) {
logger.error("设置zookeeper节点值异常，path:" + path + "，data" + data, e);
}
}
/*** 获取节点值* * @param path* @return* @throws Exception* @auth nibili 2015年5月7日*/
public byte[] getPathValue(String path) throws Exception {
if (!exists(this.client, path)) {
throw new RuntimeException("Path " + path + " does not exists.");
}
return client.getData().forPath(path);
}
/*** 节点是否存在* * @param client* @param path* @return* @throws Exception* @auth nibili 2015年5月7日*/
private boolean exists(CuratorFramework client, String path) throws Exception {
Stat stat = client.checkExists().forPath(path);
return !(stat == null);
}
/*** 获取子节点* * @param path* @return* @throws Exception* @auth nibili 2015年5月7日*/
public List<String> getSubPaths(String path) throws Exception {
return client.getChildren().forPath(path);
}
}

三、在appliction.properties文件中添加zookeeper地址

zk.servers=192.168.1.120:2181

四、定义applictionContext.xml

<!--属性文件 -->
<context:property-placeholder
location="classpath*:applicationContext-zookeeper-watch-demo.properties" />
<!-- 使用zookeeper节点监听，通知，功能时，要添加这个bean -->
<bean id="zookeeperService" class="cn.com.easy.zookeeper.ZooKeeperService">
<constructor-arg>
<value>${zk.servers}</value>
</constructor-arg>
</bean>

五、测试

package cn.com.easy.zookeeper;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("classpath:/applicationContext-zookeeper-watch-demo.xml")
public class ZookeeperWatchServiceTest {
@Autowired
private ZooKeeperService zookeeperService;
private String path = "/summall/conf/jdbc.username/aa";
@Test
public void addNodeWatch() throws Exception {
try {
zookeeperService.addNodeWatch(new IZookeeperWatch() {
@Override
public void handChange(String data) {
System.out.println("获取到数据变化-1" + path + "：" + data);
}
@Override
public String getWatchPath() {
return path;
}
@Override
public void handLoad(String data) {
System.out.println("获取到数据Loaded-1" + path + "：" + data);
}
});
IZookeeperWatch zookeeperWatch = new IZookeeperWatch() {
@Override
public void handChange(String data) {
System.out.println("获取到数据变化-2" + path + "：" + data);
}
@Override
public String getWatchPath() {
return path;
}
@Override
public void handLoad(String data) {
System.out.println("获取到数据Loaded-2" + path + "：" + data);
}
};
zookeeperService.addNodeWatch(zookeeperWatch);
Thread.sleep(10000);
// 移除监听
zookeeperService.removeNodeWatch(zookeeperWatch);
Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
@Test
public void setPathValue() {
zookeeperService.setPathValue(path, "mypath.......");
}
}

‍‍六、注意‍‍

有同事说，只有在节点数值有变化时，才会通知到客户端。

但是测试过程中的现象是：设置一个节点的值，尽管每次都设成一样的，一样能通知到客户端。