zookeeper的事件监听机制

一、watcker的概念

• zookeeper提供了数据的发布/订阅功能，多个订阅者可同时监听某一特定主题对象，当该主题对象的自身状态发生变化时例如节点内容改变、节点下的子节点列表改变等，会实时、主动通知全部订阅者。
• zookeeper采用了 Watcher机制实现数据的发布订阅功能。该机制在被订阅对象发生变化时会异步通知客户端，所以客户端没必要在 Watcher注册后轮询阻塞，从而减轻了客户端压力。
• watcher机制事件上与观察者模式相似，也可看做是一种观察者模式在分布式场景下的实现方式。

二、watcher架构

watcher实现由三个部分组成：java

• zookeeper服务端
• zookeeper客户端
• 客户端的ZKWatchManager对象

客户端首先将 Watcher注册到服务端，同时将 Watcher对象保存到客户端的watch管理器中。当Zookeeper服务端监听的数据状态发生变化时，服务端会主动通知客户端，接着客户端的 Watch管理器会**触发相关 Watcher**来回调相应处理逻辑，从而完成总体的数据 发布/订阅流程。
node

三、watcher特性

四、watcher接口设计

Watcher是一个接口，任何实现了Watcher接口的类就算一个新的Watcher。Watcher内部包含了两个枚举类：KeeperState、EventType

（1）Watcher通知状态(KeeperState)web

KeeperState是客户端与服务端链接状态发生变化时对应的通知类型。路径为org.apache.zookeeper.Watcher.EventKeeperState，是一个枚举类，其枚举属性以下：apache

（2）Watcher事件类型(EventType)服务器

EventType是数据节点znode发生变化时对应的通知类型。EventType变化时KeeperState永远处于SyncConnected通知状态下；当keeperState发生变化时，EventType永远为None。其路径为org.apache.zookeeper.Watcher.Event.EventType，是一个枚举类，枚举属性以下：session

注意：客户端接收到的相关事件通知中只包含状态以及类型等信息，不包含节点变化先后的具体内容，变化前的数据需业务自身存储，变化后的数据须要调用get等方法从新获取架构

（3）捕获相应的事件异步

上面讲到zookeeper客户端链接的状态和zookeeper对znode节点监听的事件类型，下面咱们来说解如何创建zookeeper的***watcher监听***。在zookeeper中采用zk.getChildren(path,watch)、zk.exists(path,watch)、zk.getData(path,watcher,stat)这样的方式来为某个znode注册监听 。分布式

下表以node-x节点为例，说明调用的注册方法和可用监听事件间的关系：ide

（6）注册watcher的方法

客户端与服务器端的链接状态

• KeeperState：通知状态

• SyncConnected：客户端与服务器正常链接时

• Disconnected：客户端与服务器断开链接时

• Expired：会话session失效时

• AuthFailed：身份认证失败时

• 事件类型为：None

  public class ZkConnectionWatcher implements Watcher { 
   
      
      @Override
      public void process(WatchedEvent watchedEvent) { 
   
      
          Event.KeeperState state = watchedEvent.getState();
          if(state == Event.KeeperState.SyncConnected){ 
   
      
              // 正常
              System.out.println("正常链接");
          }else if (state == Event.KeeperState.Disconnected){ 
   
      
              // 能够用Windows断开虚拟机网卡的方式模拟
              // 当会话断开会出现，断开链接不表明不能重连，在会话超时时间内重连能够恢复正常
              System.out.println("断开链接");
          }else if (state == Event.KeeperState.Expired){ 
   
      
              // 没有在会话超时时间内从新链接，而是当会话超时被移除的时候重连会走进这里
              System.out.println("链接过时");
          }else if (state == Event.KeeperState.AuthFailed){ 
   
      
              // 在操做的时候权限不够会出现
              System.out.println("受权失败");
          }
          countDownLatch.countDown();
      }
      private static final String IP = "192.168.133.133:2181"
  ;
      private static CountDownLatch countDownLatch = new CountDownLatch(1);
  
      public static void main(String[] args) throws Exception { 
   
      
          // 5000为会话超时时间
          ZooKeeper zooKeeper = new ZooKeeper(IP, 5000, new ZkConnectionWatcher());
          countDownLatch.await();
          // 模拟受权失败
          zooKeeper.addAuthInfo("digest1","itcast1:123451".getBytes());
          byte[] data = zooKeeper.getData("/hadoop", false, null);
          System.out.println(new String(data));
          TimeUnit.SECONDS.sleep(50);
      }
  }

watcher检查节点

exists

• exists(String path, boolean b)

• exists(String path, Watcher w)

• NodeCreated：节点建立

• NodeDeleted：节点删除

• NodeDataChanged：节点内容

  public class EventTypeTest { 
   
      
      private static final String IP = "192.168.133.133:2181";
      private static CountDownLatch countDownLatch = new CountDownLatch(1);
      private static ZooKeeper zooKeeper;
  
      // 采用zookeeper链接建立时的监听器
      public static void exists1() throws Exception{ 
   
      
          zooKeeper.exists("/watcher1",true);
      }
      // 自定义监听器
      public static void exists2() throws Exception{ 
   
      
          zooKeeper.exists("/watcher1",(WatchedEvent w) -> { 
   
      
              System.out.println("自定义" + w.getType());
          });
      }
      // 演示使用屡次的监听器
      public static void exists3() throws Exception{ 
   
      
          zooKeeper.exists("/watcher1", new Watcher() { 
   
      
              @Override
              public void process(WatchedEvent watchedEvent) { 
   
      
                  try { 
   
      
                      System.out.println("自定义的" + watchedEvent.getType());
                  } finally { 
   
      
                      try { 
   
      
                          zooKeeper.exists("/watcher1",this);
                      } catch (Exception e) { 
   
      
                          e.printStackTrace();
                      }
                  }
              }
          });
      }
      // 演示一节点注册多个监听器
      public static void exists4() throws Exception{ 
   
      
          zooKeeper.exists("/watcher1",(WatchedEvent w) -> { 
   
      
              System.out.println("自定义1" + w.getType());
          });
          zooKeeper.exists("/watcher1", new Watcher() { 
   
      
              @Override
              public void process(WatchedEvent watchedEvent) { 
   
      
                  try { 
   
      
                      System.out.println("自定义2" + watchedEvent.getType());
                  } finally { 
   
      
                      try { 
   
      
                          zooKeeper.exists("/watcher1",this);
                      } catch (Exception e) { 
   
      
                          e.printStackTrace();
                      }
                  }
              }
          });
      }
      // 测试
      public static void main(String[] args) throws Exception { 
   
      
          zooKeeper = new ZooKeeper(IP, 5000, new ZKWatcher());
          countDownLatch.await();
          exists4();
          TimeUnit.SECONDS.sleep(50);
      }
  
      static class ZKWatcher implements Watcher{ 
   
      
          @Override
          public void process(WatchedEvent watchedEvent) { 
   
      
              countDownLatch.countDown();
              System.out.println("zk的监听器" + watchedEvent.getType());
          }
      }
  
  }

getData

• getData(String path, boolean b, Stat stat)
• getData(String path, Watcher w, Stat stat)
• NodeDeleted：节点删除
• NodeDataChange：节点内容发生变化

getChildren

• getChildren(String path, boolean b)
• getChildren(String path, Watcher w)
• NodeChildrenChanged：子节点发生变化
• NodeDeleted：节点删除