安科网

五、Zookeeper基于API操作Node节点

ErixHao

ErixHao

2020-05-30

关注 关注

安装zookeeper ：linux下安装Zookeeper 3.4.14

zookeeper 分为5个包：

  1. org.apache.zookeeper //客户端主要类文件

  2. org.apache.zookeeper.data //

  3. org.apache.zookeeper.server

  4. org.apache.zookeeper.server.quorum

  5. org.apache.zookeeper.server.upgrade

建立会话

引入依赖：

<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.14</version>
</dependency>
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

public class CreateSession implements Watcher {

    private  static CountDownLatch countDownLatch = new CountDownLatch(1);


    /*
      建立会话
     */
    public static void main(String[] args) throws IOException, InterruptedException {

     /*
        客户端可以通过创建一个zk实例来连接zk服务器
        new Zookeeper(connectString,sesssionTimeOut,Wather)
        connectString: 连接地址：IP：端口
        sesssionTimeOut：会话超时时间：单位毫秒
        Wather：监听器(当特定事件触发监听时，zk会通过watcher通知到客户端)
     */

        ZooKeeper zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new CreateSession());
        System.out.println(zooKeeper.getState());

        // 计数工具类：CountDownLatch:不让main方法结束，让线程处于等待阻塞
        countDownLatch.await();

        System.out.println("客户端与服务端会话真正建立了");

    }

    /*
        回调方法：处理来自服务器端的watcher通知
     */
    public void process(WatchedEvent watchedEvent) {
        // SyncConnected
        if(watchedEvent.getState() == Event.KeeperState.SyncConnected){

            //解除主程序在CountDownLatch上的等待阻塞
            System.out.println("process方法执行了...");
            countDownLatch.countDown();

        }


    }
}

注意：ZooKeeper客户端和服务端会话的建立是一个异步的过程,也就是说在程序中,构造方法会在处理完客户端初始化工作后立即返回,在大多数情况下,此时并没有真正建立好一个可用的会话,在会话的生命周期中处于“ CONNECTING的状态。当该会话真正创建完毕后 ZooKeeper服务端会向会话对应的客户端发送一个事件通知,以告知客户端,客户端只有在获取这个通知之后,才算真正建立了会话

创建节点

节点介绍：

/**
* path ：节点创建的路径
* data[] ：节点创建要保存的数据，是个byte类型的
* acl ：节点创建的权限信息(4种类型)
*         ANYONE_ID_UNSAFE : 表示任何人
*         AUTH_IDS ：此ID仅可用于设置ACL。它将被客户机验证的ID替换。
*         OPEN_ACL_UNSAFE ：这是一个完全开放的ACL(常用)--> world:anyone
*         CREATOR_ALL_ACL ：此ACL授予创建者身份验证ID的所有权限
* createMode ：创建节点的类型(4种类型)
*        PERSISTENT：持久节点
*        PERSISTENT_SEQUENTIAL：持久顺序节点
*        EPHEMERAL：临时节点
*        EPHEMERAL_SEQUENTIAL：临时顺序节点
String node = zookeeper.create(path,data,acl,createMode);
*/

import org.apache.zookeeper.*;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

public class CreateNote implements Watcher {

    private  static CountDownLatch countDownLatch = new CountDownLatch(1);

    private static ZooKeeper zooKeeper;


    /*
      建立会话
     */
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {

     /*
        客户端可以通过创建一个zk实例来连接zk服务器
        new Zookeeper(connectString,sesssionTimeOut,Wather)
        connectString: 连接地址：IP：端口
        sesssionTimeOut：会话超时时间：单位毫秒
        Wather：监听器(当特定事件触发监听时，zk会通过watcher通知到客户端)
     */

         zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new CreateNote());
        System.out.println(zooKeeper.getState());

        // 计数工具类：CountDownLatch:不让main方法结束，让线程处于等待阻塞
        //countDownLatch.await();\
        Thread.sleep(Integer.MAX_VALUE);

    }



    /*
        回调方法：处理来自服务器端的watcher通知
     */
    public void process(WatchedEvent watchedEvent) {
        // SyncConnected
        if(watchedEvent.getState() == Event.KeeperState.SyncConnected){

            //解除主程序在CountDownLatch上的等待阻塞
            System.out.println("process方法执行了...");
            // 创建节点
            try {
                createNoteSync();
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }


    }


    /*
       创建节点的方法
   */
    private static void createNoteSync() throws KeeperException, InterruptedException {

        /**
         *  path        ：节点创建的路径
         *  data[]      ：节点创建要保存的数据，是个byte类型的
         *  acl         ：节点创建的权限信息(4种类型)
         *                 ANYONE_ID_UNSAFE    : 表示任何人
         *                 AUTH_IDS    ：此ID仅可用于设置ACL。它将被客户机验证的ID替换。
         *                 OPEN_ACL_UNSAFE    ：这是一个完全开放的ACL(常用)--> world:anyone
         *                 CREATOR_ALL_ACL  ：此ACL授予创建者身份验证ID的所有权限
         *  createMode    ：创建节点的类型(4种类型)
         *                  PERSISTENT：持久节点
         *                    PERSISTENT_SEQUENTIAL：持久顺序节点
         *                  EPHEMERAL：临时节点
         *                  EPHEMERAL_SEQUENTIAL：临时顺序节点
         String node = zookeeper.create(path,data,acl,createMode);
         */

        // 持久节点
        String note_persistent = zooKeeper.create("/lg-persistent", "持久节点内容".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        // 临时节点
        String note_ephemeral = zooKeeper.create("/lg-ephemeral", "临时节点内容".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

        // 持久顺序节点
        String note_persistent_sequential = zooKeeper.create("/lg-persistent_sequential", "持久顺序节点内容".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);

        System.out.println("创建的持久节点" + note_persistent);
        System.out.println("创建的临时节点" + note_ephemeral);
        System.out.println("创建的持久顺序节点" + note_persistent_sequential);

    }
}

读取节点：

public class GetNoteData implements Watcher {

    private  static CountDownLatch countDownLatch = new CountDownLatch(1);

    private static ZooKeeper zooKeeper;


    /*
      建立会话
     */
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {

     /*
        客户端可以通过创建一个zk实例来连接zk服务器
        new Zookeeper(connectString,sesssionTimeOut,Wather)
        connectString: 连接地址：IP：端口
        sesssionTimeOut：会话超时时间：单位毫秒
        Wather：监听器(当特定事件触发监听时，zk会通过watcher通知到客户端)
     */

         zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new GetNoteData());
        System.out.println(zooKeeper.getState());

        // 计数工具类：CountDownLatch:不让main方法结束，让线程处于等待阻塞
        //countDownLatch.await();\
        Thread.sleep(Integer.MAX_VALUE);

    }



    /*
        回调方法：处理来自服务器端的watcher通知
     */
    public void process(WatchedEvent watchedEvent) {

        /*
            子节点列表发生改变时，服务器端会发生noteChildrenChanged事件通知
            要重新获取子节点列表，同时注意：通知是一次性的，需要反复注册监听
         */
        if(watchedEvent.getType() == Event.EventType.NodeChildrenChanged){

            List<String> children = null;
            try {
                children = zooKeeper.getChildren("/lg-persistent", true);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(children);

        }



        // SyncConnected
        if(watchedEvent.getState() == Event.KeeperState.SyncConnected){

            //解除主程序在CountDownLatch上的等待阻塞
            System.out.println("process方法执行了...");

            // 获取节点数据的方法
            try {
                getNoteData();

                // 获取节点的子节点列表方法
                getChildrens();
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }


        }


    }

    /*
        获取某个节点的内容
     */
    private void getNoteData() throws KeeperException, InterruptedException {

        /**
         * path    : 获取数据的路径
         * watch    : 是否开启监听
         * stat    : 节点状态信息
         *        null: 表示获取最新版本的数据
         *  zk.getData(path, watch, stat);
         */
        byte[] data = zooKeeper.getData("/lg-persistent", false, null);
        System.out.println(new String(data));


    }


    /*
        获取某个节点的子节点列表方法
     */
    public static void getChildrens() throws KeeperException, InterruptedException {

        /*
            path:路径
            watch:是否要启动监听，当子节点列表发生变化，会触发监听
            zooKeeper.getChildren(path, watch);
         */
        List<String> children = zooKeeper.getChildren("/lg-persistent", true);
        System.out.println(children);



    }



}

删除节点

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

public class DeleteNote implements Watcher {

    private  static CountDownLatch countDownLatch = new CountDownLatch(1);

    private static ZooKeeper zooKeeper;


    /*
      建立会话
     */
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {

     /*
        客户端可以通过创建一个zk实例来连接zk服务器
        new Zookeeper(connectString,sesssionTimeOut,Wather)
        connectString: 连接地址：IP：端口
        sesssionTimeOut：会话超时时间：单位毫秒
        Wather：监听器(当特定事件触发监听时，zk会通过watcher通知到客户端)
     */

         zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new DeleteNote());
        System.out.println(zooKeeper.getState());

        // 计数工具类：CountDownLatch:不让main方法结束，让线程处于等待阻塞
        //countDownLatch.await();\
        Thread.sleep(Integer.MAX_VALUE);

    }



    /*
        回调方法：处理来自服务器端的watcher通知
     */
    public void process(WatchedEvent watchedEvent) {
        // SyncConnected
        if(watchedEvent.getState() == Event.KeeperState.SyncConnected){

            //解除主程序在CountDownLatch上的等待阻塞
            System.out.println("process方法执行了...");
            // 删除节点
            try {
                deleteNoteSync();
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }


        }


    }

    /*
        删除节点的方法
     */
    private void deleteNoteSync() throws KeeperException, InterruptedException {

        /*
          zooKeeper.exists(path,watch) :判断节点是否存在
          zookeeper.delete(path,version) : 删除节点
      */

        Stat stat = zooKeeper.exists("/lg-persistent/c1", false);
        System.out.println(stat == null ? "该节点不存在":"该节点存在");

        if(stat != null){
            zooKeeper.delete("/lg-persistent/c1",-1);
        }

        Stat stat2 = zooKeeper.exists("/lg-persistent/c1", false);
        System.out.println(stat2 == null ? "该节点不存在":"该节点存在");



    }


}

更新节点

/*
        更新数据节点内容的方法
     */
    private void updateNoteSync() throws KeeperException, InterruptedException {

         /*
            path:路径
            data:要修改的内容 byte[]
            version:为-1，表示对最新版本的数据进行修改
            zooKeeper.setData(path, data,version);
         */


        byte[] data = zooKeeper.getData("/lg-persistent", false, null);
        System.out.println("修改前的值：" + new String(data));

        //修改/lg-persistent 的数据 stat: 状态信息对象
        Stat stat = zooKeeper.setData("/lg-persistent", "客户端修改了节点数据".getBytes(), -1);

        byte[] data2 = zooKeeper.getData("/lg-persistent", false, null);
        System.out.println("修改后的值：" + new String(data2));

    }

countdownlatch apache

ErixHao

ErixHao

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

sql优化：数据量过大查询优化

如果数据量比较少，几十万数据 没什么问题。刚开始自己的想法是这样的：使用多线程的方式，因为查询每天的数据量很少，那么是不是可以使用多线程的方式，每个线程查询一天的，查询一个月30天，就用30个线程，这样速度会不会快些？于是，用多线程的方式实现了下。

ALiDan 2020-04-16

Redisson的CountDownLatch实现对比 VS Java的CountDownLat

核心接口连个方法，一个是countDown，另一个是await，countDown是改变状态的，await可以类比对锁的等待。Redisson版本的CountDownLatch设计思路和Redisson实现分布式锁类似，需要利用到Redis的订阅/发布来实

王道革 2020-04-11

【分布式锁】05-使用Redisson中Semaphore和CountDownLatch原理

剩下还有Semaphore和CountDownLatch两块，我们就趁热打铁，赶紧看看Redisson是如何实现的吧。我们在JDK中都知道Semaphore和CountDownLatch两兄弟，这里就不多赘述，不了解的可以再回头看看。第一次数据为0， 然后

sunzxh 2020-03-23

线程安全

线程安全就是多线程访问时，采用了加锁机制，当一个线程访问该类的某个数据时，进行保护，其他线程不能进行访问直到该线程读取完，其他线程才可使用。不会出现数据不一致或者数据污染。线程不安全就是不提供数据访问保护，有可能出现多个线程先后更改数据造成所得到的数据是脏

付春杰Blog 2020-02-17

CountDownLatch源码解析

CountDownLatch基于AQS实现的同步器，允许一个或者多个线程通过await()方法进入阻塞等待，直到一个或者多个线程执行countDown()完成。CountDownLatch在创建时需要传入一个count值，一旦某个或者多个线程调用了awai

spb 2020-02-01

CountDownLatch源码分析

CountDownLatch允许一个或多个线程等待直到在其他线程中执行的一组操作完成的同步辅助。简单来说，就是CountDownLatch内部维护了一个计数器，每个线程完成自己的操作之后都会将计数器减一，然后会在计数器的值变为0之前一直阻塞，直到计数器的值

xzkjgw 2020-01-17

java并发测试帮助方法

public static void run(int num, Consumer<Thread> action) {. CountDownLatch countDownLatch = new CountDownLatch(1);e.printS

abdstime 2020-01-17

SpringCloud分布式微服务b2b2c电子商务-在springboot中用redis实现消息队

准备阶段安装redis,可参考我的另一篇文章java 1.8maven 3.0idea环境依赖创建一个新的springboot工程，了解springcloud架构可以加求求：三五三六二四七二五九。在其pom文件,加入spring-boot-starter-

Cheetahcubs 2020-01-04

JUC包中的CountDownLatch源码实现分析

CountDownLatch是JUC包中提供的线程同步工具，使用CountDownLatch可以实现一个或多个线程等待直到一组操作在其他线程中被执行完成。CountDownLatch基于AQS实现,代码不多,所以很好分析。// Decrement coun

zhujiangtaotaise 2019-12-26

windows下 java使用zookeeper案例

import java.util.List;private ZooKeeper zookeeper;private static final int SESSION_TIME_OUT = 2000;private CountDownLatch countD

zhangxiaocc 2019-11-07

( 转载 )Zookeeper KeeperErrorCode = ConnectionLoss

error是在调用zk.exists（）时抛出的。于是开始google，发现问题是原因是：new zookeeper之后，zookeeper的还没有连接好，就去调用，当然会抛错。

Wisher 2017-06-07

Android 并发工具类与线程池

前言上一篇说到了Android并发编程中的 原子类与并发容器,那么本篇呢,继续上一篇记录一下Android并发编程中常用的一些工具类,以及面试必问知识点--线程池.并发工具类CountDownLatchCountDownLatch允许一个或多个线程等待其他

思捻如枫 2019-07-01

警报：线上事故之CountDownLatch的威力

2019.2.22号凌晨3点半，是一个让人难以忘怀的、和瑞哥最后一次一起奋战的夜晚。线上事故应用每天凌晨1点10分进行自清理重启后，会进行数据源连接池的初始化操作。总结不要过分相信监控指标等信息针对长耗时的业务，一定要做超时限制，不可无所谓的放任Count

leys 2019-07-01

Java并发多线程 - 并发工具类JUC

例如，应用程序的主线程希望在负责启动框架服务的线程已经启动所有框架服务之后执行。CountDownLatch是通过一个计数器来实现的，计数器的初始化值为线程的数量。当计数器到达0时，表示所有的线程都已完成任务，然后在闭锁上等待的线程就可以恢复执行任务。

讨厌什么变成什么 2019-06-30

Redis+Lua实现限流

分布式限流最关键的是要将限流服务做成原子化，而解决方案可以使使用redis+lua或者nginx+lua技术进行实现，通过这两种技术可以实现的高并发和高性能。首先我们来使用redis+lua实现时间窗内某个接口的请求数限流，实现了该功能后可以改造为限流总并

浪里xiao白龙 2019-06-28

线程间的同步与通信(8)——Semaphore源码分析

Semaphore（信号量）也是常用的并发工具之一，它常常用于流量控制。通常情况下，公共的资源常常是有限的，例如数据库的连接数。使用Semaphore可以帮助我们有效的管理这些有限资源的使用。Semaphore的结构和ReentrantLock以及Coun

mojianc 2019-06-28

线程间的同步与通信(7)——CyclicBarrier源码分析

上一篇 我们学习了基于AQS共享锁实现的CountDownLatch，本篇我们来看看另一个和它比较像的并发工具CyclicBarrier。与CountdownLatch的区别将count值递减的线程在CountDownLatch中，执行countDown方

liwenbocsu 2019-06-28

线程间的同步与通信(6)——CountDownLatch源码分析

CountDownLatch是一个很有用的工具，latch是门闩的意思，该工具是为了解决某些操作只能在一组操作全部执行完成后才能执行的情景。CountDown是倒数计数，所以CountDownLatch的用法通常是设定一个大于0的值，该值即代表需要等待的总

了缺 2019-06-28

SpringBoot非官方教程 | 第十四篇：在springboot中用redis实现消息队列

准备阶段安装redis,可参考我的另一篇文章，5分钟带你入门Redis。idea环境依赖创建一个新的springboot工程，在其pom文件,加入spring-boot-starter-data-redis依赖：。</dependency>创建

helowken 2019-06-25

android 简单线程同步之CountDownLatch

推荐安卓开发神器当涉及到多线程的时候，一个经常出现的问题就是线程同步，这里介绍android一个很简单的多线程同步方案-------------CountDownLatch. 我们直接看官方例子。这个例子有两个功能，doSomethingElse();

maplele 2011-12-30
ErixHao

ErixHao

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号