Zookeeper的Watcher方法

Posted 2020-11-22 whtt

Watcher接口

        在ZooKeeper中，接口类Watcher用于表示一个标准的事件处理器，其定义了事件通知相关的逻辑，包含KeeperState和EventType两个枚举类，分别代表了通知状态和事件类型，同时定义了事件的回调方法：process(WatchedEvent event)。

Watcher事件

        同一个事件类型在不同的通知状态中代表的含义有所不同，下表列举了常见的通知状态和事件类型

 

回调方法process（）

process方法是Watcher接口中的一个回调方法，当ZooKeeper向客户端发送一个Watcher事件通知时，客户端就会对相应的process方法进行回调，从而实现对事件的处理。process方法的定义如下：

abstract public void process(WatchedEvent event);

这个回调方法的定义非常简单，我们重点看下方法的参数定义：WatchedEvent。

WatchedEvent包含了每一个事件的三个基本属性：通知状态（keeperState），事件类型（EventType）和节点路径（path），其数据结构如图7-5所示。ZooKeeper使用WatchedEvent对象来封装服务端事件并传递给Watcher，从而方便回调方法process对服务端事件进行处理。

提到WatchedEvent，不得不讲下WatcherEvent实体。笼统地讲，两者表示的是同一个事物，都是对一个服务端事件的封装。不同的是，WatchedEvent是一个逻辑事件，用于服务端和客户端程序执行过程中所需的逻辑对象，而WatcherEvent因为实现了序列化接口，因此可以用于网络传输。

服务端在生成WatchedEvent事件之后，会调用getWrapper方法将自己包装成一个可序列化的WatcherEvent事件，以便通过网络传输到客户端。客户端在接收到服务端的这个事件对象后，首先会将WatcherEvent还原成一个WatchedEvent事件，并传递给process方法处理，回调方法process根据入参就能够解析出完整的服务端事件了。

需要注意的一点是，无论是WatchedEvent还是WatcherEvent，其对ZooKeeper服务端事件的封装都是机及其简单的。举个例子来说，当/zk-book这个节点的数据发生变更时，服务端会发送给客户端一个“ZNode数据内容变更”事件，客户端只能够接收到如下信

小案例

　　

package com.ZooMckz;

import org.apache.zookeeper.*;

import java.io.IOException;

public class Mckz implements Watcher {
    /**
     * 连接地址
     */
    private static final String ZK_ADDRESS="0.0.0.0:2181";
    /**
     * 会话时间
     */
    private static final Integer ZK_SESSION_TIMEOUT=2000;
    /**
     * Zookerpee对象
     */
    private ZooKeeper zooKeeper;


    public Mckz() {
        openConnection(ZK_ADDRESS,ZK_SESSION_TIMEOUT);
    }

    /**
     * 回调方法     监听连接，监听增删改节点事件
     * @param event
     */
    @Override
    public void process(WatchedEvent event) {
        //获取当前状态
        Event.KeeperState state = event.getState();
        //获取通知类型
        Event.EventType type = event.getType();
        //获取节点路径
        String path = event.getPath();
        System.out.println("当前状态:"+state);
        System.out.println("通知类型:"+type);
        System.out.println("节点路径:"+path);
        //已经成功连接
        if (Event.KeeperState.SyncConnected==state){
            if (Event.EventType.None==type){
                System.out.println("=============连接成功=============");
            }
            if (Event.EventType.NodeCreated==type){
                System.out.println("=============创建成功=============");
            }
            if (Event.EventType.NodeDataChanged==type){
                System.out.println("=============修改成功=============");
            }
            if (Event.EventType.NodeDeleted==type){
                System.out.println("=============删除成功=============");
            }
        }
    }

    /**
     * 连接zk方法
     */
    public void openConnection(String zk_address,Integer zk_session){
        try {
            zooKeeper=new ZooKeeper(zk_address,zk_session,this);
            System.out.println("连接成功！！！");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建节点
     */
    public void createNode(String path,String data){
        try {
            //启动监听
            zooKeeper.exists(path,true);
            //创建节点
            zooKeeper.create(path,data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println("创建成功！！！");
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 修改节点
     */
    public void setNode(String path,String data){
        try {
            //启动监听
            zooKeeper.exists(path,true);
            //修改节点
            zooKeeper.setData(path,data.getBytes(),-1);
            System.out.println("修改成功！！！");
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除节点
     */
    public void deleteNode(String path){
        try {
            //启动监听
            zooKeeper.exists(path,true);
            //删除
            zooKeeper.delete(path,-1);
            System.out.println("删除成功！！！");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    /**
     * 关闭zk方法
     * @param
     */
    public void closeConnection(){
        try {
            if(zooKeeper!=null){
                zooKeeper.close();
            }

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

    public static void main(String[] args) {
        Mckz mckz=new Mckz();
        //创建节点
        mckz.createNode("/mckz","MCKZ");
        //修改
        mckz.setNode("/mckz","wahahaha");
        //删除
        mckz.deleteNode("/mckz");
        //关闭
        mckz.closeConnection();
    }
}