原生API操作Zookeeper

2018-06-19 本文已影响31人 deve_雨轩

image.png

Zookeeper系列文章

1.Zookeeper简介

2.Zookeeper集群安装

3.原生API操作Zookeeper

4.zkClient框架操作Zookeeper

5.Curator框架操作Zookeeper

原生API操作Zookeeper

首先要使用 java 操作 zookeeper, zookeeper 的 javaclient 使我们更轻松的去对 zookeeper 进行各种操作，我们引入 zookeeper-3.3.4. Jar 和 zkclient-0.1. Jar 即可。
zookeeper-3.3.4. Jar 为官方提供的 javaAPI, zkclient-0.1. Jar 则为在源生 api 基础之上进行扩展的开源 JAVA 客户端。

zookeeper-3.3.4. Jar API介绍

创建会话方法：客户端可以通过创建一个 zookeeper 实例来连接 zookeeper 服务器。

    ZooKeeper zk = new ZooKeeper();  //后面会有样例详细介绍怎么使用
     //Zookeeper类一共了 4 个构造方法，根据参数不同,参数说明如下：
    //connectString：连接服务器列表，用”，“分割。
    //sessionTimeout：心跳检测时间周期（毫秒）
    //wather：事件处理通知器。
    //canBeReadOnly：标识当前会话是否支持只读。
    //sessionld 和 sessionPasswd：提供连接 zookeeper 的 sessionld 和密码，通过这俩个确定唯一一台客户端，目的是可以提供重复会话。

创建节点（znode）方法：create:
- 提供了两套创建节点的方法，同步和异步创建节点方式。
- 同步方式：
  - 参数 1，节点路径（名称）: /nodeName（不允许递归创建节点，也就是说在父节点不存在的情况下，不允许创建子节点）
  - 参数 2，节点内容：要求类型是字节数组（也就是说，不支持序列化方式，如果需要实现序列化，可使用 java 相关序列化框架，如 Hessian、Kryo 框架）
  - 参数 3，节点权限：使用 Ids. OPEN_ ACL UNSAFE 开放权限即可。（这个参数一般在权限没有太高要求的场景下，没必要关注）
  - 参数 4，节点类型：创建节点的类型：CreateMode.“，提供四种节点类型
    - PERSISTENT（持久节点）
    - PERSISTENT SEQUENTIAL（持久顺序节点）
    - EPHEMERAL（临时节点）
    - EPHEMERAL SEQUENTIAL（临时顺序节点）
- 异步方式：（在同步参数基础上增加俩个参数）
  - 参数 5，注册一个异步回调函数，要实现 AsynCallBack. StringCallBack 接口，重写processResult (int rc, String path, Object ctx, String name）方法，当节点创建完毕后执行此方法。
    - rc：为服务端响应码 0 表示调用成功、4 表示端口连接、-110 表示指定节点存在、-112 表示会话已经过期。
    - path：接口调用时传入 API 的数据节点的路径参数
    - ctx：为调用接口传入 API 的 ctx 值
    - name：实际在服务器端创建节点的名称
- 参数 6，传递给回调函数的参数，一般为上下文（Context）信息
删除节点：delete 方法（api 提供了两个接口：同步删除和异步删除方式）
- 同步方式：
  - 参数1，节点名称/deletePath
  - 参数 2，版本号，即表明本次删除操作是针对该数据的某个版本进行的操作。
- 异步方式：（也是在同步方法基础上增加两个参数,使用方式和 create 方法一致）
  - 参数 3:，一个异步回调函数
  - 参数 4: 用于传递上下文信息的对象。

注意：在 zookeeper 中，只允许删除叶子节点信息，也就是说如果当前节点不是叶子节点则无法删除，或必须先删除其下所有子节点。

样例程序(创建会话,创建节点,删除节点)


import java.util.concurrent.CountDownLatch;

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

/**
 * Zookeeper base学习笔记
 */
public class ZookeeperBase {

    /** zookeeper地址 */
    static final String CONNECT_ADDR = "192.168.0.160:2181,192.168.0.161:2181,192.168.0.162:2181";
    /** session超时时间 */
    static final int SESSION_OUTTIME = 2000;//ms 
    /** 信号量，阻塞程序执行，用于等待zookeeper连接成功，发送成功信号 */
    static final CountDownLatch connectedSemaphore = new CountDownLatch(1);
    
    public static void main(String[] args) throws Exception{
        
        ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher(){
            @Override
            public void process(WatchedEvent event) {
                //获取事件的状态
                KeeperState keeperState = event.getState();
                EventType eventType = event.getType();
                //如果是建立连接
                if(KeeperState.SyncConnected == keeperState){
                    if(EventType.None == eventType){
                        //如果建立连接成功，则发送信号量，让后续阻塞程序向下执行
                        connectedSemaphore.countDown();
                        System.out.println("zk 建立连接");
                    }
                }
            }
        });

        //进行阻塞
        connectedSemaphore.await();
        
        System.out.println("..");
        //创建父节点
//      zk.create("/testRoot", "testRoot".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        
        //创建子节点
//      zk.create("/testRoot/children", "children data".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        
        //获取节点洗信息
//      byte[] data = zk.getData("/testRoot", false, null);
//      System.out.println(new String(data));
//      System.out.println(zk.getChildren("/testRoot", false));
        
        //修改节点的值
//      zk.setData("/testRoot", "modify data root".getBytes(), -1);
//      byte[] data = zk.getData("/testRoot", false, null);
//      System.out.println(new String(data));       
        
        //判断节点是否存在
//      System.out.println(zk.exists("/testRoot/children", false));
        //删除节点
//      zk.delete("/testRoot/children", -1);
//      System.out.println(zk.exists("/testRoot/children", false));
        
        zk.close(); 
    }   
}

GetChildren 读取数据方法：包括子节点列表的获取和子节点数据的获取。
- 参数 1，path：获取指定节点的下的数据（获取子节点列表）
- 参数 2，watcher：注册的 watcher，一旦在本次子节点获取后，子节点列表发生变更的话，那么就会向客户端发送通知。该参数允许为 null。(如果为null,但wath参数为true时,会去取实例化ZooKeeper()连接对象时设置的那个watcher)
- 参数 3，wath：表明是否需要注册一个 watcher；如果为 true，则会使用到 zookeeper 客户端上文中提到的那个默认 watcher。如果为 false，则表明不需要注册 Watcher。
- 参数 4，cb：回调函数。
- 参数 5，ctx：上下文信息对象。
- 参数 6，stat：指定数据节点的节点状态信息。

注意：当我们获取指定节点的子节点列表后，还需要订阅这个子节点列表的变化通知，这时候就可以通过注册一个 watcher 来实现，当子节点被添加或删除时，服务器端就会触发一个“NodeChildrenChanged“类型的事件通知，需要注意的是服务器端发送给客户端的事件通知中，是不包含最新的节点列表的，客户端必须主动从新进行获取，通常在客户端收到这个事件通知后，就可以再次主动获取最新的子节点列表了。也就是说，zookeeper 服务端在向客户端发送 watcher"NodeChildrenChanged“事件通知的时候，仅仅只发了一个通知，不会把节点变化情况发给客户端，需要客户端自己重新获取，另外 Watcher 通知是一次性的，即触发后失效，因此客户端需要反复注册 Watcher 才行。

getData 方法：获取指定节点的数据内容。
- 参数 1，path：路径
- 参数 2，watcher：注册的 watcher 对象。一旦之后节点内容有变更，则会像客户端发送通知，该参数允许为 null。
- 参数 3，stat：指定节点的状态信息。
- 参数 4，watch：是否使用 watcher，如果为 true 则使用默认上文中的 watcher, false 则不使用 watcher。
- 参数 5，cb：回调函数。
- 参数 6，ctx：用于传递的下文信息对象。

注意：该方法和 getChildren 方法基本相同，主要是注册的 watcher 有所不同，客户端在获取一个阶段数据内容时，是可以进行 watcher 注册的，一旦节点发生变更，则服务器端会发送给客户端一个“NodeDataChanged“的事件通知。

setData 方法：修改指定节点的数据内容。
- 参数 1, path：路径。
- 参数 2, data：数据内容。
- 参数 3，版本号（-1 覆盖之前所有的版本）
- 参数 4，cb：回调函数。
- 参数 5，ctx：用于传递的下文信息对象。
exists 方法：检测节点是否存在。
- 参数 1，path：路径
- 参数 2，watcher：注册的 watcher 对象。一旦之后节点内容有变更，则会像客户端发送通知，该参数允许为 null。（用于三类事件监听：节点的创建、删除、更新）
- 参数 3，watch：是否使用 watcher，如果为 true 则使用默认上文中的 watcher, false 则不使用 watcher。
- 参数 4，cb：回调函数。
- 参数 5，ctx：用于传递的下文信息对象。

注意：exists 方法意义在于无论节点是否存在，都可以进行注册 watcher，能够对节点的创建、删除和修改进行监听，但是其子节点发送各种变化，都不会通知客户端。

Zookeeper 有 watch 事件，是一次性触发的，当 watch 监视的数据发生变化时，通知设置了该 watch 的 client，即 watcher.
同样，其 watcher 是监听数据发送了某些变化，那就一定会有对应的事件类型，和状态类型。
- 事件类型： (znode 节点相关的）
  - EventType. NodeCreated
  - EventType. NodeDataChanged
  - EventType. NodeChildrenChanged
  - EventType. NodeDeleted
- 状态类型：（是跟客户端实例相关的）
  - KeeperState. Disconnected
  - KeeperState. SyncConnected
  - KeeperState. AuthFailed
  - KeeperState. Expired
Watcher 的特性：一次性、客户端串行执行、轻量。
- 一次性：对于 ZK 的 watcher，你只需要记住一~点：zookeeper 有 watch 事件，是一次性触发的，当 watch 监视的数据发生变化时，通知设置了该 watch 的 client，即 watcher，由于 zookeeper 的监控都是一次性的所以每次必须设置监控。
- 客户端串行执行：客户端 Watcher 回调的过程是一个串行同步的过程，这为我们保证了顺序，同时需要开发人员注意一点，千万不要因为一个 Watcher 的处理逻辑影响了整个客户端的 Watcher 回调。
- 轻量：WatchedEvent 是 Zookeeper 整个 Watcher 通知机制的最小通知单元，整个结构只包含三部分：通知状态、事件类型和节点路径。也就是说 Watcher 通知非常的简单，只会告诉客户端发生了事件而不会告知其具体内容，需要客户自己去进行获取，比如 NodeDataChanged 事件，Zookeeper 只会通知客户端指定节点的数据发生了变更，而不会直接提供具体的数据内容。

我们通过一个示例，详细学习下 Watcher 的概念和其目的。Watcher 示例：


import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

/**
 * Zookeeper Wathcher 
 * 本类就是一个Watcher类（实现了org.apache.zookeeper.Watcher类）
 */
public class ZooKeeperWatcher implements Watcher {

    /** 定义原子变量 */
    AtomicInteger seq = new AtomicInteger();
    /** 定义session失效时间 */
    private static final int SESSION_TIMEOUT = 10000;
    /** zookeeper服务器地址 */
    private static final String CONNECTION_ADDR = "192.168.0.160:2181";
    /** zk父路径设置 */
    private static final String PARENT_PATH = "/testWatch";
    /** zk子路径设置 */
    private static final String CHILDREN_PATH = "/testWatch/children";
    /** 进入标识 */
    private static final String LOG_PREFIX_OF_MAIN = "【Main】";
    /** zk变量 */
    private ZooKeeper zk = null;
    /** 信号量设置，用于等待zookeeper连接建立之后 通知阻塞程序继续向下执行 */
    private CountDownLatch connectedSemaphore = new CountDownLatch(1);

    /**
     * 创建ZK连接
     * @param connectAddr ZK服务器地址列表
     * @param sessionTimeout Session超时时间
     */
    public void createConnection(String connectAddr, int sessionTimeout) {
        this.releaseConnection();
        try {
            zk = new ZooKeeper(connectAddr, sessionTimeout, this);
            System.out.println(LOG_PREFIX_OF_MAIN + "开始连接ZK服务器");
            connectedSemaphore.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 关闭ZK连接
     */
    public void releaseConnection() {
        if (this.zk != null) {
            try {
                this.zk.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 创建节点
     * @param path 节点路径
     * @param data 数据内容
     * @return 
     */
    public boolean createPath(String path, String data) {
        try {
            //设置监控(由于zookeeper的监控都是一次性的所以 每次必须设置监控)
            this.zk.exists(path, true);
            System.out.println(LOG_PREFIX_OF_MAIN + "节点创建成功, Path: " + 
                               this.zk.create(  /**路径*/ 
                                                path, 
                                                /**数据*/
                                                data.getBytes(), 
                                                /**所有可见*/
                                                Ids.OPEN_ACL_UNSAFE, 
                                                /**永久存储*/
                                                CreateMode.PERSISTENT ) +   
                               ", content: " + data);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    /**
     * 读取指定节点数据内容
     * @param path 节点路径
     * @return
     */
    public String readData(String path, boolean needWatch) {
        try {
            return new String(this.zk.getData(path, needWatch, null));
        } catch (Exception e) {
            e.printStackTrace();
            return "";
        }
    }

    /**
     * 更新指定节点数据内容
     * @param path 节点路径
     * @param data 数据内容
     * @return
     */
    public boolean writeData(String path, String data) {
        try {
            System.out.println(LOG_PREFIX_OF_MAIN + "更新数据成功，path：" + path + ", stat: " +
                                this.zk.setData(path, data.getBytes(), -1));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 删除指定节点
     * 
     * @param path
     *            节点path
     */
    public void deleteNode(String path) {
        try {
            this.zk.delete(path, -1);
            System.out.println(LOG_PREFIX_OF_MAIN + "删除节点成功，path：" + path);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 判断指定节点是否存在
     * @param path 节点路径
     */
    public Stat exists(String path, boolean needWatch) {
        try {
            return this.zk.exists(path, needWatch);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 获取子节点
     * @param path 节点路径
     */
    private List<String> getChildren(String path, boolean needWatch) {
        try {
            return this.zk.getChildren(path, needWatch);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 删除所有节点
     */
    public void deleteAllTestPath() {
        if(this.exists(CHILDREN_PATH, false) != null){
            this.deleteNode(CHILDREN_PATH);
        }
        if(this.exists(PARENT_PATH, false) != null){
            this.deleteNode(PARENT_PATH);
        }       
    }
    
    /**
     * 收到来自Server的Watcher通知后的处理。
     */
    @Override
    public void process(WatchedEvent event) {
        
        System.out.println("进入 process 。。。。。event = " + event);
        
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        if (event == null) {
            return;
        }
        
        // 连接状态
        KeeperState keeperState = event.getState();
        // 事件类型
        EventType eventType = event.getType();
        // 受影响的path
        String path = event.getPath();
        
        String logPrefix = "【Watcher-" + this.seq.incrementAndGet() + "】";

        System.out.println(logPrefix + "收到Watcher通知");
        System.out.println(logPrefix + "连接状态:\t" + keeperState.toString());
        System.out.println(logPrefix + "事件类型:\t" + eventType.toString());

        if (KeeperState.SyncConnected == keeperState) {
            // 成功连接上ZK服务器
            if (EventType.None == eventType) {
                System.out.println(logPrefix + "成功连接上ZK服务器");
                connectedSemaphore.countDown();
            } 
            //创建节点
            else if (EventType.NodeCreated == eventType) {
                System.out.println(logPrefix + "节点创建");
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                this.exists(path, true);
            } 
            //更新节点
            else if (EventType.NodeDataChanged == eventType) {
                System.out.println(logPrefix + "节点数据更新");
                System.out.println("我看看走不走这里........");
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(logPrefix + "数据内容: " + this.readData(PARENT_PATH, true));
            } 
            //更新子节点
            else if (EventType.NodeChildrenChanged == eventType) {
                System.out.println(logPrefix + "子节点变更");
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(logPrefix + "子节点列表：" + this.getChildren(PARENT_PATH, true));
            } 
            //删除节点
            else if (EventType.NodeDeleted == eventType) {
                System.out.println(logPrefix + "节点 " + path + " 被删除");
            }
            else ;
        } 
        else if (KeeperState.Disconnected == keeperState) {
            System.out.println(logPrefix + "与ZK服务器断开连接");
        } 
        else if (KeeperState.AuthFailed == keeperState) {
            System.out.println(logPrefix + "权限检查失败");
        } 
        else if (KeeperState.Expired == keeperState) {
            System.out.println(logPrefix + "会话失效");
        }
        else ;

        System.out.println("--------------------------------------------");

    }

    /**
     * <B>方法名称：</B>测试zookeeper监控<BR>
     * <B>概要说明：</B>主要测试watch功能<BR>
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {

        //建立watcher
        ZooKeeperWatcher zkWatch = new ZooKeeperWatcher();
        //创建连接
        zkWatch.createConnection(CONNECTION_ADDR, SESSION_TIMEOUT);
        //System.out.println(zkWatch.zk.toString());
        
        Thread.sleep(1000);
        
        // 清理节点
        zkWatch.deleteAllTestPath();
        
        if (zkWatch.createPath(PARENT_PATH, System.currentTimeMillis() + "")) {
            
            Thread.sleep(1000);
            
            
            // 读取数据
            System.out.println("---------------------- read parent ----------------------------");
            //zkWatch.readData(PARENT_PATH, true);
            
            // 读取子节点
            System.out.println("---------------------- read children path ----------------------------");
            zkWatch.getChildren(PARENT_PATH, true);

            // 更新数据
            zkWatch.writeData(PARENT_PATH, System.currentTimeMillis() + "");
            
            Thread.sleep(1000);
            
            // 创建子节点
            zkWatch.createPath(CHILDREN_PATH, System.currentTimeMillis() + "");
            
            Thread.sleep(1000);
            
            zkWatch.writeData(CHILDREN_PATH, System.currentTimeMillis() + "");
        }
        
        Thread.sleep(50000);
        // 清理节点
        zkWatch.deleteAllTestPath();
        Thread.sleep(1000);
        zkWatch.releaseConnection();
    }
}

我们再通过一个示例，模拟分布式情况下修改节点数据时,多个服务监听此节点,示例：

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class ZKWatcher implements Watcher {

    /** zk变量 */
    private ZooKeeper zk = null;
    
    /** 父节点path */
    static final String PARENT_PATH = "/super";
    
    /** 信号量设置，用于等待zookeeper连接建立之后 通知阻塞程序继续向下执行 */
    private CountDownLatch connectedSemaphore = new CountDownLatch(1);
    
    private List<String> cowaList = new CopyOnWriteArrayList<String>();
    
    
    /** zookeeper服务器地址 */
    public static final String CONNECTION_ADDR = "192.168.0.160:2181,192.168.0.161:2181,192.168.0.162:2181";
    /** 定义session失效时间 */
    public static final int SESSION_TIMEOUT = 30000;
    
    public ZKWatcher() throws Exception{
        zk = new ZooKeeper(CONNECTION_ADDR, SESSION_TIMEOUT, this);
        System.out.println("开始连接ZK服务器");
        connectedSemaphore.await();
    }


    @Override
    public void process(WatchedEvent event) {
        // 连接状态
        KeeperState keeperState = event.getState();
        // 事件类型
        EventType eventType = event.getType();
        // 受影响的path
        String path = event.getPath();
        System.out.println("受影响的path : " + path);
        
        
        if (KeeperState.SyncConnected == keeperState) {
            // 成功连接上ZK服务器
            if (EventType.None == eventType) {
                System.out.println("成功连接上ZK服务器");
                connectedSemaphore.countDown();
                try {
                    if(this.zk.exists(PARENT_PATH, false) == null){
                        this.zk.create(PARENT_PATH, "root".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);      
                    }
                    List<String> paths = this.zk.getChildren(PARENT_PATH, true);
                    for (String p : paths) {
                        System.out.println(p);
                        this.zk.exists(PARENT_PATH + "/" + p, true);
                    }
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }       
            } 
            //创建节点
            else if (EventType.NodeCreated == eventType) {
                System.out.println("节点创建");
                try {
                    this.zk.exists(path, true);
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            } 
            //更新节点
            else if (EventType.NodeDataChanged == eventType) {
                System.out.println("节点数据更新");
                try {
                    //update nodes  call function
                    this.zk.exists(path, true);
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            } 
            //更新子节点
            else if (EventType.NodeChildrenChanged == eventType) {
                System.out.println("子节点 ... 变更");
                try {
                    List<String> paths = this.zk.getChildren(path, true);
                    if(paths.size() >= cowaList.size()){
                        paths.removeAll(cowaList);
                        for(String p : paths){
                            this.zk.exists(path + "/" + p, true);
                            //this.zk.getChildren(path + "/" + p, true);
                            System.out.println("这个是新增的子节点 : " + path + "/" + p);
                            //add new nodes  call function
                        }
                        cowaList.addAll(paths);
                    } else {
                        cowaList = paths;
                    }
                    System.out.println("cowaList: " + cowaList.toString());
                    System.out.println("paths: " + paths.toString());
                    
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            } 
            //删除节点
            else if (EventType.NodeDeleted == eventType) {
                System.out.println("节点 " + path + " 被删除");
                try {
                    //delete nodes  call function
                    this.zk.exists(path, true);
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
            else ;
        } 
        else if (KeeperState.Disconnected == keeperState) {
            System.out.println("与ZK服务器断开连接");
        } 
        else if (KeeperState.AuthFailed == keeperState) {
            System.out.println("权限检查失败");
        } 
        else if (KeeperState.Expired == keeperState) {
            System.out.println("会话失效");
        }
        else ;

        System.out.println("--------------------------------------------");
    }
}

实例两个上面ZKWatcher对象,ZKWatcher其实就是简单了实现了监听了/super数据节点,你可以把这想象成两个不同的服务

public class Client1 {

    public static void main(String[] args) throws Exception{
        
        ZKWatcher myWatcher = new ZKWatcher();
        Thread.sleep(100000000);
    }
}

public class Client2 {

    public static void main(String[] args) throws Exception{
        
        ZKWatcher myWatcher = new ZKWatcher();
        Thread.sleep(100000000);
    }
}

通过下面这个小测试,测试对/super数据节点不同的操作,可以看到上面两个小程序会分别打印对应的信息

import java.util.concurrent.CountDownLatch;

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

public class Test {


    /** zookeeper地址 */
    static final String CONNECT_ADDR = "192.168.1.106:2181,192.168.1.107:2181,192.168.1.108:2181";
    /** session超时时间 */
    static final int SESSION_OUTTIME = 2000;//ms 
    /** 信号量，阻塞程序执行，用于等待zookeeper连接成功，发送成功信号 */
    static final CountDownLatch connectedSemaphore = new CountDownLatch(1);
    
    public static void main(String[] args) throws Exception{
        
        ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher(){
            @Override
            public void process(WatchedEvent event) {
                //获取事件的状态
                KeeperState keeperState = event.getState();
                EventType eventType = event.getType();
                //如果是建立连接
                if(KeeperState.SyncConnected == keeperState){
                    if(EventType.None == eventType){
                        //如果建立连接成功，则发送信号量，让后续阻塞程序向下执行
                        connectedSemaphore.countDown();
                        System.out.println("zk 建立连接");
                    }
                }
            }
        });

        //进行阻塞
        connectedSemaphore.await();
        
//      //创建子节点
//      zk.create("/super/c1", "c1".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        //创建子节点
//      zk.create("/super/c2", "c2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        //创建子节点
        zk.create("/super/c3", "c3".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        //创建子节点
//      zk.create("/super/c4", "c4".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        
//      zk.create("/super/c4/c44", "c44".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        
        //获取节点信息
//      byte[] data = zk.getData("/testRoot", false, null);
//      System.out.println(new String(data));
//      System.out.println(zk.getChildren("/testRoot", false));
        
        //修改节点的值
//      zk.setData("/super/c1", "modify c1".getBytes(), -1);
//      zk.setData("/super/c2", "modify c2".getBytes(), -1);
//      byte[] data = zk.getData("/super/c2", false, null);
//      System.out.println(new String(data));       
        
//      //判断节点是否存在
//      System.out.println(zk.exists("/super/c3", false));
//      //删除节点
//      zk.delete("/super/c3", -1);
        zk.close(); 
    }
}

Zookeeperd的ACL
- ACL (Access ControlList），Zookeeper 作为一个分布式协调框架，其内部存储的都是一些关乎分布式系统运行时状态的元数据，尤其是设计到一些分布式锁、Master 选举和协调等应用场景。我们需要有效地保障 zookeeper 中的数据安全，Zookeeper 提供一套完善的 ACL 权限控制机制来保障数据的安全。ZK 提供了三种模式。权限模式、授权对象、权限。
  - 权限模式：Scheme，开发人员最多使用的如下四种权限模式：
    - IP: ip 模式通过 ip 地址粒度来进行控制权限，例如配置了：ip:192.168.1.107 即表示权限控制都是针对这个 ip 地址的，同时也支持按网段分配，比如 192.168.1. *
    - Digest: digest 是最常用的权限控制模式，也更符合我们对权限控制的认识，其类似于_ "usermame: password“形式的权限标识进行权限配置。ZK 会对形成的权限标识先后进行俩次编码处理，分别是 SHA-1 加密算法、BASE64 编码。
    - World: World 是一直最开放的权限控制模式。这种模式可以看做为特殊的 Digest，他仅仅是一个标识而已。
    - Super：超级用户模式，在超级用户模式下可以对 ZK 任意进行操作。
- 授权对象：指的是权限赋予的用户或者一个指定的实体，例如 ip 地址或机器等。在不同的模式下，授权对象是不同的。这种模式和权限对象一一对应。
- 权限：权限就是指那些通过权限检测后可以被允许执行的操作，在 ZK 中，对数据的操作权限分为头下五大类：
  - CREATE、DELETE、READ、WRITE、ADMIN

我们通过一个示例，详细学习下 Auth 的概念和其目的。Auth 示例： [ZooKeeperAuth]

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
/**
 * Zookeeper 节点授权
 */
public class ZookeeperAuth implements Watcher {

    /** 连接地址 */
    final static String CONNECT_ADDR = "192.168.0.160:2181";
    /** 测试路径 */
    final static String PATH = "/testAuth";
    final static String PATH_DEL = "/testAuth/delNode";
    /** 认证类型 */
    final static String authentication_type = "digest";
    /** 认证正确方法 */
    final static String correctAuthentication = "123456";
    /** 认证错误方法 */
    final static String badAuthentication = "654321";
    
    static ZooKeeper zk = null;
    /** 计时器 */
    AtomicInteger seq = new AtomicInteger();
    /** 标识 */
    private static final String LOG_PREFIX_OF_MAIN = "【Main】";
    
    private CountDownLatch connectedSemaphore = new CountDownLatch(1);
    
    @Override
    public void process(WatchedEvent event) {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (event==null) {
            return;
        }
        // 连接状态
        KeeperState keeperState = event.getState();
        // 事件类型
        EventType eventType = event.getType();
        // 受影响的path
        String path = event.getPath();
        
        String logPrefix = "【Watcher-" + this.seq.incrementAndGet() + "】";

        System.out.println(logPrefix + "收到Watcher通知");
        System.out.println(logPrefix + "连接状态:\t" + keeperState.toString());
        System.out.println(logPrefix + "事件类型:\t" + eventType.toString());
        if (KeeperState.SyncConnected == keeperState) {
            // 成功连接上ZK服务器
            if (EventType.None == eventType) {
                System.out.println(logPrefix + "成功连接上ZK服务器");
                connectedSemaphore.countDown();
            } 
        } else if (KeeperState.Disconnected == keeperState) {
            System.out.println(logPrefix + "与ZK服务器断开连接");
        } else if (KeeperState.AuthFailed == keeperState) {
            System.out.println(logPrefix + "权限检查失败");
        } else if (KeeperState.Expired == keeperState) {
            System.out.println(logPrefix + "会话失效");
        }
        System.out.println("--------------------------------------------");
    }
    /**
     * 创建ZK连接
     * 
     * @param connectString
     *            ZK服务器地址列表
     * @param sessionTimeout
     *            Session超时时间
     */
    public void createConnection(String connectString, int sessionTimeout) {
        this.releaseConnection();
        try {
            zk = new ZooKeeper(connectString, sessionTimeout, this);
            //添加节点授权
            zk.addAuthInfo(authentication_type,correctAuthentication.getBytes());
            System.out.println(LOG_PREFIX_OF_MAIN + "开始连接ZK服务器");
            //倒数等待
            connectedSemaphore.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    /**
     * 关闭ZK连接
     */
    public void releaseConnection() {
        if (this.zk!=null) {
            try {
                this.zk.close();
            } catch (InterruptedException e) {
            }
        }
    }
    
    /**
     * 
     * <B>方法名称：</B>测试函数<BR>
     * <B>概要说明：</B>测试认证<BR>
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        
        ZookeeperAuth testAuth = new ZookeeperAuth();
        testAuth.createConnection(CONNECT_ADDR,2000);
        List<ACL> acls = new ArrayList<ACL>(1);
        for (ACL ids_acl : Ids.CREATOR_ALL_ACL) {
            acls.add(ids_acl);
        }

        try {
            zk.create(PATH, "init content".getBytes(), acls, CreateMode.PERSISTENT);
            System.out.println("使用授权key：" + correctAuthentication + "创建节点："+ PATH + ", 初始内容是: init content");
        } catch (Exception e) {
            e.printStackTrace();
        }
        try {
            zk.create(PATH_DEL, "will be deleted! ".getBytes(), acls, CreateMode.PERSISTENT);
            System.out.println("使用授权key：" + correctAuthentication + "创建节点："+ PATH_DEL + ", 初始内容是: init content");
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 获取数据
        getDataByNoAuthentication();
        getDataByBadAuthentication();
        getDataByCorrectAuthentication();

        // 更新数据
        updateDataByNoAuthentication();
        updateDataByBadAuthentication();
        updateDataByCorrectAuthentication();

        // 删除数据
        deleteNodeByBadAuthentication();
        deleteNodeByNoAuthentication();
        deleteNodeByCorrectAuthentication();
        //
        Thread.sleep(1000);
        
        deleteParent();
        //释放连接
        testAuth.releaseConnection();
    }
    /** 获取数据：采用错误的密码 */
    static void getDataByBadAuthentication() {
        String prefix = "[使用错误的授权信息]";
        try {
            ZooKeeper badzk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            //授权
            badzk.addAuthInfo(authentication_type,badAuthentication.getBytes());
            Thread.sleep(2000);
            System.out.println(prefix + "获取数据：" + PATH);
            System.out.println(prefix + "成功获取数据：" + badzk.getData(PATH, false, null));
        } catch (Exception e) {
            System.err.println(prefix + "获取数据失败，原因：" + e.getMessage());
        }
    }

    /** 获取数据：不采用密码 */
    static void getDataByNoAuthentication() {
        String prefix = "[不使用任何授权信息]";
        try {
            System.out.println(prefix + "获取数据：" + PATH);
            ZooKeeper nozk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            Thread.sleep(2000);
            System.out.println(prefix + "成功获取数据：" + nozk.getData(PATH, false, null));
        } catch (Exception e) {
            System.err.println(prefix + "获取数据失败，原因：" + e.getMessage());
        }
    }

    /** 采用正确的密码 */
    static void getDataByCorrectAuthentication() {
        String prefix = "[使用正确的授权信息]";
        try {
            System.out.println(prefix + "获取数据：" + PATH);
            
            System.out.println(prefix + "成功获取数据：" + zk.getData(PATH, false, null));
        } catch (Exception e) {
            System.out.println(prefix + "获取数据失败，原因：" + e.getMessage());
        }
    }

    /**
     * 更新数据：不采用密码
     */
    static void updateDataByNoAuthentication() {

        String prefix = "[不使用任何授权信息]";

        System.out.println(prefix + "更新数据： " + PATH);
        try {
            ZooKeeper nozk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            Thread.sleep(2000);
            Stat stat = nozk.exists(PATH, false);
            if (stat!=null) {
                nozk.setData(PATH, prefix.getBytes(), -1);
                System.out.println(prefix + "更新成功");
            }
        } catch (Exception e) {
            System.err.println(prefix + "更新失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 更新数据：采用错误的密码
     */
    static void updateDataByBadAuthentication() {

        String prefix = "[使用错误的授权信息]";

        System.out.println(prefix + "更新数据：" + PATH);
        try {
            ZooKeeper badzk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            //授权
            badzk.addAuthInfo(authentication_type,badAuthentication.getBytes());
            Thread.sleep(2000);
            Stat stat = badzk.exists(PATH, false);
            if (stat!=null) {
                badzk.setData(PATH, prefix.getBytes(), -1);
                System.out.println(prefix + "更新成功");
            }
        } catch (Exception e) {
            System.err.println(prefix + "更新失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 更新数据：采用正确的密码
     */
    static void updateDataByCorrectAuthentication() {

        String prefix = "[使用正确的授权信息]";

        System.out.println(prefix + "更新数据：" + PATH);
        try {
            Stat stat = zk.exists(PATH, false);
            if (stat!=null) {
                zk.setData(PATH, prefix.getBytes(), -1);
                System.out.println(prefix + "更新成功");
            }
        } catch (Exception e) {
            System.err.println(prefix + "更新失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 不使用密码 删除节点
     */
    static void deleteNodeByNoAuthentication() throws Exception {

        String prefix = "[不使用任何授权信息]";

        try {
            System.out.println(prefix + "删除节点：" + PATH_DEL);
            ZooKeeper nozk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            Thread.sleep(2000);
            Stat stat = nozk.exists(PATH_DEL, false);
            if (stat!=null) {
                nozk.delete(PATH_DEL,-1);
                System.out.println(prefix + "删除成功");
            }
        } catch (Exception e) {
            System.err.println(prefix + "删除失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 采用错误的密码删除节点
     */
    static void deleteNodeByBadAuthentication() throws Exception {

        String prefix = "[使用错误的授权信息]";

        try {
            System.out.println(prefix + "删除节点：" + PATH_DEL);
            ZooKeeper badzk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            //授权
            badzk.addAuthInfo(authentication_type,badAuthentication.getBytes());
            Thread.sleep(2000);
            Stat stat = badzk.exists(PATH_DEL, false);
            if (stat!=null) {
                badzk.delete(PATH_DEL, -1);
                System.out.println(prefix + "删除成功");
            }
        } catch (Exception e) {
            System.err.println(prefix + "删除失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 使用正确的密码删除节点
     */
    static void deleteNodeByCorrectAuthentication() throws Exception {

        String prefix = "[使用正确的授权信息]";

        try {
            System.out.println(prefix + "删除节点：" + PATH_DEL);
            Stat stat = zk.exists(PATH_DEL, false);
            if (stat!=null) {
                zk.delete(PATH_DEL, -1);
                System.out.println(prefix + "删除成功");
            }
        } catch (Exception e) {
            System.out.println(prefix + "删除失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 使用正确的密码删除节点
     */
    static void deleteParent() throws Exception {
        try {
            Stat stat = zk.exists(PATH_DEL, false);
            if (stat == null) {
                zk.delete(PATH, -1);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

原生API操作Zookeeper

Zookeeper系列文章

1.Zookeeper简介

2.Zookeeper集群安装

3.原生API操作Zookeeper

4.zkClient框架操作Zookeeper

5.Curator框架操作Zookeeper

原生API操作Zookeeper

zookeeper-3.3.4. Jar API介绍

创建会话方法：客户端可以通过创建一个 zookeeper 实例来连接 zookeeper 服务器。

创建节点（znode）方法：create:

删除节点：delete 方法（api 提供了两个接口：同步删除和异步删除方式）

注意：在 zookeeper 中，只允许删除叶子节点信息，也就是说如果当前节点不是叶子节点则无法删除，或必须先删除其下所有子节点。

样例程序(创建会话,创建节点,删除节点)

GetChildren 读取数据方法：包括子节点列表的获取和子节点数据的获取。

getData 方法：获取指定节点的数据内容。

注意：该方法和 getChildren 方法基本相同，主要是注册的 watcher 有所不同，客户端在获取一个阶段数据内容时，是可以进行 watcher 注册的，一旦节点发生变更，则服务器端会发送给客户端一个“NodeDataChanged“的事件通知。

setData 方法：修改指定节点的数据内容。

exists 方法：检测节点是否存在。

注意：exists 方法意义在于无论节点是否存在，都可以进行注册 watcher，能够对节点的创建、删除和修改进行监听，但是其子节点发送各种变化，都不会通知客户端。

Zookeeper 有 watch 事件，是一次性触发的，当 watch 监视的数据发生变化时，通知设置了该 watch 的 client，即 watcher.

同样，其 watcher 是监听数据发送了某些变化，那就一定会有对应的事件类型，和状态类型。

Watcher 的特性：一次性、客户端串行执行、轻量。

实例两个上面ZKWatcher对象,ZKWatcher其实就是简单了实现了监听了/super数据节点,你可以把这想象成两个不同的服务

通过下面这个小测试,测试对/super数据节点不同的操作,可以看到上面两个小程序会分别打印对应的信息

Zookeeperd的ACL

猜你喜欢

热点阅读