Zookeeper3：Zookeeper-api的应用

作者: 勇于自信 | 来源:发表于2020-04-28 16:25 被阅读0次

org.apache.zookeeper.Zookeeper是客户端入口主类，负责建立与server的会话
它提供了表所示几类主要方法：
create 在本地目录树中创建一个节点
delete 删除一个节点
exists 测试本地是否存在目标节点
get/set data 从目标节点上读取 / 写数据
get/set ACL 获取 / 设置目标节点访问控制列表信息
get children 检索一个子节点上的列表
sync 等待要被传送的数据

1.java对zookeeper增删改查案例

增加：给根目录下新增/idea节点，增加内容：hellozk

package core;
import org.apache.zookeeper.*;
import org.junit.Before;
import org.junit.Test;

public class SimpleZkClinet {
    private static final String connectString="192.168.131.101:2181,192.168.131.102:2181,192.168.131.103:2181";
    private static final int sessionTimeout = 30000;
    ZooKeeper zkClinet = null;

    @Before
    public void init() throws Exception {
        zkClinet = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                //收到事件通知后的回调函数（应该是我们自己的事件处理逻辑）
                System.out.println(event.getType()+"---"+event.getPath());
            }
        });
    }

    /**
     * 数据的增加
     */
    @Test
    public void testCreate() throws Exception {
        String nodeCreated = zkClinet.create("/idea","hellozk".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("nodeCreated:"+nodeCreated);
    }
}

运行代码，报错：

问题原因：经排查防火墙已关闭，而且能和zookeeper的端口ping通，但是linux上的zookeeper是认master的，而本地不认master。
解决办法：在windows添加ip映射master

再次运行代码，没报错了，去linux上检查：
先启动zkCli.sh，启动客户端，输入命令：ls / 可以看到有idea，然后get /idea可以发现有我们插入的内容

查询：获取子节点

/**
     * 数据的查：获取子节点
     */
    @Test
    public void getChildren() throws Exception {
        List<String> children = zkClinet.getChildren("/", true);
        for(String child:children){
            System.out.println(child);
        }
      Thread.sleep(Long.MAX_VALUE);
    }

最后一处 Thread.sleep(Long.MAX_VALUE);代码，它的作用是不让进程结束，以便于监听。
运行代码，查询结果如下：

我们发现，第一行打印出的不是节点，而是我们上面代码的监听。
它的意思是：当我们更改/根节点时，会打印出节点的变化。但是以上代码，只监听一次。
如果想让它一直监听，则加入以下代码：

这样，每次根节点有更改都会一直监听。

查询：获取子节点

@Test
    public void testExist() throws Exception{
        Stat stat = zkClinet.exists("/idea", false);
        System.out.println(stat==null?"not exist":"exist");
    }

运行代码，输出如下：

查询：

@Test
    public void getData() throws Exception{
        //第二个参数：是否监听数据
        //第三个参数：获取哪个版本的数据，获取最新的，给null就可以了。
        byte[] data = zkClinet.getData("/idea", false, null);
        System.out.println(new String(data));
    }

运行代码，打印结果如下：
log4j:WARN No appenders could be found for logger (org.apache.zookeeper.ZooKeeper).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
hellozk

删除节点：

@Test
    public void deleteZnode()throws Exception{
        //参数2：指定要删除的版本，-1表示删除所有版本
        zkClinet.delete("/idea",-1);
    }

运行代码，再次运行查询代码，如下：
log4j:WARN No appenders could be found for logger (org.apache.zookeeper.ZooKeeper).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
qqq
cluster
controller_epoch
brokers
zookeeper
admin
isr_change_notification
consumers
config
可以发现，已经不存在了/idea节点。

修改：更改节点数据

@Test
    public void setData() throws Exception{
        //第二个参数：是否监听数据
        //第三个参数：获取哪个版本的数据，获取最新的，给null就可以了。
        zkClinet.setData("/qqq", "ccc".getBytes(),-1);
        byte[] data = zkClinet.getData("/qqq", false, null);
        System.out.println(new String(data));
    }

运行代码，打印如下：
log4j:WARN No appenders could be found for logger (org.apache.zookeeper.ZooKeeper).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
ccc
可以看到，修改成功。

2.zookeeper应用案例（分布式应用HA||分布式锁）

1.实现分布式应用的(主节点HA)及客户端动态更新主节点状态
某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线

A、客户端实现

public class AppClient {
      private String groupNode = "sgroup";
      private ZooKeeper zk;
      private Stat stat = new Stat();
      private volatile List<String> serverList;
/**
 * 连接zookeeper
 */
public void connectZookeeper() throws Exception {
      zk = new ZooKeeper("localhost:4180,localhost:4181,localhost:4182", 5000, new Watcher() {
      public void process(WatchedEvent event) {
      // 如果发生了"/sgroup"节点下的子节点变化事件, 更新server列表, 并重新注册监听
      if (event.getType() == EventType.NodeChildrenChanged
      && ("/" + groupNode).equals(event.getPath())) {
      try {
            updateServerList();
      } catch (Exception e) {
      e.printStackTrace();
      } }
      }
      });
      updateServerList();
      }
/**
 * 更新server列表
 */

private void updateServerList() throws Exception {
      List<String> newServerList = new ArrayList<String>();
      // 获取并监听groupNode的子节点变化
      // watch参数为true, 表示监听子节点变化事件.
      // 每次都需要重新注册监听, 因为一次注册, 只能监听一次事件, 如果还想继续保持监听, 必须重新注册
      List<String> subList = zk.getChildren("/" + groupNode, true);
      for (String subNode : subList) {
      // 获取每个子节点下关联的server地址
      byte[] data = zk.getData("/" + groupNode + "/" + subNode, false, stat);
      newServerList.add(new String(data, "utf-8"));
      }
      // 替换server列表
      serverList = newServerList;
      System.out.println("server list updated: " + serverList);
}
/**
 * client的工作逻辑写在这个方法中
 * 此处不做任何处理, 只让client sleep
 */
public void handle() throws InterruptedException {
      Thread.sleep(Long.MAX_VALUE);
}

public static void main(String[] args) throws Exception {
      AppClient ac = new AppClient();
      ac.connectZookeeper();
      ac.handle();
}
}

B、服务器端实现

public class AppServer {

private String groupNode = "sgroup";

private String subNode = "sub";



/**

 * 连接zookeeper

 * @param address server的地址

 */

public void connectZookeeper(String address) throws Exception {

ZooKeeper zk = new ZooKeeper(

"localhost:4180,localhost:4181,localhost:4182",

5000, new Watcher() {

public void process(WatchedEvent event) {

// 不做处理

}

});

// 在"/sgroup"下创建子节点

// 子节点的类型设置为EPHEMERAL_SEQUENTIAL, 表明这是一个临时节点, 且在子节点的名称后面加上一串数字后缀

// 将server的地址数据关联到新创建的子节点上

String createdPath = zk.create("/" + groupNode + "/" + subNode, address.getBytes("utf-8"),

Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

System.out.println("create: " + createdPath);

}



/**

 * server的工作逻辑写在这个方法中

 * 此处不做任何处理, 只让server sleep

 */

public void handle() throws InterruptedException {

Thread.sleep(Long.MAX_VALUE);

}



public static void main(String[] args) throws Exception {

// 在参数中指定server的地址

if (args.length == 0) {

System.err.println("The first argument must be server address");

System.exit(1);

}

（2）分布式共享锁的简单实现
客户端A
public class DistributedClient {
    // 超时时间
    private static final int SESSION_TIMEOUT = 5000;
    // zookeeper server列表
    private String hosts = "localhost:4180,localhost:4181,localhost:4182";
    private String groupNode = "locks";
    private String subNode = "sub";
    private ZooKeeper zk;
    // 当前client创建的子节点
    private String thisPath;
    // 当前client等待的子节点
    private String waitPath;
    private CountDownLatch latch = new CountDownLatch(1);
    /**
     * 连接zookeeper
     */
    public void connectZookeeper() throws Exception {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {
            public void process(WatchedEvent event) {
                try {
                    // 连接建立时, 打开latch, 唤醒wait在该latch上的线程
                    if (event.getState() == KeeperState.SyncConnected) {
                        latch.countDown();
                    }
                    // 发生了waitPath的删除事件
                    if (event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) {
                        doSomething();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        // 等待连接建立
        latch.await();
        // 创建子节点
        thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
        CreateMode.EPHEMERAL_SEQUENTIAL);
        // wait一小会, 让结果更清晰一些
        Thread.sleep(10);
        // 注意, 没有必要监听"/locks"的子节点的变化情况
        List<String> childrenNodes = zk.getChildren("/" + groupNode, false);
        // 列表中只有一个子节点, 那肯定就是thisPath, 说明client获得锁
        if (childrenNodes.size() == 1) {
            doSomething();
        } else {
            String thisNode = thisPath.substring(("/" + groupNode + "/").length());
            // 排序
            Collections.sort(childrenNodes);
            int index = childrenNodes.indexOf(thisNode);
            if (index == -1) {
                // never happened
            } else if (index == 0) {
                // inddx == 0, 说明thisNode在列表中最小, 当前client获得锁
                doSomething();
            } else {
                // 获得排名比thisPath前1位的节点
                this.waitPath = "/" + groupNode + "/" + childrenNodes.get(index - 1);
                // 在waitPath上注册监听器, 当waitPath被删除时, zookeeper会回调监听器的process方法
                zk.getData(waitPath, true, new Stat());
            }
        }
    }
    private void doSomething() throws Exception {
        try {
            System.out.println("gain lock: " + thisPath);
            Thread.sleep(2000);
            // do something
        } finally {
            System.out.println("finished: " + thisPath);
            // 将thisPath删除, 监听thisPath的client将获得通知
            // 相当于释放锁
            zk.delete(this.thisPath, -1);
        }
    }
    public static void main(String[] args) throws Exception {
        for (int i = 0; i < 10; i++) {
            new Thread() {
                public void run() {
                    try {
                        DistributedClient dl = new DistributedClient();
                        dl.connectZookeeper();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }.start();
        }
        Thread.sleep(Long.MAX_VALUE);
    }
}
AppServer as = new AppServer();
as.connectZookeeper(args[0]);
as.handle();
}
}

分布式多进程模式实现：

public class DistributedClientMy {
      // 超时时间
      private static final int SESSION_TIMEOUT = 5000;
      // zookeeper server列表
      private String hosts = "spark01:2181,spark02:2181,spark03:2181";
      private String groupNode = "locks";
      private String subNode = "sub";
      private boolean haveLock = false;
      private ZooKeeper zk;
      // 当前client创建的子节点
      private volatile String thisPath;
      /**
       * 连接zookeeper
       */
public void connectZookeeper() throws Exception {
      zk = new ZooKeeper("spark01:2181", SESSION_TIMEOUT, new Watcher() {
      public void process(WatchedEvent event) {
      try {
      // 子节点发生变化
      if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) {
      // thisPath是否是列表中的最小节点
      List<String> childrenNodes = zk.getChildren("/" + groupNode, true);
      String thisNode = thisPath.substring(("/" + groupNode + "/").length());
      // 排序
      Collections.sort(childrenNodes);
      if (childrenNodes.indexOf(thisNode) == 0) {
      doSomething();
      thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
      CreateMode.EPHEMERAL_SEQUENTIAL);
      }
      }
      } catch (Exception e) {
      e.printStackTrace();
      }
      }
      });
      // 创建子节点
      thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
      CreateMode.EPHEMERAL_SEQUENTIAL);
      // wait一小会, 让结果更清晰一些
      Thread.sleep(new Random().nextInt(1000));
      // 监听子节点的变化
      List<String> childrenNodes = zk.getChildren("/" + groupNode, true);
      // 列表中只有一个子节点, 那肯定就是thisPath, 说明client获得锁
      if (childrenNodes.size() == 1) {
      doSomething();
      thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
      CreateMode.EPHEMERAL_SEQUENTIAL);
      }
}

/**
 * 共享资源的访问逻辑写在这个方法中
 */
private void doSomething() throws Exception {
      try {
      System.out.println("gain lock: " + thisPath);
      Thread.sleep(2000);
      // do something
      } finally {
      System.out.println("finished: " + thisPath);
      // 将thisPath删除, 监听thisPath的client将获得通知
      // 相当于释放锁
      zk.delete(this.thisPath, -1);
      }
}
public static void main(String[] args) throws Exception {
      DistributedClientMy dl = new DistributedClientMy();
      dl.connectZookeeper();
      Thread.sleep(Long.MAX_VALUE);
}

Zookeeper3：Zookeeper-api的应用

1.java对zookeeper增删改查案例

2.zookeeper应用案例（分布式应用HA||分布式锁）

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Zookeeper