zookeeper curator使用caches实现各种监听

喜欢ヅ旅行 2022-05-31 10:58 227阅读 0赞

zookeeper原理和搭建单机，伪集群，集群--》https://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper-code/   原理：http://lovelock.coding.me/bigdata/zookeeper-simple-practice/

参考地址：http://ifeve.com/zookeeper-path-cache/   转载：https://www.cnblogs.com/kangoroo/p/7245990.html    参考博文：https://www.jianshu.com/p/70151fc0ef5d

win10 本地安装zk实现远程连接：地址：http://blog.csdn.net/asunjianren/article/details/51848800

1、篇首语

curator是zookeeper的一个高级api开发包。封装了zookeeper众多的recipes，并且实现了一些新的recipes原语，最重要的是基于zookeeper提供的各种机制实现了更健壮的连接和异常处理。

本文将其中比较常用的一种recipe，就是cache。

2、各种Caches

cache是一种缓存机制，可以借助cache实现监听。

简单来说，cache在客户端缓存了znode的各种状态，当感知到zk集群的znode状态变化，会触发event事件，注册的监听器会处理这些事件。是不是很简单。

curator支持的cache种类有3种Path Cache，Node Cache，Tree Cache

1）Path Cache

Path Cache用来观察ZNode的子节点并缓存状态，如果ZNode的子节点被创建，更新或者删除，那么Path Cache会更新缓存，并且触发事件给注册的监听器。

Path Cache是通过PathChildrenCache类来实现的，监听器注册是通过PathChildrenCacheListener。

2）Node Cache

Node Cache用来观察ZNode自身，如果ZNode节点本身被创建，更新或者删除，那么Node Cache会更新缓存，并触发事件给注册的监听器。

Node Cache是通过NodeCache类来实现的，监听器对应的接口为NodeCacheListener。

3）Tree Cache

可以看做是上两种的合体，Tree Cache观察的是所有节点的所有数据。

3、下面给出一个例子。

1）这是在springboot中使用curator，先给出curator依赖pom

![复制代码][copycode.gif]

<dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
                <version>3.4.8</version>
            </dependency>
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-recipes</artifactId>
                <version>2.9.1</version>
            </dependency>

![复制代码][copycode.gif]

2）三种cache的实现

![复制代码][copycode.gif]

package com.dqa.prometheus.client.zookeeper;
    
    import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.framework.recipes.cache.*;
    import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    import org.apache.curator.utils.CloseableUtils;
    import org.apache.zookeeper.CreateMode;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.net.InetAddress;
    import java.util.ArrayList;
    import java.util.List;
    
    public class ZkClient {
        private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
        private CuratorFramework client;
        private NodeCache nodeCache;
        private PathChildrenCache pathChildrenCache;
        private TreeCache treeCache;
        private String zookeeperServer;
        private int sessionTimeoutMs;
        private int connectionTimeoutMs;
        private int baseSleepTimeMs;
        private int maxRetries;
    
        public void setZookeeperServer(String zookeeperServer) {
            this.zookeeperServer = zookeeperServer;
        }
        public String getZookeeperServer() {
            return zookeeperServer;
        }
        public void setSessionTimeoutMs(int sessionTimeoutMs) {
            this.sessionTimeoutMs = sessionTimeoutMs;
        }
        public int getSessionTimeoutMs() {
            return sessionTimeoutMs;
        }
        public void setConnectionTimeoutMs(int connectionTimeoutMs) {
            this.connectionTimeoutMs = connectionTimeoutMs;
        }
        public int getConnectionTimeoutMs() {
            return connectionTimeoutMs;
        }
        public void setBaseSleepTimeMs(int baseSleepTimeMs) {
            this.baseSleepTimeMs = baseSleepTimeMs;
        }
        public int getBaseSleepTimeMs() {
            return baseSleepTimeMs;
        }
        public void setMaxRetries(int maxRetries) {
            this.maxRetries = maxRetries;
        }
        public int getMaxRetries() {
            return maxRetries;
        }
    
        public void init() {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(baseSleepTimeMs, maxRetries);
            client = CuratorFrameworkFactory.builder().connectString(zookeeperServer).retryPolicy(retryPolicy)
                    .sessionTimeoutMs(sessionTimeoutMs).connectionTimeoutMs(connectionTimeoutMs).build();
            client.start();
        }
    
        public void stop() {
            if (client != null) CloseableUtils.closeQuietly(client);
            if (pathChildrenCache != null) CloseableUtils.closeQuietly(pathChildrenCache);
            if (nodeCache != null) CloseableUtils.closeQuietly(nodeCache);
            if (treeCache != null) CloseableUtils.closeQuietly(treeCache);
        }
    
        public CuratorFramework getClient() {
            return client;
        }
    
        /*
        *  设置Path Cache, 监控本节点的子节点被创建,更新或者删除，注意是子节点, 子节点下的子节点不能递归监控
        *  事件类型有3个, 可以根据不同的动作触发不同的动作
        *  本例子只是演示, 所以只是打印了状态改变的信息, 并没有在PathChildrenCacheListener中实现复杂的逻辑
        *  @Param path 监控的节点路径, cacheData 是否缓存data
        *  可重入监听
        * */
        public void setPathCacheListener(String path, boolean cacheData) {
            try {
                pathChildrenCache = new PathChildrenCache(client, path, cacheData);
                PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
                    @Override
                    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) {
                        ChildData data = event.getData();
                        switch (event.getType()) {
                            case CHILD_ADDED:
                                logger.info("子节点增加, path={}, data={}", data.getPath(), data.getData());
                                break;
                            case CHILD_UPDATED:
                                logger.info("子节点更新, path={}, data={}", data.getPath(), data.getData());
                                break;
                            case CHILD_REMOVED:
                                logger.info("子节点删除, path={}, data={}", data.getPath(), data.getData());
                                break;
                            default:
                                break;
                        }
                    }
                };
                pathChildrenCache.getListenable().addListener(childrenCacheListener);
                pathChildrenCache.start(StartMode.POST_INITIALIZED_EVENT);
            } catch (Exception e) {
                logger.error("PathCache监听失败, path=", path);
            }
    
        }
    
        /*
        *  设置Node Cache, 监控本节点的新增,删除,更新
        *  节点的update可以监控到, 如果删除会自动再次创建空节点
        *  本例子只是演示, 所以只是打印了状态改变的信息, 并没有在NodeCacheListener中实现复杂的逻辑
        *  @Param path 监控的节点路径, dataIsCompressed 数据是否压缩
        *  不可重入监听
        * */
        public void setNodeCacheListener(String path, boolean dataIsCompressed) {
            try {
                nodeCache = new NodeCache(client, path, dataIsCompressed);
                NodeCacheListener nodeCacheListener = new NodeCacheListener() {
                    @Override
                    public void nodeChanged() throws Exception {
                        ChildData childData = nodeCache.getCurrentData();
                        logger.info("ZNode节点状态改变, path={}", childData.getPath());
                        logger.info("ZNode节点状态改变, data={}", childData.getData());
                        logger.info("ZNode节点状态改变, stat={}", childData.getStat());
                    }
                };
                nodeCache.getListenable().addListener(nodeCacheListener);
                nodeCache.start();
            } catch (Exception e) {
                logger.error("创建NodeCache监听失败, path={}", path);
            }
        }
    
    
        /*
        *  设置Tree Cache, 监控本节点的新增,删除,更新
        *  节点的update可以监控到, 如果删除不会自动再次创建
        *  本例子只是演示, 所以只是打印了状态改变的信息, 并没有在NodeCacheListener中实现复杂的逻辑
        *  @Param path 监控的节点路径, dataIsCompressed 数据是否压缩
        *  可重入监听
        * */
        public void setTreeCacheListener(final String path) {
            try {
                treeCache = new TreeCache(client, path);
                TreeCacheListener treeCacheListener = new TreeCacheListener() {
                    @Override
                    public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                        ChildData data = event.getData();
                        if(data != null){
                            switch (event.getType()) {
                                case NODE_ADDED:
                                    logger.info("[TreeCache]节点增加, path={}, data={}", data.getPath(), data.getData());
                                    break;
                                case NODE_UPDATED:
                                    logger.info("[TreeCache]节点更新, path={}, data={}", data.getPath(), data.getData());
                                    break;
                                case NODE_REMOVED:
                                    logger.info("[TreeCache]节点删除, path={}, data={}", data.getPath(), data.getData());
                                    break;
                                default:
                                    break;
                            }
                        }else{
                            logger.info("[TreeCache]节点数据为空, path={}", data.getPath());
                        }
                    }
                };
                treeCache.getListenable().addListener(treeCacheListener);
                treeCache.start();
            } catch (Exception e) {
                logger.error("创建TreeCache监听失败, path={}", path);
            }
    
        }
    }

![复制代码][copycode.gif]

3）configuration

init方法是初始化zookeeper client的操作

stop是停止zookeeper是的清理动作

![复制代码][copycode.gif]

package com.dqa.prometheus.configuration;
    
    import com.xiaoju.dqa.prometheus.client.zookeeper.ZkClient;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class ZkConfiguration {
    
        @Value("${zookeeper.server}")
        private String zookeeperServer;
        @Value(("${zookeeper.sessionTimeoutMs}"))
        private int sessionTimeoutMs;
        @Value("${zookeeper.connectionTimeoutMs}")
        private int connectionTimeoutMs;
        @Value("${zookeeper.maxRetries}")
        private int maxRetries;
        @Value("${zookeeper.baseSleepTimeMs}")
        private int baseSleepTimeMs;
    
        @Bean(initMethod = "init", destroyMethod = "stop")
        public ZkClient zkClient() {
            ZkClient zkClient = new ZkClient();
            zkClient.setZookeeperServer(zookeeperServer);
            zkClient.setSessionTimeoutMs(sessionTimeoutMs);
            zkClient.setConnectionTimeoutMs(connectionTimeoutMs);
            zkClient.setMaxRetries(maxRetries);
            zkClient.setBaseSleepTimeMs(baseSleepTimeMs);
            return zkClient;
        }
    
    }

![复制代码][copycode.gif]

3）zk配置文件

其中最重要的应该是会话超时和重试机制了。

============== zookeeper ===================
    zookeeper.server=10.93.21.21:2181,10.93.18.34:2181,10.93.18.35:2181
    zookeeper.sessionTimeoutMs=6000
    zookeeper.connectionTimeoutMs=6000
    zookeeper.maxRetries=3
    zookeeper.baseSleepTimeMs=1000

[copycode.gif]: /images/20220531/52b53dc489be4008962136a596357ac1.png