惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

大猫的无限游戏
大猫的无限游戏
Webroot Blog
Webroot Blog
cs.AI updates on arXiv.org
cs.AI updates on arXiv.org
T
Threat Research - Cisco Blogs
V2EX - 技术
V2EX - 技术
L
LINUX DO - 热门话题
Google DeepMind News
Google DeepMind News
Recorded Future
Recorded Future
S
Schneier on Security
I
InfoQ
Cyber Security Advisories - MS-ISAC
Cyber Security Advisories - MS-ISAC
cs.CV updates on arXiv.org
cs.CV updates on arXiv.org
The GitHub Blog
The GitHub Blog
S
Security @ Cisco Blogs
O
OpenAI News
W
WeLiveSecurity
Vercel News
Vercel News
阮一峰的网络日志
阮一峰的网络日志
Simon Willison's Weblog
Simon Willison's Weblog
人人都是产品经理
人人都是产品经理
Cloudbric
Cloudbric
The Last Watchdog
The Last Watchdog
The Hacker News
The Hacker News
Google Online Security Blog
Google Online Security Blog
OSCHINA 社区最新新闻
OSCHINA 社区最新新闻
GbyAI
GbyAI
NISL@THU
NISL@THU
T
Tailwind CSS Blog
V
Visual Studio Blog
PCI Perspectives
PCI Perspectives
K
KPMG report finds enterprise disconnect between AI and its ROI | CIO
Jina AI
Jina AI
D
DataBreaches.Net
B
Blog RSS Feed
N
News and Events Feed by Topic
N
News and Events Feed by Topic
H
Heimdal Security Blog
cs.CL updates on arXiv.org
cs.CL updates on arXiv.org
腾讯CDC
Latest news
Latest news
V
Vulnerabilities – Threatpost
Hacker News: Ask HN
Hacker News: Ask HN
WordPress大学
WordPress大学
V
V2EX
aimingoo的专栏
aimingoo的专栏
博客园 - 司徒正美
Apple Machine Learning Research
Apple Machine Learning Research
D
Darknet – Hacking Tools, Hacker News & Cyber Security
The Register - Security
The Register - Security
Help Net Security
Help Net Security

博客园 - 菜鸟乙

提示词 大模型实战 2-4 提示词聚合网站 提示词 大模型实战 1-1 提示词工程师导学 中间件 ZK分布式专题与Dubbo微服务入门 8-5 暴露商品服务 0-4min 中间件 ZK分布式专题与Dubbo微服务入门 9-4 获取分布式锁的流程 中间件 ZK分布式专题与Dubbo微服务入门 8-8 使用dubbo内置main打包jar启动 中间件 ZK分布式专题与Dubbo微服务入门 8-7 使用main主线程启动dubbo服务 中间件 ZK分布式专题与Dubbo微服务入门 8-6 使用tomcat启动dubbo服务 中间件 ZK分布式专题与Dubbo微服务入门 8-4 重构商品服务,抽取抽象工程 中间件 ZK分布式专题与Dubbo微服务入门 8-3 单体到分层模式代码演示 中间件 ZK分布式专题与Dubbo微服务入门 9-2 分布式锁的概念与数据最终不一致性的场景 中间件 ZK分布式专题与Dubbo微服务入门 7-9 zk-watcher实例 统一更新N台节点的配置文件 中间件 ZK分布式专题与Dubbo微服务入门 7-10 curator之acl权限操作与认证授权 中间件 ZK分布式专题与Dubbo微服务入门 9-1 死锁与活锁的概念 中间件 ZK分布式专题与Dubbo微服务入门 7-8 curator之PathChildrenCache子节点监听 中间件 ZK分布式专题与Dubbo微服务入门 8-1 架构演变过程 中间件 ZK分布式专题与Dubbo微服务入门 7-6 curator之usingWatcher 中间件 ZK分布式专题与Dubbo微服务入门 7-5 查询节点相关信息 中间件 ZK分布式专题与Dubbo微服务入门 8-2 dubbo 入门简介 中间件 ZK分布式专题与Dubbo微服务入门 7-4 修改节点以及删除节点
中间件 ZK分布式专题与Dubbo微服务入门 7-7 curator之nodeCache一次注册N次监听
菜鸟乙 · 2024-04-09 · via 博客园 - 菜鸟乙

0    课程地址

https://coding.imooc.com/lesson/201.html#mid=12736

1    重点关注

1.1    本节内容

curator自带节点监听多次(增删改,不针对子节点)

1.2    关键代码

// 为节点添加watcher
        // NodeCache: 监听数据节点的变更,会触发事件
        final NodeCache nodeCache = new NodeCache(cto.client, nodePath);
//        // buildInitial : 初始化的时候获取node的值并且缓存
        nodeCache.start(true);
        if (nodeCache.getCurrentData() != null) {
            System.out.println("节点初始化数据为:" + new String(nodeCache.getCurrentData().getData()));
        } else {
            System.out.println("节点初始化数据为空...");
        }
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            public void nodeChanged() throws Exception {
                if (nodeCache.getCurrentData() == null) {
                    System.out.println("空");
                    return;
                }
                String data = new String(nodeCache.getCurrentData().getData());
                System.out.println("节点路径:" + nodeCache.getCurrentData().getPath() + "数据:" + data);
            }
        });
        

2    课程内容



3    Coding

3.1    Curator使用监听

  • 启动服务端
    进入到
cd /usr/local/zookeeper/bin

 
    重启zookeeper服务端
./zkServer.sh restart
  • 主类
package com.imooc.curator;

import java.util.List;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;

public class CuratorOperator {

    public CuratorFramework client = null;
    public static final String zkServerPath = "172.26.139.4:2181";

    /**
     * 实例化zk客户端
     */
    public CuratorOperator() {
        /**
         * 同步创建zk示例,原生api是异步的
         * 
         * curator链接zookeeper的策略:ExponentialBackoffRetry
         * baseSleepTimeMs:初始sleep的时间
         * maxRetries:最大重试次数
         * maxSleepMs:最大重试时间
         */
//        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
        
        /**
         * curator链接zookeeper的策略:RetryNTimes
         * n:重试的次数
         * sleepMsBetweenRetries:每次重试间隔的时间
         */
        RetryPolicy retryPolicy = new RetryNTimes(3, 5000);
        
        /**
         * curator链接zookeeper的策略:RetryOneTime
         * sleepMsBetweenRetry:每次重试间隔的时间
         */
//        RetryPolicy retryPolicy2 = new RetryOneTime(3000);
        
        /**
         * 永远重试,不推荐使用
         */
//        RetryPolicy retryPolicy3 = new RetryForever(retryIntervalMs)
        
        /**
         * curator链接zookeeper的策略:RetryUntilElapsed
         * maxElapsedTimeMs:最大重试时间
         * sleepMsBetweenRetries:每次重试间隔
         * 重试时间超过maxElapsedTimeMs后,就不再重试
         */
//        RetryPolicy retryPolicy4 = new RetryUntilElapsed(2000, 3000);
        
        client = CuratorFrameworkFactory.builder()
                .connectString(zkServerPath)
                .sessionTimeoutMs(10000).retryPolicy(retryPolicy)
                .namespace("workspace").build();
        client.start();
    }
    
    /**
     * 
     * @Description: 关闭zk客户端连接
     */
    public void closeZKClient() {
        if (client != null) {
            this.client.close();
        }
    }
    
    public static void main(String[] args) throws Exception {
        // 实例化
        CuratorOperator cto = new CuratorOperator();
        boolean isZkCuratorStarted = cto.client.isStarted();
        System.out.println("当前客户的状态:" + (isZkCuratorStarted ? "连接中" : "已关闭"));
        
        // 创建节点
        String nodePath = "/super/imooc";
//        byte[] data = "superme".getBytes();
//        cto.client.create().creatingParentsIfNeeded()
//            .withMode(CreateMode.PERSISTENT)
//            .withACL(Ids.OPEN_ACL_UNSAFE)
//            .forPath(nodePath, data);
        
        // 更新节点数据
//        byte[] newData = "batman".getBytes();
//        cto.client.setData().withVersion(0).forPath(nodePath, newData);
        
        // 删除节点
//        cto.client.delete()
//                  .guaranteed()                    // 如果删除失败,那么在后端还是继续会删除,直到成功
//                  .deletingChildrenIfNeeded()    // 如果有子节点,就删除
//                  .withVersion(0)
//                  .forPath(nodePath);
        
        
        
        // 读取节点数据
//        Stat stat = new Stat();
//        byte[] data = cto.client.getData().storingStatIn(stat).forPath(nodePath);
//        System.out.println("节点" + nodePath + "的数据为: " + new String(data));
//        System.out.println("该节点的版本号为: " + stat.getVersion());
        
        
        // 查询子节点
//        List<String> childNodes = cto.client.getChildren()
//                                            .forPath(nodePath);
//        System.out.println("开始打印子节点:");
//        for (String s : childNodes) {
//            System.out.println(s);
//        }
        
                
        // 判断节点是否存在,如果不存在则为空
//        Stat statExist = cto.client.checkExists().forPath(nodePath + "/abc");
//        System.out.println(statExist);
        
        
        // watcher 事件  当使用usingWatcher的时候,监听只会触发一次,监听完毕后就销毁
//        cto.client.getData().usingWatcher(new MyCuratorWatcher()).forPath(nodePath);
//        cto.client.getData().usingWatcher(new MyWatcher()).forPath(nodePath);
        
        // 为节点添加watcher
        // NodeCache: 监听数据节点的变更,会触发事件
        final NodeCache nodeCache = new NodeCache(cto.client, nodePath);
//        // buildInitial : 初始化的时候获取node的值并且缓存
        nodeCache.start(true);
        if (nodeCache.getCurrentData() != null) {
            System.out.println("节点初始化数据为:" + new String(nodeCache.getCurrentData().getData()));
        } else {
            System.out.println("节点初始化数据为空...");
        }
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            public void nodeChanged() throws Exception {
                if (nodeCache.getCurrentData() == null) {
                    System.out.println("空");
                    return;
                }
                String data = new String(nodeCache.getCurrentData().getData());
                System.out.println("节点路径:" + nodeCache.getCurrentData().getPath() + "数据:" + data);
            }
        });
        
        
        // 为子节点添加watcher
        // PathChildrenCache: 监听数据节点的增删改,会触发事件
//        String childNodePathCache =  nodePath;
//        // cacheData: 设置缓存节点的数据状态
//        final PathChildrenCache childrenCache = new PathChildrenCache(cto.client, childNodePathCache, true);
//        /**
//         * StartMode: 初始化方式
//         * POST_INITIALIZED_EVENT:异步初始化,初始化之后会触发事件
//         * NORMAL:异步初始化
//         * BUILD_INITIAL_CACHE:同步初始化
//         */
//        childrenCache.start(StartMode.POST_INITIALIZED_EVENT);
//        
//        List<ChildData> childDataList = childrenCache.getCurrentData();
//        System.out.println("当前数据节点的子节点数据列表:");
//        for (ChildData cd : childDataList) {
//            String childData = new String(cd.getData());
//            System.out.println(childData);
//        }
//        
//        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
//            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
//                if(event.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)){
//                    System.out.println("子节点初始化ok...");
//                }
//                
//                else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)){
//                    String path = event.getData().getPath();
//                    if (path.equals(ADD_PATH)) {
//                        System.out.println("添加子节点:" + event.getData().getPath());
//                        System.out.println("子节点数据:" + new String(event.getData().getData()));
//                    } else if (path.equals("/super/imooc/e")) {
//                        System.out.println("添加不正确...");
//                    }
//                    
//                }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){
//                    System.out.println("删除子节点:" + event.getData().getPath());
//                }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
//                    System.out.println("修改子节点路径:" + event.getData().getPath());
//                    System.out.println("修改子节点数据:" + new String(event.getData().getData()));
//                }
//            }
//        });
        
        Thread.sleep(100000);
        
        cto.closeZKClient();
        boolean isZkCuratorStarted2 = cto.client.isStarted();
        System.out.println("当前客户的状态:" + (isZkCuratorStarted2 ? "连接中" : "已关闭"));
    }
    
    public final static String ADD_PATH = "/super/imooc/d";
    
}

  • linux客户端修改节点(修改多次)
--启动linux客户端
zkCli.sh

--第1次修改
[zk: localhost:2181(CONNECTED) 17] set /workspace/super/imooc 32   
cZxid = 0xf7
ctime = Tue Apr 09 06:58:32 CST 2024
mZxid = 0xfa
mtime = Tue Apr 09 07:06:47 CST 2024
pZxid = 0xf7
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 2
numChildren = 0

--第2次修改
[zk: localhost:2181(CONNECTED) 18] set /workspace/super/imooc 31
cZxid = 0xf7
ctime = Tue Apr 09 06:58:32 CST 2024
mZxid = 0xfb
mtime = Tue Apr 09 07:06:49 CST 2024
pZxid = 0xf7
cversion = 0
dataVersion = 2
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 2
numChildren = 0

--删除
[zk: localhost:2181(CONNECTED) 19] delete /workspace/super/imooc   


--新增
[zk: localhost:2181(CONNECTED) 20] create /workspace/super/imooc 78
Created /workspace/super/imooc
[zk: localhost:2181(CONNECTED) 21] 
  • 打印日志(监听多次)
2024-04-09 07:06:35,045 [main] [org.apache.zookeeper.Environment.logEnv(Environment.java:100)] - [INFO] Client environment:user.home=C:\Users\18612
2024-04-09 07:06:35,045 [main] [org.apache.zookeeper.Environment.logEnv(Environment.java:100)] - [INFO] Client environment:user.dir=D:\project\xin\zk_sts_imooc\imooc-zk-curator-maven
2024-04-09 07:06:35,046 [main] [org.apache.zookeeper.ZooKeeper.<init>(ZooKeeper.java:441)] - [INFO] Initiating client connection, connectString=172.26.139.4:2181 sessionTimeout=10000 watcher=org.apache.curator.ConnectionState@5158b42f
2024-04-09 07:06:35,154 [main-SendThread(172.26.139.4:2181)] [org.apache.zookeeper.ClientCnxn$SendThread.logStartConnect(ClientCnxn.java:1035)] - [INFO] Opening socket connection to server 172.26.139.4/172.26.139.4:2181. Will not attempt to authenticate using SASL (unknown error)
2024-04-09 07:06:35,155 [main] [org.apache.curator.framework.imps.CuratorFrameworkImpl.start(CuratorFrameworkImpl.java:326)] - [INFO] Default schema
当前客户的状态:连接中
2024-04-09 07:06:35,158 [main-SendThread(172.26.139.4:2181)] [org.apache.zookeeper.ClientCnxn$SendThread.primeConnection(ClientCnxn.java:877)] - [INFO] Socket connection established to 172.26.139.4/172.26.139.4:2181, initiating session
2024-04-09 07:06:35,165 [main-SendThread(172.26.139.4:2181)] [org.apache.zookeeper.ClientCnxn$SendThread.onConnected(ClientCnxn.java:1302)] - [INFO] Session establishment complete on server 172.26.139.4/172.26.139.4:2181, sessionid = 0x1003feb17e7000b, negotiated timeout = 10000
2024-04-09 07:06:35,171 [main-EventThread] [org.apache.curator.framework.state.ConnectionStateManager.postState(ConnectionStateManager.java:237)] - [INFO] State change: CONNECTED
节点初始化数据为:44
节点路径:/super/imooc数据:32
节点路径:/super/imooc数据:31
空
节点路径:/super/imooc数据:78