javaAPI操作zookeeper

2019年6月9日 241次阅读 来源: 先生_吕

【前言】zookeeper应用的场景主要有四种：通知服务、配置管理、集群管理、分布式锁。其还有一最大的特点就是数据同步。zookeeper任何一个节点上的数据都可以同步到其他ZK节点上。
业务需求，将服务器信息日志存储到zookeeper集群上，通过java操作zookeeper获取日志信息来展现并分析。
【准备】
zookeeper-3.4.5.tar.gz
【1】
开启zookeeper服务

《javaAPI操作zookeeper》

【2】
创建java工程，添加jar包（jar可以解压tar.gz包，在lib中获取）

《javaAPI操作zookeeper》

【3】
代码编写

ZK.java

import java.util.concurrent.CountDownLatch;

import org.apache.log4j.Logger;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;

public class ZK implements Watcher {
    private static Logger log = Logger.getLogger(ZK.class);

    // 缓存时间
    private static final int SESSION_TIME = 2000;
    protected ZooKeeper zookeeper;
    protected CountDownLatch countDownLatch = new CountDownLatch(1);

    // 链接zk集群
    public void connect(String hosts) throws Exception {
        zookeeper = new ZooKeeper(hosts, SESSION_TIME, this);
        countDownLatch.await();
    }

    @Override
    public void process(WatchedEvent event) {
        // TODO Auto-generated method stub
        if (event.getState() == KeeperState.SyncConnected) {
            countDownLatch.countDown();
        }
    }

    // 关闭集群
    public void close() throws InterruptedException {
        zookeeper.close();
    }

}

ZKOperator.java

import java.util.Arrays;
import java.util.List;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;

public class ZKOperator extends ZK {

    private static Logger logger = Logger.getLogger(ZKOperator.class);

    /**
     * 
     * <b>function:</b>创建持久态的znode,比支持多层创建.比如在创建/parent/child的情况下,无/parent.无法通过
     * 
     * @author lvfang
     * @createDate 2017-01-10 17:22:22
     * @param path
     * @param data
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void create(String path, byte[] data) throws KeeperException,
            InterruptedException {
        /**
         * 此处采用的是CreateMode是PERSISTENT 表示The znode will not be automatically
         */
        this.zookeeper.create(path, data, Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
    }

    /**
     * 
     * <b>function:</b>获取节点信息
     * 
     * @author lvfang
     * @createDate 2017-01-10 17:22:22
     * @param path
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void getChild(String path) throws KeeperException,
            InterruptedException {
        try {
            List<String> list = this.zookeeper.getChildren(path, false);
            if (list.isEmpty()) {
                logger.debug(path + "中没有节点");
            } else {
                logger.debug(path + "中存在节点");
                for (String child : list) {
                    logger.debug("节点为：" + child);
                }
            }
        } catch (KeeperException.NoNodeException e) {
            // TODO: handle exception
            throw e;

        }
    }

    public byte[] getData(String path) throws KeeperException,
            InterruptedException {
        return this.zookeeper.getData(path, false, null);
    }

    public static void main(String[] args) {
        try {
            ZKOperator zkoperator = new ZKOperator();
            zkoperator.connect("192.168.1.201");

            byte[] data = new byte[] { 'a', 'b', 'c', 'd' };

            //创建root目录，此root目录是创建在zookeeper中zoo.cfg配置中的path下
            zkoperator.create("/root", null);
            System.out.println(Arrays.toString(zkoperator.getData("/root")));

            //root下创建child1目录，并写入内容
            zkoperator.create("/root/child1", data);
            System.out.println(Arrays.toString(zkoperator
                    .getData("/root/child1")));

            zkoperator.create("/root/child2", data);
            System.out.println(Arrays.toString(zkoperator
                    .getData("/root/child2")));

            //root下创建child3目录，并写入内容
            String zktest = "ZooKeeper的Java API测试";
            zkoperator.create("/root/child3", zktest.getBytes());
            logger.debug("获取设置的信息："
                    + new String(zkoperator.getData("/root/child3")));

            //获取集群中的子节点信息
            System.out.println("节点孩子信息:");
            zkoperator.getChild("/root");

            //关闭zk
            zkoperator.close();

        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

《javaAPI操作zookeeper》 2017-01-10_172433.png

单个java文件完成zookeeper操作

import java.util.concurrent.CountDownLatch;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class TestZK implements Watcher {

    private static Logger logger = Logger.getLogger(ZKOperator.class);

    protected CountDownLatch countDownLatch = new CountDownLatch(1);

    @Override
    public void process(WatchedEvent event) {
        // TODO Auto-generated method stub
        if (event.getState() == KeeperState.SyncConnected) {
            countDownLatch.countDown();
        }
    }

    public static void main(String[] args) throws Exception {
        // 创建一个与服务器的连接 需要(服务端的 ip+端口号)(session过期时间)(Watcher监听注册)
        ZooKeeper zk = new ZooKeeper("192.168.1.201", 3000, new Watcher() {
            // 监控所有被触发的事件
            public void process(WatchedEvent event) {
                // TODO Auto-generated method stub
                System.out.println("已经触发了" + event.getType() + "事件！");
            }
        });

        // 创建一个目录节点
        /**
         * CreateMode: PERSISTENT (持续的，相对于EPHEMERAL，不会随着client的断开而消失)
         * PERSISTENT_SEQUENTIAL（持久的且带顺序的） EPHEMERAL (短暂的，生命周期依赖于client session)
         * EPHEMERAL_SEQUENTIAL (短暂的，带顺序的)
         */
        zk.create("/testRootPath", "testRootData".getBytes(),
                Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        // 创建一个子目录节点
        zk.create("/testRootPath/testChildPathOne",
                "testChildDataOne".getBytes(), Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
        System.out
                .println(new String(zk.getData("/testRootPath", false, null)));

        // 取出子目录节点列表
        System.out.println(zk.getChildren("/testRootPath", true));

        // 创建另外一个子目录节点
        zk.create("/testRootPath/testChildPathTwo",
                "testChildDataTwo".getBytes(), Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
        System.out.println(zk.getChildren("/testRootPath", true));

        // 修改子目录节点数据
        zk.setData("/testRootPath/testChildPathOne", "hahahahaha".getBytes(),
                -1);
        byte[] datas = zk.getData("/testRootPath/testChildPathOne", true, null);
        String str = new String(datas, "utf-8");
        System.out.println(str);

        // 删除整个子目录 -1代表version版本号，-1是删除所有版本
        zk.delete("/testRootPath/testChildPathOne", -1);
        System.out.println(zk.getChildren("/testRootPath", true));
        System.out.println(str);

    }

}

其他API方法

/**
          * create(): 发起一个create操作. 可以组合其他方法 (比如mode 或background) 最后以forPath()方法结尾

            delete(): 发起一个删除操作. 可以组合其他方法(version 或background) 最后以forPath()方法结尾

            checkExists(): 发起一个检查ZNode 是否存在的操作. 可以组合其他方法(watch 或background) 最后以forPath()方法结尾

            getData(): 发起一个获取ZNode数据的操作. 可以组合其他方法(watch, background 或get stat) 最后以forPath()方法结尾

            setData(): 发起一个设置ZNode数据的操作. 可以组合其他方法(version 或background) 最后以forPath()方法结尾

            getChildren(): 发起一个获取ZNode子节点的操作. 可以组合其他方法(watch, background 或get stat) 最后以forPath()方法结尾

            inTransaction(): 发起一个ZooKeeper事务. 可以组合create, setData, check, 和/或delete 为一个操作, 然后commit() 提交
          */
    原文作者：先生_吕
    原文地址: https://www.jianshu.com/p/0e3ca2f16dce
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞