ZooKeeper与Curator注册和监控方法

这篇文章主要介绍“ZooKeeper与Curator注册和监控方法”,在日常操作中,相信很多人在ZooKeeper与Curator注册和监控方法问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”ZooKeeper与Curator注册和监控方法”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!

这篇文章主要介绍“ZooKeeper与Curator注册和监控方法”,在日常操作中,相信很多人在ZooKeeper与Curator注册和监控方法问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”ZooKeeper与Curator注册和监控方法”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!

创新互联公司坚持“要么做到,要么别承诺”的工作理念,服务领域包括:成都网站设计、成都网站建设、企业官网、英文网站、手机端网站、网站推广等服务,满足客户于互联网时代的贞丰网站设计、移动媒体设计的需求,帮助企业找到有效的互联网解决方案。努力成为您成熟可靠的网络建设合作伙伴!

Curator提供了对zookeeper客户端的封装,并监控连接状态和会话session,特别是会话session过期后,curator能够重新连接zookeeper,并且创建一个新的session。

对于zk的使用者来说,session的概念至关重要,如果想了解更多session的说明,请访问:

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

zk客户端和zk间主要可能存在下面几种异常情况:

短暂失去连接:此时客户端检测到与服务端的连接已经断开,但是服务端维护的客户端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类型的节点如何监控。

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 watchers = new ConcurrentSkipListSet();    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    }}


文章名称:ZooKeeper与Curator注册和监控方法
分享网址:http://csdahua.cn/article/choic.html
扫二维码与项目经理沟通

我们在微信上24小时期待你的声音

解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流