Java Zookeeper - Curator的使用

川长思鸟来 2023-07-01 09:00 19阅读 0赞

## 前言 ##

在上文中 [Java Zookeeper - 极简入门][Java Zookeeper -] ，是不是感觉low的不行，这里我们来使用一种更好的方法

## 背景 ##

Zookeeper原生客户端的不足

*  在连接zk超时的时候，不支持自动重连，需要手动重连。
 *  Watch注册一次就会失效，需要反复注册。
 *  不支持递归创建节点。

Zookeeper开源客户端 - Curator

*  提供更简单易用的API。
 *  解决Watch注册一次就会失效的问题
 *  提供更多解决方法并且实现简单，例如：分布式锁

## Zookeeper节点类型 ##

ZooKeeper 节点是有生命周期的，这取决于节点的类型

<table> 
 <thead> 
  <tr> 
   <th>类型</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>持久节点<br>PERSISTENT</td> 
   <td>所谓持久节点，是指在节点创建后，就一直存在，直到有删除操作来主动清除这个节点——不会因为创建该节点的客户端会话失效而消失。</td> 
  </tr> 
  <tr> 
   <td>持久顺序节点PERSISTENT_SEQUENTIAL</td> 
   <td>这类节点的基本特性和上面的节点类型是一致的。额外的特性是，在ZK中，每个父节点会为他的第一级子节点维护一份时序，会记录每个子节点创建的先后顺序。基于这个特性，在创建子节点的时候，可以设置这个属性，那么在创建节点过程中，ZK会自动为给定节点名加上一个数字后缀，作为新的节点名。这个数字后缀的范围是整型的最大值。</td> 
  </tr> 
  <tr> 
   <td>临时节点<br>EPHEMERAL</td> 
   <td>和持久节点不同的是，临时节点的生命周期和客户端会话绑定。也就是说，如果客户端会话失效，那么这个节点就会自动被清除掉。注意，这里提到的是会话失效，而非连接断开。另外，在临时节点下面不能创建子节点。</td> 
  </tr> 
  <tr> 
   <td>临时顺序节点EPHEMERAL_SEQUENTIAL</td> 
   <td>可以用来实现分布式锁</td> 
  </tr> 
 </tbody> 
</table>

## Curator节点操作 ##

<table> 
 <thead> 
  <tr> 
   <th>操作类型</th> 
   <th>示例说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>创建节点</td> 
   <td>curatorFramework.create().forPath("/home")</td> 
  </tr> 
  <tr> 
   <td>删除节点</td> 
   <td>curatorFramework.delete().forPath("/home")</td> 
  </tr> 
  <tr> 
   <td>更新节点数据</td> 
   <td>curatorFramework.setData().forPath("/home")</td> 
  </tr> 
  <tr> 
   <td>获取节点信息</td> 
   <td>curatorFramework.getData().forPath("/home")</td> 
  </tr> 
  <tr> 
   <td>更新节点权限</td> 
   <td>curatorFramework.setACL().withACL(list).forPath("/create")</td> 
  </tr> 
  <tr> 
   <td>获取节点权限</td> 
   <td>curatorFramework.getACL().forPath("/home")</td> 
  </tr> 
 </tbody> 
</table>

创建一个临时节点

curatorFramework.create()
    .withMode(CreateMode.EPHEMERAL)
    .forPath("/home");

递归创建节点

curatorFramework.create()
      .creatingParentsIfNeeded()
      .forPath("/home/test");

删除节点下的全部子节点

curatorFramework
    .delete()
    .deletingChildrenIfNeeded()
    .forPath("/home");

如果出现故障一次删除失败，那就继续重试，直到删除

curatorFramework.delete()
      .guaranteed()
      .forPath("/home");

获取Stat

Stat stat = new Stat();
    byte[] bytes = curatorFramework
      .getData()
      .storingStatIn(stat)
      .forPath("/home");
    log.info("stat:{} value:{}", stat, new String(bytes));

## Curator-watch通知 ##

<table> 
 <thead> 
  <tr> 
   <th>类型</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>NodeCache</td> 
   <td>监听节点内容变化、节点是否存在</td> 
  </tr> 
  <tr> 
   <td>PathChildrenCache</td> 
   <td>监听子节点变化信息</td> 
  </tr> 
 </tbody> 
</table>

### NodeCache ###

通过命令行创建，home

create /home ww

set /home hello

可以看到代码中监听并输出了

www
    hello

### PathChildrenCache ###

public class CuratorWatch { 
        public static void main(String[] args) throws Exception { 
            CuratorFramework curatorFramework = CuratorFrameworkFactory.
              newClient("127.0.0.1:2181", new RetryNTimes(3, 1000));
            curatorFramework.start();
            // 启动
            String path = "/home";
            PathChildrenCache childrenCache = new PathChildrenCache(
              curatorFramework, path, true);
            childrenCache.start();
            childrenCache.getListenable().addListener((framework, event) -> { 
                switch (event.getType()) { 
                    case CHILD_ADDED:
                        System.out.println("节点创建:" + event.getData().getPath());
                        break;
                    case CHILD_UPDATED:
                        System.out.println("节点更新:" +event.getData().getPath());
                        break;
                    case CHILD_REMOVED:
                        System.out.println("节点删除:" +event.getData().getPath());
                        break;
                    default:
                        break;
                }
            });
    
            // 卡住主线程，不让他直接终止
            Scanner scanner = new Scanner(System.in);
            System.out.println(scanner.nextLine());
        }
    }

通过命令行操作

create /home/cc w

set /home/cc ae

delete /home/cc

## 在SpringBoot中的使用 ##

application.properties

#重试次数
    curator.retryCount=5
    #重试间隔时间
    curator.elapsedTimeMs=5000
    # zookeeper 地址
    curator.connectString=127.0.0.1:2181
    # session超时时间
    curator.sessionTimeoutMs=60000
    # 连接超时时间
    curator.connectionTimeoutMs=5000

配置类

@Configuration
    public class CuratorConfiguration { 
    
        @Value("${curator.retryCount}")
        private int retryCount;
    
        @Value("${curator.elapsedTimeMs}")
        private int elapsedTimeMs;
    
        @Value("${curator.connectString}")
        private String connectString;
    
        @Value("${curator.sessionTimeoutMs}")
        private int sessionTimeoutMs;
    
        @Value("${curator.connectionTimeoutMs}")
        private int connectionTimeoutMs;
    
        @Bean(initMethod = "start")
        public CuratorFramework curatorFramework() { 
            return CuratorFrameworkFactory.builder()
                    .connectString(connectString)
                    .sessionTimeoutMs(sessionTimeoutMs)
                    .connectionTimeoutMs(connectionTimeoutMs)
                    .retryPolicy(new RetryNTimes(retryCount, elapsedTimeMs))
                    .build();
    
        }
    }

Controller

@Slf4j
    @RestController
    @RequiredArgsConstructor(onConstructor = @__(@Autowired))
    public class RestBizController { 
        private final CuratorFramework curatorFramework;
    
        @GetMapping("/zk")
        public String zk() { 
            String value = null;
            try { 
                byte[] bytes = curatorFramework.getData().forPath("/node_test");
                value = new String(bytes);
            } catch (Exception e) { 
                e.printStackTrace();
            }
            return value;
        }
    
    }

访问：http://localhost:8080/zk

[Java Zookeeper -]: https://xiamen.blog.csdn.net/article/details/80951156