通过Curator操作ZooKeeper_zk curator 一直调用同一个ip

Curator

Curator是Netflix公司开源的一套zookeeper客户端框架，解决了很多Zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册Watcher和NodeExistsException异常等等。

Curator 最初是 Netfix 研发的,后来捐献了 Apache 基金会,目前是 Apache 的顶级项目。

Curator建立与客户端的连接与关闭连接

1. 方式一

   /**
    * 创建连接
    */
   @Before
   public void connectionTest() {
   
   /*
   		定义重试策略
   		参数1：表示如果客户端没有连接成功，那么3000毫秒就尝试连接1次
   		参数2：最多连接10次就不再连接
   	*/
   
   ExponentialBackoffRetry retry = new ExponentialBackoffRetry(3000, 10);
   
   /*
   		创建连接
   		connectString():设置连接的zookeeper服务器的ip:端口号
   		connectionTimeoutMs():设置最大连接超时时间
   		sessionTimeoutMs():设置会话时间，如果在设定的时间内客户端与服务器端没有进行交流，那么断开连接。
   		retryPolicy():定义重试的策略
   
   	*/
   CuratorFramework client = CuratorFrameworkFactory.builder().connectString("192.168.242.130:2181")
   	.connectionTimeoutMs(15 * 1000)
   	.sessionTimeoutMs(1000 * 60)
   	.retryPolicy(retry).build();
   
   //开启连接
   client.start();
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30

2. 方式二

   /**
    * 创建连接
    */
   @Before
   public void getConnection() {
   
   	/*
   		String connectString：连接zookeeper服务器的参数字符串 ip:port
   		int sessionTimeoutMs：会话超时时间，在设置的时间内，客户端和服务器不交流，就断开连接
   		int connectionTimeoutMs：连接超时时间
   		RetryPolicy retryPolicy：重试策略；当客户端连接不上服务器的时候，使用什么样的重试策略。
   									策略如：1.一直继续连接，直到连接上位置
   										   2.在尝试连接几次，连接不上就放弃连接。
   										   ....
   
   	   RetryUntilElapsed: 是retryPolicy的实现类，表示一直重试，直到重试的时间超过最大时间后，就不再重试
   	   maxElapsedTimeMs 最大重试时间
   	   sleepMsBetweenRetries 每次重试的间隔时间
   	*/
   	client = CuratorFrameworkFactory.newClient("192.168.242.130:2181",
   			1000 * 60,
   			1000 * 10,
   			new RetryUntilElapsed(10 * 1000, 1000));
   
   	//开启连接
   	client.start();
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27

3. 关闭连接

   /**
        * 关闭连接
        */
   u
   public void close() {
       //关闭客户端与服务器的连接
       client.close();
   }
   1
   2
   3
   4
   5
   6
   7
   8

Curator创建节点

​ 注意以下的client是创建的客户端连接对象

• 创建普通节点

  /**
   * 创建普通节点
   */
  @Test
  public void createTest() throws Exception {
      String path = client.create().forPath("/app1");
      System.out.println(path);
  }
  1
  2
  3
  4
  5
  6
  7
  8

• 创建普通节点，并设置节点数据

  /**
   * 创建一个节点并设置数据
   *
   * @throws Exception
   */
  @Test
  public void createTest01() throws Exception {
      String path = client.create().forPath("/app2", "肖肖".getBytes());
      System.out.println(path);
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

• 创建临时节点，并设置节点数据

  /**
   * 创建一个临时节点并设置数据
   *
   * @throws Exception
   */
  @Test
  public void createTest02() throws Exception {
      String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3");
      System.out.println(path);
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

• 创建一个顺序持久节点

  /**
       * 创建一个顺序持久节点
       *
       * @throws Exception
       */
  @Test
  public void createTest03() throws Exception {
      String path = client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/app4");
      System.out.println(path);
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

Curator查询节点

• 查询节点的数据

  /**
       * 查询节点的数据
       */
  @Test
  public void getTest() throws Exception {
      byte[] bytes = client.getData().forPath("/app2");
      System.out.println(new String(bytes));
  }
  1
  2
  3
  4
  5
  6
  7
  8

• 查询指定节点的子节点有哪些

  /**
   * 查询指定节点的子节点有哪些
   */
  @Test
  public void getChildNode() throws Exception {
      List<String> strings = client.getChildren().forPath("/");
      strings.forEach(node -> System.out.println(node));
  }
  
  1
  2
  3
  4
  5
  6
  7
  8
  9

• 查询指定节点的状态信息

  /**
       * 查询当前节点的状态信息
       */
  @Test
  public void getInfoTest() throws Exception {
      //创建节点的状态信息对象
      Stat stat = new Stat();
      //查询指定节点的状态信息，存入stat对象中
      client.getData().storingStatIn(stat).forPath("/app2");
      System.out.println(stat);
  }
  
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12

Curator修改节点

• 修改节点的数据

  /**
       * 修改节点的数据
       */
  @Test
  public void setTest() throws Exception {
      //要求要修改节点的节点必须存在
      Stat path = client.setData().forPath("/app40000000015", "张三".getBytes());
      System.out.println(path);
  }
  
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

• 根据版本号，修改节点的数据

  /**
       * 根据版本号修改节点的数据
       */
  @Test
  public void setVersion() throws Exception {
  
      //创建节点的状态信息独享
      Stat status = new Stat();
  
      //获取节点的状态信息，获取到的状态信息会存入status对象中
      client.getData().storingStatIn(status).forPath("/app2");
      //从节点状态信息对象中获取节点的版本号
      int version = status.getVersion();
      System.out.println(version);
  
      /*
              根据版本修改节点的数据
              当withVersion()中的版本数据与从对象信息中获取的版本相同的时候，才会执行修改，否则报错。
              这样就可以保证修改数据的时候的原子性，必须保持版本一致，才会进行节点的数据修改。
          */
      client.setData().withVersion(version).forPath("/app2", "橘右京".getBytes());
  }
  
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23

Curator删除节点

• 删除节点

  /**
       * 删除节点
       */
  @Test
  public void deleteTest() throws Exception {
      client.delete().forPath("/app1");
  }
  
  1
  2
  3
  4
  5
  6
  7
  8

• 删除指定节点及其子节点

  /**
       * 删除指定节点及其子节点
       */
  @Test
  public void deleteNodeAndChildNodeTest() throws Exception {
      client.delete().deletingChildrenIfNeeded().forPath("/app1");
  }
  
  1
  2
  3
  4
  5
  6
  7
  8