谈谈Zookeeper

用户3467126

发布于 2019-07-03 18:07:55

5070

发布于 2019-07-03 18:07:55

文章被收录于专栏：爱编码

简介

ZooKeeper是一种高性能的分布式应用协调服务。它在一个简单的接口中公开公共服务，例如命名、配置管理、同步和组服务，这样你就不必从头编写它们。你可以使用它来实现共识、组管理、领导人选举和到场协议。您可以根据自己的特定需要在此基础上进行构建。

特点

顺序一致性——来自客户机的更新将按照发送的顺序应用。
原子性——更新成功或失败。没有部分结果。
单个系统映像——客户机将看到服务的相同视图，而不管它连接到哪个服务器。
可靠性——一旦更新被应用，它将从那时起一直持续到客户端覆盖更新。
及时性——保证客户对系统的看法在一定的时间范围内是最新的。

目标

了解基本原理
掌握基本使用。

原理

1.设计目标 ZooKeeper允许分布式进程通过类似于标准文件系统的共享层次命名空间相互协调。名称空间由数据寄存器(ZooKeeper中称为znodes)组成，这些寄存器类似于文件和目录。与设计用于存储的典型文件系统不同，ZooKeeper数据保存在内存中，这意味着ZooKeeper可以获得高吞吐量和低延迟数。

zookeeper服务架构服务器必须相互了解。它们在内存中维护状态映像，以及持久存储中的事务日志和快照。只要大部分服务器可用，ZooKeeper服务就会可用。客户端连接到单个ZooKeeper服务器。客户机维护一个TCP连接，通过该连接发送请求、获取响应、获取监视事件和发送心跳。如果到服务器的TCP连接中断，客户机将连接到另一个服务器。

2.数据模型

数据模型

ZooKeeper提供的名称空间与标准文件系统非常相似。名称是用斜杠(/)分隔的路径元素序列。ZooKeeper名称空间中的每个节点都由一个路径标识。

3.节点和短暂的节点 存储在名称空间中的每个znode上的数据都是按原子方式读写的。read获取与znode关联的所有数据字节，write替换所有数据。每个节点都有一个访问控制列表(ACL)，它限制谁可以做什么。 ZooKeeper也有短暂节点的概念。只要创建znode的会话处于活动状态，这些znode就存在。当会话结束时，删除znode。当您希望实现时，临时节点非常有用。

4.条件更新和监视 znode可以被监控，包括这个目录节点中存储的数据的修改，子节点目录的变化等，一旦变化可以通知设置监控的客户端，这个是zookeeper的核心特性，zookeeper的很多功能都是基于这个特性实现的。

5.zkServer中的角色 根据其身份特性分为三种 leader：负责客户端的writer类型请求。 follower：负责客户端的reader类型请求，参与leader选举等。 observer：特殊的”follower“，其可以接受客户端reader请求，但不参与选举。（扩容系统支撑能力，提高了读取速度。因为它不接受任何同步的写入请求，只负责与leader同步数据）

其中Follower和observer又统称learner

使用

1.安装教程 我就不做重复介绍参考链接： https://www.cnblogs.com/huangjianping/p/8012580.html

2.Java连接操作 官网操作： http://zookeeper.apache.org/doc/current/javaExample.html#sc_design

推荐使用Curator的方式

public class CuratorUtil {

    /** zookeeper地址 */
    static final String CONNECT_ADDR = "120.79.226.4:2181";
    /** session超时时间 */
    static final int SESSION_OUTTIME = 5000;//ms

    public static void main(String[] args) throws Exception {

        //1 重试策略：初试时间为1s 重试10次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
        //2 通过工厂创建连接
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                .connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_OUTTIME)
                .retryPolicy(retryPolicy)
//                    .namespace("super")
                .build();
        //3 开启连接
        cf.start();

        System.out.println(ZooKeeper.States.CONNECTED);
        System.out.println(cf.getState());

        // 新加、删除
        /**
         //4 建立节点 指定节点类型（不加withMode默认为持久类型节点）、路径、数据内容    */
//        cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c1","c1内容".getBytes());
        //5 删除节点
//        cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");


        // 读取、修改
        /**
         //创建节点        */
//        cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c1","c1内容".getBytes());
//        cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c2","c2内容".getBytes());
        //读取节点
//        String ret1 = new String(cf.getData().forPath("/super/c2"));
//        System.out.println(ret1);
        //修改节点
//        cf.setData().forPath("/super/c2", "修改c2内容".getBytes());
//        String ret2 = new String(cf.getData().forPath("/super/c2"));
//        System.out.println(ret2);


        // 绑定回调函数
        /**
         ExecutorService pool = Executors.newCachedThreadPool();
         cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
         .inBackground(new BackgroundCallback() {
        @Override
        public void processResult(CuratorFramework cf, CuratorEvent ce) throws Exception {
        System.out.println("code:" + ce.getResultCode());
        System.out.println("type:" + ce.getType());
        System.out.println("线程为:" + Thread.currentThread().getName());
        }
        }, pool)
         .forPath("/super/c3","c3内容".getBytes());
         Thread.sleep(Integer.MAX_VALUE);
         */


        // 读取子节点getChildren方法 和 判断节点是否存在checkExists方法
        /**
         List<String> list = cf.getChildren().forPath("/super");
         for(String p : list){
         System.out.println(p);
         }

         Stat stat = cf.checkExists().forPath("/super/c3");
         System.out.println(stat);

         Thread.sleep(2000);
         cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");
         */


        //cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");

    }
}