ZooKeeper与Curator注册和监控

时间 2019-11-11

标签 zookeeper curator 注册监控栏目 Zookeeper 繁體版

原文原文链接

Curator提供了对zookeeper客户端的封装，并监控链接状态和会话session，特别是会话session过时后，curator可以从新链接zookeeper，而且建立一个新的session。html

对于zk的使用者来讲，session的概念相当重要，若是想了解更多session的说明，请访问：java

http://zookeeper.apache.org/doc/trunk/zookeeperProgrammers.htmlapache

zk客户端和zk服务器间主要可能存在下面几种异常状况：api

短暂失去链接：此时客户端检测到与服务端的链接已经断开，可是服务端维护的客户端session还没有过时，以后客户端和服务端从新创建了链接；当客户端从新链接后，因为session没有过时，zookeeper可以保证链接恢复后保持正常服务。
失去链接时间很长：此时服务器相对于客户端的session已通过期了，与先前session相关的watcher和ephemeral的路径和数据都会消失；当Curator从新建立了与zk的链接后，会获取到session expired异常，Curator会销毁先前的session，而且会建立一个新的session，须要注意的是，与以前session相关的watcher和ephemeral类型的路径和数据在新的session中也不会存在，须要开发者在CuratorFramework.getConnectionStateListenable().addListener()中添加状态监听事件，对ConnectionState.LOST事件进行监听，当session过时后，使得以前的session状态得以恢复。对于ephemeral类型，在客户端应该保持数据的状态，以便及时恢复。
客户端从新启动：不论先前的zk session是否已通过期，都须要从新建立临时节点、添加数据和watch事件，先前的session也会在稍后的一段时间内过时。
Zk服务器从新启动：因为zk将session信息存放到了硬盘上，所以重启后，先前未过时的session仍然存在，在zk服务器启动后，客户端与zk服务器建立新的链接，并使用先前的session，与1相同。
须要注意的是，当session过时了，在session过时期间另外的客户端修改了zk的值，那么这个修改在客户端从新链接到zk上时，zk客户端不会接收到这个修改的watch事件（尽管添加了watch），若是须要严格的watch逻辑，就须要在curator的状态监控中添加逻辑。

特别提示：watcher仅仅是一次性的，zookeeper通知了watcher事件后，就会将这个watcher从session中删除，所以，若是想继续监控，就要添加新的watcher。服务器

下面提供了对persistent和ephemeral两种类型节点的监控方法，其中get方法说明了persistent节点如何监控，而register方法说明了ephemeral类型的节点如何监控。session

package demo;

import java.net.InetAddress;
import java.nio.charset.Charset;
import java.util.concurrent.ConcurrentSkipListSet;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

public class CuratorTest {
    private CuratorFramework zkTools;
    private ConcurrentSkipListSet<String> watchers = new ConcurrentSkipListSet<String>();
    private static Charset charset = Charset.forName("utf-8");

    public CuratorTest() {
        zkTools = CuratorFrameworkFactory.builder()
                .connectString("192.168.0.216:3306")
                .namespace("zk/test")
                .retryPolicy(new RetryNTimes(2000, 20000))
                .build();
        zkTools.start();
    }

    public void addReconnectionWatcher(final String path, final ZookeeperWatcherType watcherType,
            final CuratorWatcher watcher) {
        synchronized (this) {
            if (!watchers.contains(watcher.toString()))// 不要添加剧复的监听事件
            {
                watchers.add(watcher.toString());
                System.out.println("add new watcher " + watcher);
                zkTools.getConnectionStateListenable().addListener(new ConnectionStateListener() {
                    @Override
                    public void stateChanged(CuratorFramework client, ConnectionState newState) {
                        System.out.println(newState);
                        if (newState == ConnectionState.LOST) {// 处理session过时
                            try {
                                if (watcherType == ZookeeperWatcherType.EXITS) {
                                    zkTools.checkExists().usingWatcher(watcher).forPath(path);
                                } else if (watcherType == ZookeeperWatcherType.GET_CHILDREN) {
                                    zkTools.getChildren().usingWatcher(watcher).forPath(path);
                                } else if (watcherType == ZookeeperWatcherType.GET_DATA) {
                                    zkTools.getData().usingWatcher(watcher).forPath(path);
                                } else if (watcherType == ZookeeperWatcherType.CREATE_ON_NO_EXITS) {
                                    // ephemeral类型的节点session过时了，须要从新建立节点，而且注册监听事件，以后监听事件中，
                                    // 会处理create事件，将路径值恢复到先前状态
                                    Stat stat = zkTools.checkExists().usingWatcher(watcher)
                                            .forPath(path);
                                    if (stat == null) {
                                        System.err.println("to create");
                                        zkTools.create().creatingParentsIfNeeded()
                                                .withMode(CreateMode.EPHEMERAL)
                                                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(path);
                                    }
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }
                });
            }
        }
    }

    public void create() throws Exception {
        zkTools.create()// 建立一个路径
                .creatingParentsIfNeeded()// 若是指定的节点的父节点不存在，递归建立父节点
                .withMode(CreateMode.PERSISTENT)// 存储类型（临时的仍是持久的）
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)// 访问权限
                .forPath("zk/test");// 建立的路径
    }

    public void put() throws Exception {
        // 对路径节点赋值
        zkTools.setData().forPath("zk/test", "hello world".getBytes(Charset.forName("utf-8")));
    }

    public void get() throws Exception {
        String path = "zk/test";
        ZKWatch watch = new ZKWatch(path);
        byte[] buffer = zkTools.getData().usingWatcher(watch).forPath(path);
        System.out.println(new String(buffer, charset));
        // 添加session过时的监控
        addReconnectionWatcher(path, ZookeeperWatcherType.GET_DATA, watch);
    }

    public void register() throws Exception {
        String ip = InetAddress.getLocalHost().getHostAddress();
        String registeNode = "zk/register/" + ip;// 节点路径
        byte[] data = "disable".getBytes(charset);// 节点值
        CuratorWatcher watcher = new ZKWatchRegister(registeNode, data); // 建立一个register watcher
        Stat stat = zkTools.checkExists().forPath(registeNode);
        if (stat != null) {
            zkTools.delete().forPath(registeNode);
        }
        zkTools.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(registeNode, data);// 建立的路径和值
        // 添加到session过时监控事件中
        addReconnectionWatcher(registeNode, ZookeeperWatcherType.CREATE_ON_NO_EXITS, watcher);
        data = zkTools.getData().usingWatcher(watcher).forPath(registeNode);
        System.out.println("get path form zk : " + registeNode + ":" + new String(data, charset));
    }

    public static void main(String[] args) throws Exception {
        CuratorTest test = new CuratorTest();
        test.get();
        test.register();
        Thread.sleep(10000000000L);
    }


    public class ZKWatch implements CuratorWatcher {
        private final String path;

        public String getPath() {
            return path;
        }

        public ZKWatch(String path) {
            this.path = path;
        }

        @Override
        public void process(WatchedEvent event) throws Exception {
            System.out.println(event.getType());
            if (event.getType() == EventType.NodeDataChanged) {
                byte[] data = zkTools.getData().usingWatcher(this).forPath(path);
                System.out.println(path + ":" + new String(data, Charset.forName("utf-8")));
            }
        }
    }

    public class ZKWatchRegister implements CuratorWatcher {
        private final String path;
        private byte[] value;

        public String getPath() {
            return path;
        }

        public ZKWatchRegister(String path, byte[] value) {
            this.path = path;
            this.value = value;
        }

        @Override
        public void process(WatchedEvent event) throws Exception {
            System.out.println(event.getType());
            if (event.getType() == EventType.NodeDataChanged) {
                // 节点数据改变了，须要记录下来，以便session过时后，可以恢复到先前的数据状态
                byte[] data = zkTools.getData().usingWatcher(this).forPath(path);
                value = data;
                System.out.println(path + ":" + new String(data, charset));
            } else if (event.getType() == EventType.NodeDeleted) {
                // 节点被删除了，须要建立新的节点
                System.out.println(path + ":" + path + " has been deleted.");
                Stat stat = zkTools.checkExists().usingWatcher(this).forPath(path);
                if (stat == null) {
                    zkTools.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
                            .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(path);
                }
            } else if (event.getType() == EventType.NodeCreated) {
                // 节点被建立时，须要添加监听事件（建立多是因为session过时后，curator的状态监听部分触发的）
                System.out.println(path + ":" + " has been created!" + "the current data is "
                        + new String(value));
                zkTools.setData().forPath(path, value);
                zkTools.getData().usingWatcher(this).forPath(path);
            }
        }
    }
    public enum ZookeeperWatcherType {
        GET_DATA, GET_CHILDREN, EXITS, CREATE_ON_NO_EXITS
    }

}