骆驼有没有做网站的公司/百度技术培训中心
第二部分可参看：https://blog.csdn.net/Dongguabai/article/details/82986476
ZooKeeper提供了分布式数据的发布/订阅功能。一个典型的发布/订阅模型系统定义了一种一对多的订阅关系，能够让多个订阅者同时监听某一个主题对象，当这个主题对象自身状态变化时，会通知所有订阅者，使它们能够做出相应的处理。在Zoo Keeper中，引入了Watcher机制来实现这种分布式的通知功能。ZooKeeper允许客户端向服务端注册一个Watcher监听，当服务端的一些指定事件触发了这个Watcher，那么就会向指定客户端发送一个事件通知来实现分布式的通知功能。
Watcher 监听机制是 ZooKeeper中非常重要的特性，我们基于 ZooKeeper上创建的节点，可以对这些节点绑定监听事件，比如可以监听节点数据变更、节点删除、子节点状态变更等事件，通过这个事件机制，可以基于 ZooKeeper实现分布式锁、集群管理等功能。
Watcher特性：当数据发生变化的时候，ZooKeeper会产生一个 Watcher事件，并且会发送到客户端。但是客户端只会收到一次通知。如果后续这个节点再次发生变化，那么之前设置 Watcher的客户端不会再次收到消息（一个Watcher实例是一个回调函数，被回调一次后就被移除了。如果还需要关注数据的变化，需要再次注册watcher，Watcher是一次性的操作）。可以通过循环监听去达到永久监听效果。
绑定Watcher的操作有：getData（获取数据）、exists（是否存在）、getChildren（获取子节点）
触发Watcher的操作（凡是事务类型操作，都会触发）有：create、delete、setData
测试代码：
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;import java.io.IOException;
import java.util.concurrent.CountDownLatch;/*** @author Dongguabai* @date 2018/10/8 18:04*/
public class ZooKeeperWatcherDemo {static final String CONNECT_ADDR = "192.168.220.135:2181,192.168.220.136:2181,192.168.220.137:2181";static final int SESSION_OUTTIME = 2000;//ms/*** 阻塞程序执行，等待zookeeper连接成功*/static final CountDownLatch connectedSemaphore = new CountDownLatch(1);static final String PATH = "/dongguabai";public static void main(String[] args) {try {//连接ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, event -> {//如果收到了服务端的响应事件，连接成功if (Watcher.Event.KeeperState.SyncConnected == event.getState()) {connectedSemaphore.countDown();}});connectedSemaphore.await();System.out.println("连接成功！");System.out.println(zk.getState());//创建节点zk.create(PATH, "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);//可以绑定事件的操作：getData、exists、getChildren// 通过exists绑定事件//如果第二个参数为true,事件会触发至创建该ZooKeeper中的Watcher中//zk.exists(PATH,true);//在当前exists中单独触发一个事件Stat stat = zk.exists(PATH, event -> {System.out.println(event.getType() + "--->" + event.getPath());});//触发Watcher事件System.out.println("执行setData方法");zk.setData(PATH,"1".getBytes(),stat.getVersion());zk.close();} catch (IOException | InterruptedException | KeeperException e) {e.printStackTrace();}}
}执行结果：
再测试：
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;import java.io.IOException;
import java.util.concurrent.CountDownLatch;/*** @author Dongguabai* @date 2018/10/8 18:04*/
public class ZooKeeperWatcherDemo {static final String CONNECT_ADDR = "192.168.220.135:2181,192.168.220.136:2181,192.168.220.137:2181";static final int SESSION_OUTTIME = 2000;//ms/*** 阻塞程序执行，等待zookeeper连接成功*/static final CountDownLatch connectedSemaphore = new CountDownLatch(1);static final String PATH = "/dongguabai";public static void main(String[] args) {try {//连接ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, event -> {//如果收到了服务端的响应事件，连接成功if (Watcher.Event.KeeperState.SyncConnected == event.getState()) {System.out.println("ppppppppppppppppppppppppppp");connectedSemaphore.countDown();}});connectedSemaphore.await();System.out.println("连接成功！");System.out.println(zk.getState());//创建节点zk.create(PATH, "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);//可以绑定事件的操作：getData、exists、getChildren// 通过exists绑定事件//如果第二个参数为true,事件会触发至创建该ZooKeeper中的Watcher中//Stat stat = zk.exists(PATH,true);//在当前exists中单独触发一个事件Stat stat = zk.exists(PATH, event -> {System.out.println(event.getType() + "--->" + event.getPath());});//触发Watcher事件System.out.println("执行setData方法");Stat stat2 = zk.setData(PATH, "1".getBytes(), stat.getVersion());Thread.sleep(1000);System.out.println("执行delete方法");zk.delete(PATH,stat2.getVersion());zk.close();} catch (IOException | InterruptedException | KeeperException e) {e.printStackTrace();}}
}输出结果：
可以看到事件只会被触发一次。但是如果想实现永久监听怎么办呢，比如可以在exists的事件中再绑定一个事件，代码如下：
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;import java.io.IOException;
import java.util.concurrent.CountDownLatch;/*** @author Dongguabai* @date 2018/10/8 18:04*/
public class ZooKeeperWatcherDemo {static final String CONNECT_ADDR = "192.168.220.135:2181,192.168.220.136:2181,192.168.220.137:2181";static final int SESSION_OUTTIME = 2000;//ms/*** 阻塞程序执行，等待zookeeper连接成功*/static final CountDownLatch connectedSemaphore = new CountDownLatch(1);static final String PATH = "/dongguabai";public static void main(String[] args) {try {//连接ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, event -> {System.out.println("事件是："+event.getType());//如果收到了服务端的响应事件，连接成功if (Watcher.Event.KeeperState.SyncConnected == event.getState()) {connectedSemaphore.countDown();}});connectedSemaphore.await();System.out.println("连接成功！");System.out.println(zk.getState());//创建节点zk.create(PATH, "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);//可以绑定事件的操作：getData、exists、getChildren// 通过exists绑定事件//如果第二个参数为true,事件会触发至创建该ZooKeeper中的Watcher中//Stat stat = zk.exists(PATH,true);//在当前exists中单独触发一个事件Stat stat = zk.exists(PATH, event -> {System.out.println(event.getType() + "--->" + event.getPath());try {zk.exists(event.getPath(),true);} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}});//触发Watcher事件System.out.println("执行setData方法");Stat stat2 = zk.setData(PATH, "1".getBytes(), stat.getVersion());Thread.sleep(1000);System.out.println("执行delete方法");zk.delete(PATH,stat2.getVersion());zk.close();} catch (IOException | InterruptedException | KeeperException e) {e.printStackTrace();}}
}运行结果：
可以看到在执行了delete方法后，又触发了事件，我在exists方法中设置的参数为true，就会执行创建ZooKeeper的时候注册的Watcher方法。