SpringBoot整合Zookeeper详细教程

目录

• 一、引言
• 二、引入依赖
• 三、编写客户端
• 3.1、ZookeeperConfig
• 3.2、ZookeeperWatches
• 3.3、ZookeeperController

一、引言

使用原生的zookeeper时候会遇到watcher一次注册生效一次等情况，因此使用curator

curator是Netflix公司开源的一个 zookeeper客户端原生API接口上进行了包装，解决了很多问题并提供Zookeeper分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等的应用的抽象封装

二、引入依赖

<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>
    <grou编程客栈pId>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;
    //会话存活时间,根据业务灵活指定
    private Integer sessionTimeOut=5000;
    //重试机制时间参数
    private Integer sleepMsBetweenRetry=1000;
    //重试机制重试次数
    private Integer maxRetries=3;
    //命名空间(父节点名称)
    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))
                //设置命名空间 在操作节点的时候，会以这个为父节点
       python         .namespace(namespace)
                .build();
        client.start();
        //注册监听器
        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("=======节点改变===========");
                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("节点监听注册完成");
    }
    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("=======节点子节点www.devze.com改变===========");
                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("子节点监听类型："+type);
                System.out.println("子节点路径："+childrenPath);
                System.out.println("子节点数据："+childrenData);
                System.out.println("子节点元数据："+childrenStat);
            }
        });
 http://www.devze.com       System.out.println("子节点监听注册完成");
    }
}

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()  //没有父节点时 创建父节点
       js             .withMode(CreateMode.PERSISTENT)  //节点类型
                    .withACL(aclList)   //配置权限
                    .forPath(path, data.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
            return "节点创建失败"+e.getMessage();
        }
        return "节点创建成功";
    }
    @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","不存在该节点");
            }
            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)
                    //异步回调   增删改都有异步方法
                    .inBackground(new BackgroundCallback() {
                        @Override
                        public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
                            System.out.println("异步回调--获取权限:"+client.getACL().forPath(path));
                            System.out.println("异步回调--获取数据:"+new String(client.getData().forPath(path)));
                            System.out.println("异步回调--获取事件名称:"+event.getName());
                            System.out.println("异步回调--获取事件类型:"+event.getType());
                        }
                    })
  开发者_Go学习                  .forPath(path, data.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
            return "节点创建失败"+e.getMessage();
        }
        return "节点创建成功";
    }
}

到此这篇关于SpringBoot整合Zookeeper详细教程的文章就介绍到这了,更多相关SpringBoot整合Zookeeper内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！