zookeeper 事件监听

末蓝、 2022-12-10 01:14 273阅读 0赞

zookeeper 事件监听

官网：https://github.com/apache/curator/tree/master/curator-examples/src/main/java

**********************

相关类与接口

CuratorCache

  1. public interface CuratorCache extends Closeable, CuratorCacheAccessor {
  2.     static CuratorCache build(CuratorFramework client, String path, CuratorCache.Options... options) {
  3.         return builder(client, path).withOptions(options).build();
  4.     }
  6.     static CuratorCacheBuilder builder(CuratorFramework client, String path) {
  7.         return new CuratorCacheBuilderImpl(client, path);
  8.     }
  10.     static CuratorCacheBridgeBuilder bridgeBuilder(CuratorFramework client, String path) {
  11.         return new CuratorCacheBridgeBuilderImpl(client, path);
  12.     }
  14.     void start();
  15.     void close();
  17.     Listenable<CuratorCacheListener> listenable();
  19.     Optional<ChildData> get(String var1);
  21.     int size();
  22.     Stream<ChildData> stream();
  24.     public static enum Options {
  25.         SINGLE_NODE_CACHE,
  26.         COMPRESSED_DATA,
  27.         DO_NOT_CLEAR_ON_CLOSE;
  29.         private Options() {
  30.         }
  31.     }
  32. }

Listeneable：添加、删除监听接口

  1. public interface Listenable<T> {
  3.     void addListener(T var1);
  4.     void addListener(T var1, Executor var2);  //添加监听接口
  6.     void removeListener(T var1);              //删除监听接口
  7. }

CuratorCacheListener：监听接口

  1. @FunctionalInterface
  2. public interface CuratorCacheListener {
  3.     void event(CuratorCacheListener.Type var1, ChildData var2, ChildData var3);
  5.     default void initialized() {
  6.     }
  8.     static CuratorCacheListenerBuilder builder() {
  9.         return new CuratorCacheListenerBuilderImpl();
  10.     }
  12.     public static enum Type {
  13.         NODE_CREATED,
  14.         NODE_CHANGED,
  15.         NODE_DELETED;
  17.         private Type() {
  18.         }
  19.     }
  20. }

CuratorCacheListenerBuilder：创建CuratorCacheListener

  1. public interface CuratorCacheListenerBuilder {
  3.     CuratorCacheListenerBuilder forAll(CuratorCacheListener var1);      //所有操作件
  4.     CuratorCacheListenerBuilder forCreates(Consumer<ChildData> var1);   //创建操作
  5.     CuratorCacheListenerBuilder forChanges(CuratorCacheListenerBuilder.ChangeListener var1);  //修改操作
  6.     CuratorCacheListenerBuilder forCreatesAndChanges(CuratorCacheListenerBuilder.ChangeListener var1);  //创建修改操作
  7.     CuratorCacheListenerBuilder forDeletes(Consumer<ChildData> var1);   //删除操作
  9.     CuratorCacheListenerBuilder forInitialized(Runnable var1);          //初始化节点
  11.     CuratorCacheListenerBuilder forPathChildrenCache(String var1, CuratorFramework var2, PathChildrenCacheListener var3);   //子节点
  12.     CuratorCacheListenerBuilder forTreeCache(CuratorFramework var1, TreeCacheListener var2);  //监控整个树节点状态
  13.     CuratorCacheListenerBuilder forNodeCache(NodeCacheListener var1); //监控当前节点
  15.     CuratorCacheListenerBuilder afterInitialized();
  16.     CuratorCacheListener build();
  18.     @FunctionalInterface
  19.     public interface ChangeListener {
  20.         void event(ChildData var1, ChildData var2);
  21.     }
  22. }

NodeCacheListener：监听当前节点

  1. public interface NodeCacheListener {
  2.     void nodeChanged() throws Exception;
  3. }

说明：创建当前节点、更新数据会触发通知，删除当前节点不会触发通知；添加、修改、删除子节点会触发通知

PathChildrenCacheListener：监听当前节点子节点

  1. public interface PathChildrenCacheListener {
  2.     void childEvent(CuratorFramework var1, PathChildrenCacheEvent var2) throws Exception;
  3. }

PathChildrenCacheEvent：子节点创建、数据更新、删除

  1. public class PathChildrenCacheEvent {
  2.     private final PathChildrenCacheEvent.Type type;
  3.     private final ChildData data;
  5.     public PathChildrenCacheEvent(PathChildrenCacheEvent.Type type, ChildData data) {
  6.         this.type = type;
  7.         this.data = data;
  8.     }
  10.     public PathChildrenCacheEvent.Type getType() {
  11.         return this.type;
  12.     }
  14.     public ChildData getData() {
  15.         return this.data;
  16.     }
  18.     public List<ChildData> getInitialData() {
  19.         return null;
  20.     }
  22.     public String toString() {
  23.         return "PathChildrenCacheEvent{type=" + this.type + ", data=" + this.data + '}';
  24.     }
  26.     public static enum Type {
  27.         CHILD_ADDED,
  28.         CHILD_UPDATED,
  29.         CHILD_REMOVED,
  30.         CONNECTION_SUSPENDED,
  31.         CONNECTION_RECONNECTED,
  32.         CONNECTION_LOST,
  33.         INITIALIZED;
  35.         private Type() {
  36.         }
  37.     }
  38. }

TreeCacheListener：监听整个目录

  1. public interface TreeCacheListener {
  2.     void childEvent(CuratorFramework var1, TreeCacheEvent var2) throws Exception;
  3. }

TreeCacheEvent：节点添加、更新、删除触发

  1. public class TreeCacheEvent {
  2.     private final TreeCacheEvent.Type type;
  3.     private final ChildData data;
  4.     private final ChildData oldData;
  6.     public TreeCacheEvent(TreeCacheEvent.Type type, ChildData data) {
  7.         this(type, data, (ChildData)null);
  8.     }
  10.     public TreeCacheEvent(TreeCacheEvent.Type type, ChildData data, ChildData oldData) {
  11.         this.type = type;
  12.         this.data = data;
  13.         this.oldData = oldData;
  14.     }
  16.     public TreeCacheEvent.Type getType() {
  17.         return this.type;
  18.     }
  20.     public ChildData getData() {
  21.         return this.data;
  22.     }
  24.     public ChildData getOldData() {
  25.         return this.oldData;
  26.     }
  28.     public String toString() {
  29.         return TreeCacheEvent.class.getSimpleName() + "{type=" + this.type + ", data=" + this.data + '}';
  30.     }
  32.     public static enum Type {
  33.         NODE_ADDED,
  34.         NODE_UPDATED,
  35.         NODE_REMOVED,
  36.         CONNECTION_SUSPENDED,
  37.         CONNECTION_RECONNECTED,
  38.         CONNECTION_LOST,
  39.         INITIALIZED;
  41.         private Type() {
  42.         }
  43.     }
  44. }

**********************

示例

*****************

config 层

WebConfig

  1. @Configuration
  2. public class WebConfig {
  4.     @Value("${zookeeper.connect-string}")
  5.     private String connectString;
  7.     @Bean
  8.     public CuratorFramework initCuratorFramework(){
  9.         CuratorFramework curatorFramework= CuratorFrameworkFactory.builder()
  10.                 .connectString(connectString)
  11.                 .sessionTimeoutMs(5000)
  12.                 .retryPolicy(new ExponentialBackoffRetry(1000,3,3000))
  13.                 .namespace("test")
  14.                 .build();
  15.         curatorFramework.start();
  17.         return curatorFramework;
  18.     }
  19. }

*****************

controller 层

HelloController

  1. @RestController
  2. public class HelloController {
  4.     @Resource
  5.     private CuratorFramework curatorFramework;
  7.     private final String path="/cache";
  8.     private final String childPath="/cache/child";
  10.     private final String path2="/cache2";
  11.     private final String childPath2="/cache2/child";
  13.     @RequestMapping("/hello")
  14.     public void hello(){
  15.         CuratorCache cache=CuratorCache.build(curatorFramework,path);
  16.         CuratorCacheListener curatorCacheListener=CuratorCacheListener.builder()
  17.                 .forNodeCache(() -> {
  18.                     if (cache.get(path).isPresent()){
  19.                         System.out.println("当前数据节点变更，新数据："+new String(cache.get(path).get().getData()));
  20.                     }
  21.                 }).build();
  23.         cache.listenable().addListener(curatorCacheListener);
  24.         cache.start();
  26.         try {
  27.             curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath(path,"瓜田李下".getBytes());
  28.             curatorFramework.setData().forPath(path,"海贼王".getBytes());
  30.             curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath(childPath,"海贼王".getBytes());
  31.             curatorFramework.setData().forPath(childPath,"瓜田李下".getBytes());
  33.             curatorFramework.delete().deletingChildrenIfNeeded().forPath(childPath);
  34.             curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
  35.         }catch (Exception e){
  36.             System.out.println(e.getMessage());
  37.         }
  38.     }
  40.     @RequestMapping("/hello2")
  41.     public void hello2(){
  42.         CuratorCache cache=CuratorCache.build(curatorFramework,path2);
  43.         CuratorCacheListener curatorCacheListener=CuratorCacheListener.builder()
  44.                 .forPathChildrenCache(path2, curatorFramework, (curatorFramework, pathChildrenCacheEvent) -> {
  45.                     switch (pathChildrenCacheEvent.getType()){
  46.                         case CHILD_ADDED -> {
  47.                             System.out.println("添加子节点："+pathChildrenCacheEvent.getData().getPath());
  48.                         }
  50.                         case CHILD_UPDATED -> {
  51.                             System.out.println("更新子节点："+pathChildrenCacheEvent.getData().getPath());
  52.                         }
  54.                         case CHILD_REMOVED -> {
  55.                             System.out.println("删除子节点："+pathChildrenCacheEvent.getData().getPath());
  56.                         }
  57.                     }
  58.                 }).build();
  60.         cache.listenable().addListener(curatorCacheListener);
  61.         cache.start();
  63.         try {
  64.             curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath(path2,"瓜田李下".getBytes());
  65.             curatorFramework.setData().forPath(path2,"海贼王".getBytes());
  67.             curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath(childPath2,"海贼王".getBytes());
  68.             curatorFramework.setData().forPath(childPath2,"瓜田李下".getBytes());
  70.             curatorFramework.delete().deletingChildrenIfNeeded().forPath(childPath2);
  71.             curatorFramework.delete().deletingChildrenIfNeeded().forPath(path2);
  72.         }catch (Exception e){
  73.             System.out.println(e.getMessage());
  74.         }
  75.     }
  76. }

**********************

使用测试

localhost:8080/hello

  1. 当前数据节点变更，新数据：瓜田李下
  2. 当前数据节点变更，新数据：海贼王
  3. 当前数据节点变更，新数据：海贼王
  4. 当前数据节点变更，新数据：海贼王
  5. 当前数据节点变更，新数据：海贼王

触发通知：创建、修改当前节点，创建、修改、删除子节点

不触发通知：删除当前节点

localhost:8080/hello2

  1. 添加子节点：/cache2/child
  2. 更新子节点：/cache2/child
  3. 删除子节点：/cache2/child

发表评论

表情：
评论列表 （有 0 条评论，273人围观）

还没有评论，来说两句吧...

相关阅读