SpringBoot整合Zookeeper詳細教程

更新時間：2022年12月23日 14:18:00 作者：擅長開發(fā)Bug的Mr.NaCl

Curator是Netflix公司開源的?套zookeeper客戶端框架，Curator是對Zookeeper?持最好的客戶端框架。Curator封裝了?部分Zookeeper的功能，?如Leader選舉、分布式鎖等，減少了技術(shù)?員在使?Zookeeper時的底層細節(jié)開發(fā)?作

一、引言

使用原生的zookeeper時候會遇到watcher一次注冊生效一次等情況，因此使用curator

curator是Netflix公司開源的一個 zookeeper客戶端原生API接口上進行了包裝，解決了很多問題并提供Zookeeper分布式鎖服務、集群領(lǐng)導選舉、共享計數(shù)器、緩存機制、分布式隊列等的應用的抽象封裝

二、引入依賴

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.6.2</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>5.2.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.2.0</version>
</dependency>
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>2.0.10</version>
</dependency>

三、編寫客戶端

要改Windows的host文件。host文件位置是C:\Windows\System32\drivers\etc

3.1、ZookeeperConfig

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ZookeeperConfig {
    //集群地址,分割不能有空格 在程序里寫connectString不可使用ip,必須使用主機名
    private String connectString = "master:2181,slave1:2181,slave2:2181";
    //連接超時時間
    private int sessionTimeout = 5000;
    //會話存活時間,根據(jù)業(yè)務靈活指定
    private Integer sessionTimeOut=5000;
    //重試機制時間參數(shù)
    private Integer sleepMsBetweenRetry=1000;
    //重試機制重試次數(shù)
    private Integer maxRetries=3;
    //命名空間(父節(jié)點名稱)
    private String namespace="";
    /**
     - `session`重連策略
     - `RetryPolicy retry Policy = new RetryOneTime(3000);`
     - 說明：三秒后重連一次，只重連一次
     - `RetryPolicy retryPolicy = new RetryNTimes(3,3000);`
     - 說明：每三秒重連一次，重連三次
     - `RetryPolicy retryPolicy = new RetryUntilElapsed(1000,3000);`
     - 說明：每三秒重連一次，總等待時間超過個`10`秒后停止重連
     - `RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3)`
     - 說明：這個策略的重試間隔會越來越長
     - 公式：`baseSleepTImeMs * Math.max(1,random.nextInt(1 << (retryCount + 1)))`
     - `baseSleepTimeMs` = `1000` 例子中的值
     - `maxRetries` = `3` 例子中的值
     */
    @Bean("curatorClient")
    public CuratorFramework curatorClient() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .connectionTimeoutMs(sessionTimeout)
                .sessionTimeoutMs(sessionTimeOut)
                //session重連策略
                .retryPolicy(new ExponentialBackoffRetry(sleepMsBetweenRetry,maxRetries))
                //設(shè)置命名空間 在操作節(jié)點的時候，會以這個為父節(jié)點
                .namespace(namespace)
                .build();
        client.start();
        //注冊監(jiān)聽器
        ZookeeperWatches watches = new ZookeeperWatches(client);
        watches.znodeWatcher();
        watches.znodeChildrenWatcher();
        return client;
    }
}

3.2、ZookeeperWatches

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.zookeeper.data.Stat;
public class ZookeeperWatches {
    private CuratorFramework client;
    public ZookeeperWatches(CuratorFramework client) {
        this.client = client;
    }
    public void znodeWatcher() throws Exception {
        NodeCache nodeCache = new NodeCache(client, "/");
        nodeCache.start();
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("=======節(jié)點改變===========");
                String path = nodeCache.getPath();
                String currentDataPath = nodeCache.getCurrentData().getPath();
                String currentData = new String(nodeCache.getCurrentData().getData());
                Stat stat = nodeCache.getCurrentData().getStat();
                System.out.println("path:"+path);
                System.out.println("currentDataPath:"+currentDataPath);
                System.out.println("currentData:"+currentData);
            }
        });
        System.out.println("節(jié)點監(jiān)聽注冊完成");
    }
    public void znodeChildrenWatcher() throws Exception {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/",true);
        pathChildrenCache.start();
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                System.out.println("=======節(jié)點子節(jié)點改變===========");
                PathChildrenCacheEvent.Type type = event.getType();
                String childrenData = new String(event.getData().getData());
                String childrenPath = event.getData().getPath();
                Stat childrenStat = event.getData().getStat();
                System.out.println("子節(jié)點監(jiān)聽類型："+type);
                System.out.println("子節(jié)點路徑："+childrenPath);
                System.out.println("子節(jié)點數(shù)據(jù)："+childrenData);
                System.out.println("子節(jié)點元數(shù)據(jù)："+childrenStat);
            }
        });
        System.out.println("子節(jié)點監(jiān)聽注冊完成");
    }
}

3.3、ZookeeperController

@RestController
@RequestMapping(value = "/zookeeper")
public class ZookeeperController {
    @Resource(name = "curatorClient")
    private CuratorFramework client;
    @RequestMapping("/createZnode")
    public String createZnode(){
        String path = "/nacl";
        String data = "shuaige";
        List<ACL> aclList = new ArrayList<>();
        Id id = new Id("world", "anyone");
        aclList.add(new ACL(ZooDefs.Perms.ALL, id));
        try {
            client.create()
                    .creatingParentsIfNeeded()  //沒有父節(jié)點時 創(chuàng)建父節(jié)點
                    .withMode(CreateMode.PERSISTENT)  //節(jié)點類型
                    .withACL(aclList)   //配置權(quán)限
                    .forPath(path, data.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
            return "節(jié)點創(chuàng)建失敗"+e.getMessage();
        }
        return "節(jié)點創(chuàng)建成功";
    }
    @RequestMapping("/selectZnode")
    public String  selectZnode(){
        HashMap<String,String> hashMap=new HashMap();
        String path="/nacl";
        Stat stat;
        try {
            stat = client.checkExists().forPath(path);
            if (stat == null) {
                hashMap.put("Error","不存在該節(jié)點");
            }
            String dataString = new String(client.getData().forPath(path));
            hashMap.put(path, dataString);
            hashMap.put("stat", stat.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return JSON.toJSONString(hashMap);
    }
    @RequestMapping("/selectChildrenZnode")
    public String selectChildrenZnode(){
        Map<String, String> hashMap = new HashMap<>();
        String path="/";
        try {
            List<String> list = client.getChildren().forPath(path);
            for (String s : list) {
                String dataString = new String(client.getData().forPath(path+s));
                hashMap.put(path+s, dataString);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return JSON.toJSONString(hashMap);
    }
    @RequestMapping("/setData")
    public String setData() {
        String path="/nacl";
        String data="big";
        Integer version=0;
        HashMap<String,String> hashMap=new HashMap<>();
        try {
            Stat stat = client.setData().withVersion(version).forPath(path, data.getBytes());
            hashMap.put("success", "修改成功");
            hashMap.put("version", String.valueOf(stat.getVersion()));
        } catch (Exception e) {
            e.printStackTrace();
            hashMap.put("error", "修改失敗:"+e.getMessage());
        }
        return JSON.toJSONString(hashMap);
    }
    @RequestMapping("/delete")
    public String delete() {
        HashMap<String,String> hashMap=new HashMap<>();
        String path="/nacl";
        String data="big";
        Integer version=1;
        try {
            client.delete().withVersion(version).forPath(path);
            hashMap.put("success", "刪除成功");
        } catch (Exception e) {
            e.printStackTrace();
            hashMap.put("error", "刪除失敗:"+e.getMessage());
        }
        return JSON.toJSONString(hashMap);
    }
    @RequestMapping("/createAsyncZnode")
    public String createAsyncZnode(){
        String path = "/nacl";
        String data = "shuaige";
        try {
            client.create()
                    .creatingParentsIfNeeded()
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                    //異步回調(diào)   增刪改都有異步方法
                    .inBackground(new BackgroundCallback() {
                        @Override
                        public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
                            System.out.println("異步回調(diào)--獲取權(quán)限:"+client.getACL().forPath(path));
                            System.out.println("異步回調(diào)--獲取數(shù)據(jù):"+new String(client.getData().forPath(path)));
                            System.out.println("異步回調(diào)--獲取事件名稱:"+event.getName());
                            System.out.println("異步回調(diào)--獲取事件類型:"+event.getType());
                        }
                    })
                    .forPath(path, data.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
            return "節(jié)點創(chuàng)建失敗"+e.getMessage();
        }
        return "節(jié)點創(chuàng)建成功";
    }
}

到此這篇關(guān)于SpringBoot整合Zookeeper詳細教程的文章就介紹到這了,更多相關(guān)SpringBoot整合Zookeeper內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: