Curator實(shí)現(xiàn)zookeeper的節(jié)點(diǎn)監(jiān)聽詳解

更新時間：2023年12月18日 10:23:13 作者：暴躁的程序猿啊

這篇文章主要介紹了Curator實(shí)現(xiàn)zookeeper的節(jié)點(diǎn)監(jiān)聽詳解,Curtor框架中一共有三個實(shí)現(xiàn)監(jiān)聽的方式,一種是NodeCache監(jiān)聽指定節(jié)點(diǎn),一種是pathChildrenCache監(jiān)聽子節(jié)點(diǎn),一種是TreeCache可以監(jiān)控所有節(jié)點(diǎn) 相當(dāng)于以上兩種的合集,需要的朋友可以參考下

Curator實(shí)現(xiàn)zookeeper的節(jié)點(diǎn)監(jiān)聽

Curtor框架中一共有三個實(shí)現(xiàn)監(jiān)聽的方式一種是NodeCache監(jiān)聽指定節(jié)點(diǎn) 一種是pathChildrenCache監(jiān)聽子節(jié)點(diǎn) 一種是TreeCache可以監(jiān)控所有節(jié)點(diǎn) 相當(dāng)于以上兩種的合集

引入依賴

        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-client</artifactId>
            <version>4.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.3.0</version>
        </dependency>

創(chuàng)建一個測試類連接好客戶端

public class CuratorTest {
    private CuratorFramework curatorFramework;
    @Before
    public void testCreate() {
        ExponentialBackoffRetry retry = new ExponentialBackoffRetry(2, 10);
        curatorFramework = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .connectionTimeoutMs(60 * 1000)
                .sessionTimeoutMs(15 * 10000)
                .retryPolicy(retry).build();
        curatorFramework.start();
    }
   @After
    public void close() {
        if (curatorFramework != null) {
            curatorFramework.close();
        }
    }
}

Watch監(jiān)聽之NodeCache

監(jiān)聽一個指定節(jié)點(diǎn)

 @Test
    public void testUpdate() throws Exception {
        //監(jiān)聽一個節(jié)點(diǎn)
        NodeCache nodeCache = new NodeCache(curatorFramework,"/dongwuyuan");
        //注冊監(jiān)聽
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("節(jié)點(diǎn)變化了?。。?！");
            }
        });
        //開啟監(jiān)聽 參數(shù) 如果設(shè)置為true 則開啟監(jiān)聽時加載緩存數(shù)據(jù)
        nodeCache.start(true);
        while (true){
        }
    }

set /dongwuyuan "laohu"

在這里插入圖片描述

Watch監(jiān)聽之PathChildrenCache

監(jiān)聽子節(jié)點(diǎn)的變化

 @Test
    public void testPathChildrenCache() throws Exception {
        // 參數(shù)  客戶端，路徑 ，緩存數(shù)據(jù)，是否壓縮，線程池
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework,"/dongwuyuan",true);
        //綁定監(jiān)聽器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                System.out.println("子節(jié)點(diǎn)變化了");
                System.out.println(pathChildrenCacheEvent);
                //監(jiān)聽子節(jié)點(diǎn)的變更,并且拿到變更后的數(shù)據(jù)
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                //判斷類型是否是update
                if(type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
                     //拿到數(shù)據(jù)
                    byte[] data = pathChildrenCacheEvent.getData().getData();
                    System.out.println(data);
                }
            }
        });
        //開啟監(jiān)聽
        pathChildrenCache.start();
        while (true){
        }
    }

watch監(jiān)聽之TreeCache

  /**
     * TreeCache:監(jiān)聽節(jié)點(diǎn)自己和所有子節(jié)點(diǎn)們
     */
    @Test
    public void testTreeCache(){
        //1.創(chuàng)建監(jiān)聽器
        TreeCache treeCache = new TreeCache(curatorFramework, "/dongwuyuan");
        //2.注冊監(jiān)聽
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                System.out.println("節(jié)點(diǎn)變化了");
                System.out.println(event);
            }
        });
        //開啟監(jiān)聽
        try {
            treeCache.start();
            while (true){
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

[zk: localhost:2181(CONNECTED) 13] delete /dongwuyuan/node1

在這里插入圖片描述

[zk: localhost:2181(CONNECTED) 15] set /dongwuyuan/node2  "shizi"

在這里插入圖片描述

完整代碼

public class CuratorTest {
    private CuratorFramework curatorFramework;
    @Before
    public void testCreate() {
        ExponentialBackoffRetry retry = new ExponentialBackoffRetry(2, 10);
        curatorFramework = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .connectionTimeoutMs(60 * 1000)
                .sessionTimeoutMs(15 * 10000)
                .retryPolicy(retry).build();
        curatorFramework.start();
    }
    @Test
    public void testUpdate() throws Exception {
        //監(jiān)聽一個節(jié)點(diǎn)
        NodeCache nodeCache = new NodeCache(curatorFramework,"/dongwuyuan");
        //注冊監(jiān)聽
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("節(jié)點(diǎn)變化了！?。?！");
            }
        });
        //開啟監(jiān)聽 參數(shù) 如果設(shè)置為true 則開啟監(jiān)聽時加載緩存數(shù)據(jù)
        nodeCache.start(true);
        while (true){
        }
    }
    @Test
    public void testPathChildrenCache() throws Exception {
        // 參數(shù)  客戶端，路徑 ，緩存數(shù)據(jù)，是否壓縮，線程池
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework,"/dongwuyuan",true);
        //綁定監(jiān)聽器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                System.out.println("子節(jié)點(diǎn)變化了");
                System.out.println(pathChildrenCacheEvent);
                //監(jiān)聽子節(jié)點(diǎn)的變更,并且拿到變更后的數(shù)據(jù)
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                //判斷類型是否是update
                if(type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
                     //拿到數(shù)據(jù)
                    byte[] data = pathChildrenCacheEvent.getData().getData();
                    System.out.println(data);
                }
            }
        });
        //開啟監(jiān)聽
        pathChildrenCache.start();
        while (true){
        }
    }
    /**
     * TreeCache:監(jiān)聽節(jié)點(diǎn)自己和所有子節(jié)點(diǎn)們
     */
    @Test
    public void testTreeCache(){
        //1.創(chuàng)建監(jiān)聽器
        TreeCache treeCache = new TreeCache(curatorFramework, "/dongwuyuan");
        //2.注冊監(jiān)聽
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                System.out.println("節(jié)點(diǎn)變化了");
                System.out.println(event);
            }
        });
        //開啟監(jiān)聽
        try {
            treeCache.start();
            while (true){
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    @After
    public void close() {
        if (curatorFramework != null) {
            curatorFramework.close();
        }
    }
}

到此這篇關(guān)于Curator實(shí)現(xiàn)zookeeper的節(jié)點(diǎn)監(jiān)聽詳解的文章就介紹到這了,更多相關(guān)zookeeper節(jié)點(diǎn)監(jiān)聽內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: