ZooKeeper基础

清疚 2022-05-13 12:20 211阅读 0赞

### ZooKeeper概述 ###

ZooKeeper 是一个管理分布式应用的分布式协调服务. ZooKeeper允许开发人员专注于核心应用程序逻辑，而不必担心应用程序的分布式特性. ZooKeeper提供的常见服务如下 :

*  命名服务 - 按名称标识集群中的节点。
 *  配置管理 - 加入节点的最近的和最新的系统配置信息。
 *  集群管理 - 实时地在集群和节点状态中加入/离开节点。
 *  选举算法 - 选举一个节点作为协调目的的leader。
 *  锁定和同步服务 - 在修改数据的同时锁定数据。此机制可帮助你在连接其他分布式应用程序（如Apache HBase）时进行自动故障恢复。
 *  高度可靠的数据注册表 - 即使在一个或几个节点关闭时也可以获得数据。

### ZooKeeper架构 ###

![在这里插入图片描述][20190328182041387.jpg]

Client是我们的分布式应用集群中的一个节点, Server是ZooKeeper集群的一个节点. 当Client连接上ZooKeeper集群中的任何一个Server节点, 就会与ZooKeeper集群建立一个TCP连接. 通过TCP连接发送requests, 获取responses, 获取watch events, 发送heart beats. 如果与这个Server节点的连接断开了, Client会自动连接ZooKeeper集群其他Server节点.

### ZooKeeper基本概念 ###

**数据模型和层次命名空间**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70]

ZooKeeper的内部是一个类似于文件系统的命名空间, 节点的名字由一系列通过斜线(/)分隔的path组成. ZooKeeper命名空间的节点被称为Znode, Znode通常用来存储一些状态信息、配置、位置信息等, 所以Znode的存储空间不太, 一般在1M以内. 值得注意的是, 对Znode的读写操作是满足原子性的.

Znode还维护着一个stat结构, 其属性如下图所示:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 1]

Znode节点主要有三种, 分别是持久节点, 临时节点, 顺序节点.

*  持久节点 - 一旦创建, 除非主动移除, 否则一直存在.
 *  临时节点 - 生命周期与session绑定, 一旦session失效, 该节点就会被移除. 所以临时节点不能有子节点.
 *  顺序节点 - 顺序节点可以是持久的或临时的. 当创建顺序节点时, ZooKeeper会在其路径后面自动添加10位的计数器, 例如  0000000001,  0000000002, ……这个计数器在同一个父节点下是唯一的。计数器是一个4字节的有符号整形, 当数值超过2147483647时将会溢出.

**ZooKeeper追踪时间的方式**

*  Zxid - 每次ZooKeeper发生改变都会接受到一个zxid(ZooKeeper Transaction Id), 如果zxid1小于zxid2, 那么zxid1发生在zxid2前面.
 *  Version numbers - 每次Znode发生改变, 版本号都会增加. 版本号有三种, 分别是version, cversion, aversion.
 *  Ticks - 当搭建ZooKeeper集群时, 服务器节点之间通过ticks来确定事件的时间. 如status uploads, session timeouts, connection timeouts.
 *  Real time - 只有Stat结构的ctime和mtime才会用到真正的时间格式.

**Session**

Client只要连接上ZooKeeper集群中的任何一台Server, 就能创建session. ZooKeeper在创建session时会生成一个64-bit的session id和一个password发送到客户端, 当客户端要和服务器重新建立session连接时, 只需向服务器发送相应的session id和password. 当服务器压力太大、网络故障或客户端更新服务器列表等各种原因断开连接时, 需要重连.

当Client申请连接Server时需要指定session的timeout, timeout最小是tickTime的2倍, 最大是tickTime的20倍(tickTime在配置文件zoo.cfg中配置). 当Server在超时时间间隔内没有收到来自Client的请求(heartbeat)时, session就会失效, Server将会删除其所有的临时节点.

**Watcher**

客户端能对Znode设置一个监听器, 当Znode发生变化时, 监听器将会被触发, 给客户端发送一个事件通知. 该事件通知是异步发送到客户端的. 监听器只能被触发一次, 被触发后会被删除.

监听器分为两种, 一种是data watches, 一种是child watches. data监听器通过getData() 和 exists() 设置, child监听器通过getChildren()设置. setData() 会触发当前节点的data监听器, create()会触发当前节点的data监听器和父节点的child 监听器, delete()将触发当前节点的data监听器和child监听器, 以及父节点的child监听器.

(1)监听器事件

*  None - 与服务器断开连接时客户端收到此事件.
 *  NodeCreated - Znode创建事件. 通过exists()启用
 *  NodeDeleted - Znode删除事件. 通过exists(), getData(), getChildren()启用.
 *  NodeDataChanged - Znode数据内容更新事件. 通过exists(), getData()启用.
 *  NodeChildrenChanged - Znode子节点改变事件, 只关注子节点数量的变更, 子节点内容的变更是不会通知的. 通过getChildren()启用.

(2)事件发生后ZooKeeper的状态

*  Disconnected(0)- 断开连接
 *  SyncConnected(3)- 连接状态
 *  AuthFailed(4) - 身份验证失败
 *  ConnectedReadOnly(5) - 只读连接
 *  SaslAuthenticated(6) - SASL认证
 *  Expired(-112) - 会话超时

**ACL**  
ZooKeeper 使用ACL(Access Control List)来控制Znode节点的访问权限. ACL权限控制只对当前节点有效, 对其子节点无效. ACL格式为 scheme:permissions, 其中scheme表示采用哪一种机制, permissions表示相关权限.

(1)Permissions

*  create - 能创建子节点
 *  delete - 能删除子节点
 *  read - 能读取数据和显示子节点
 *  write - 能写入数据
 *  admin - 能设置权限

(2)Scheme

*  world - 默认机制, world:anyone, 所有人都能访问.
 *  auth - 只要是通过authentication的用户都能访问.
 *  digest - 格式: digest:username:password:permissions. 访问节点之前先通过username:password的方式进行认证.
 *  ip - 格式: ip:addr/bits

### ZooKeeper API ###

(1)建立连接

public class ZooKeeperConnection {
        private ZooKeeper zoo;
        final CountDownLatch connectedSignal = new CountDownLatch(1);
    
        // 连接ZooKeeper服务器
        public ZooKeeper connect(String host) throws IOException,InterruptedException {
            zoo = new ZooKeeper(host,5000,new Watcher() {
                public void process(WatchedEvent we) {
                    if (we.getState() == KeeperState.SyncConnected) {
                        connectedSignal.countDown();
                    }
                }
            });
            //因为监听器是异步操作, 要保证监听操作先完成
            connectedSignal.await();
            return zoo;
        }
    
        // 关闭连接
        public void close() throws InterruptedException {
            zoo.close();
        }
    }

(2)创建Znode

public class ZKCreate {
        private static ZooKeeper zk;
        private static ZooKeeperConnection conn;
    
        /**
         * 在ZooKeeper服务器上创建Znode节点
         * 服务器路径, 数据, 访问权限, 节点类型
         */
        public static void create(String path, byte[] data) throws
                KeeperException,InterruptedException {
            //创建所有人都能访问的持久节点
            zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.PERSISTENT);
        }
    
        public static void main(String[] args) {
            String path = "/MyFirstZnode";
            byte[] data = "My first zookeeper app".getBytes();
    
            try {
                conn = new ZooKeeperConnection();
                zk = conn.connect("192.168.1.130");
                create(path, data);
                conn.close();
            } catch (Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }

(3)检查节点是否存在

public class ZKExists {
        private static ZooKeeper zk;
        private static ZooKeeperConnection conn;
    
        // 检查指定路径的Znode是否存在, 返回该节点的状态信息
        public static Stat znode_exists(String path) throws
                KeeperException,InterruptedException {
            return zk.exists(path, true);
        }
    
        public static void main(String[] args) throws InterruptedException,KeeperException {
            String path = "/MyFirstZnode";
            try {
                conn = new ZooKeeperConnection();
                zk = conn.connect("192.168.1.130");
                Stat stat = znode_exists(path);
                if(stat != null) {
                    System.out.println("Node exists and the node version is " +
                            stat.getVersion());
                } else {
                    System.out.println("Node does not exists");
                }
            } catch(Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }

(4)查询节点数据

public class ZKGetData {
        private static ZooKeeper zk;
        private static ZooKeeperConnection conn;
    
        public static Stat znode_exists(String path) throws
                KeeperException,InterruptedException {
            return zk.exists(path,true);
        }
    
        public static void main(String[] args) throws InterruptedException, KeeperException {
            String path = "/MyFirstZnode";
            final CountDownLatch connectedSignal = new CountDownLatch(1);
    
            try {
                conn = new ZooKeeperConnection();
                zk = conn.connect("192.168.1.130");
                //先判断节点是否存在
                Stat stat = znode_exists(path);
                if(stat != null) {
                    byte[] b = zk.getData(path, true,null);
                    String data = new String(b, "UTF-8");
                    System.out.println(data);
                    connectedSignal.await();
                } else {
                    System.out.println("Node does not exists");
                }
            } catch(Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }

(5)修改节点数据

public class ZKSetData {
        private static ZooKeeper zk;
        private static ZooKeeperConnection conn;
    
        // 修改ZooKeeper服务器Znode节点上的数据
        public static void update(String path, byte[] data) throws
                KeeperException,InterruptedException {
            zk.setData(path, data, zk.exists(path,true).getVersion());
        }
    
        public static void main(String[] args) throws InterruptedException,KeeperException {
            String path= "/MyFirstZnode";
            byte[] data = "Success".getBytes(); 
            try {
                conn = new ZooKeeperConnection();
                zk = conn.connect("192.168.1.130");
                update(path, data);
            } catch(Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }

(6)获取所有子节点

public class ZKGetChildren {
        private static ZooKeeper zk;
        private static ZooKeeperConnection conn;
    
        public static Stat znode_exists(String path) throws
                KeeperException,InterruptedException {
            return zk.exists(path,true);
        }
    
        public static void main(String[] args) throws InterruptedException,KeeperException {
            String path = "/MyFirstZnode";
            try {
                conn = new ZooKeeperConnection();
                zk = conn.connect("192.168.1.130");
                Stat stat = znode_exists(path);
                if(stat != null) {
                    List <String> children = zk.getChildren(path, false);
                    for(int i = 0; i < children.size(); i++)
                        System.out.println(children.get(i));
                } else {
                    System.out.println("Node does not exists");
                }
            } catch(Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }

(7)删除节点(注意不能级联删除)

public class ZKDelete {
        private static ZooKeeper zk;
        private static ZooKeeperConnection conn;
    
        //当有子节点时, 要先删除子节点
        public static void delete(String path) throws KeeperException,InterruptedException {
            zk.delete(path,zk.exists(path,true).getVersion());
        }
    
        public static void main(String[] args) throws InterruptedException,KeeperException {
            String path = "/MyFirstZnode/Child2";
            try {
                conn = new ZooKeeperConnection();
                zk = conn.connect("192.168.1.130");
                delete(path);
            } catch(Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }

[20190328182041387.jpg]: /images/20220513/31f868132ed641ba9951e976efdc7a3c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70]: /images/20220513/f8e98a38d2b343229c87a5ef92eb2b73.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 1]: /images/20220513/76d06fe3ac3544c4a0b44c8d233247ab.png