基于Curator检测zookeeper分布式集群节点状态变化

男娘i 2024-04-17 19:17 79阅读 0赞

## 基于Curator检测zookeeper分布式集群节点状态变化 ##

Curator是apache下面的一个开源项目，是基于zookeeper的一个项目。旨在简化zookeeper节点数据操作。Curator的项目主页：

[http://curator.apache.org/][http_curator.apache.org]

Curator在github上的开源项目主页：[https://github.com/apache/curator][https_github.com_apache_curator]

在idea中开发，需要在pom.xml添加对Curator的引用。pom.xml引用：

<dependencies>
            <dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
                <version>3.5.5</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-recipes</artifactId>
                <version>4.2.0</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-framework</artifactId>
                <version>4.2.0</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-core</artifactId>
                <version>2.12.1</version>
            </dependency>
        </dependencies>

添加好依赖后，就可以基于Curator检测zookeeper节点在分布式环境中的状态。

import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.framework.api.CuratorWatcher;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    import org.apache.zookeeper.WatchedEvent;
    
    
    public class Main {
        public static void main(String[] args) {
            //初始化log4j，zookeeper否则报错。
            org.apache.log4j.BasicConfigurator.configure();
    
            try {
                Main m = new Main();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        public Main() throws Exception {
            String ip = "localhost";
            String addrs = ip + ":2181," + ip + ":2182," + ip + ":2183";
    
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
    
            String nodePath="/test_data";
    
            CuratorFramework client = CuratorFrameworkFactory.builder()
                    .connectString(addrs)
                    .sessionTimeoutMs(10*1000)
                    .retryPolicy(retryPolicy)
                    .build();  // 建立连接通道。
    
            client.start();
    
            client.getData().usingWatcher(new MyCuratorWatcher()).forPath(nodePath);
    
            synchronized (Main.class) {
                Main.class.wait();
            }
        }
    
        private class MyCuratorWatcher implements CuratorWatcher {
            public void process(WatchedEvent event) throws Exception {
                System.out.println("路径:"+event.getPath());
                System.out.println("事件:"+event.toString());
            }
        }
    }

其中，nodePath是事先已经在zookeeper节点中写入的节点path。程序启动后，再往/test\_data节点路径下写入数据，就会被CuratorWatcher捕获到。

[http_curator.apache.org]: http://curator.apache.org/
[https_github.com_apache_curator]: https://github.com/apache/curator