学习Zookeeper_大数据系统

1.Zookeeper概念

Zookeeper是 Apache Hadoop项目下的一个子项目，是一个树形目录服务。
Zookeeper翻译过来就是动物园管理员，他是用来管Hadoop (大象)、Hive(蜜蜂)、Pig(小猪)的管理员。简称水
Zookeeper是一个分布式的、开源的分布式应用程序的协调服务。
Zookeeper提供的主要功能包括：
        配置管理
        分布式锁
        集群管理

2.Zookeeper 命令操作

Zookeeper 数据模型

ZooKeeper是一个树形目录服务,其数据模型和Unix的文件系统目录树很类似，拥有一个层次化结构。
这里面的每一个节点都被称为：ZNode，每个节点上都会保存自己的数据和节点信息。
节点可以拥有子节点，同时也允许少量 (1MB) 数据存储在该节点之下。
节点可以分为四大类：
        PERSISTENT 持久化节点
        EPHEMERAL 临时节点：-e
        PERSISTENT_SEQUENTIAL 持久化顺序节点：-s
        EPHEMERAL_SEQUENTIAL 临时顺序节点：-es

ZK服务端命令：
./zkServer.sh start 开启ZK
./zkServer.sh status 查看ZK
./zkServer.sh stop 关闭ZK
./zkServer.sh restart 重启ZK

ZK客户端命令：
连接ZK服务端    ./zkCli.sh -server ip:port
创建节点 creat /节点path value
获取节点值 get /节点path
设置节点值 set /节点path value
删除单个子节点 delete /节点path
删除带有子节点的节点 deleteall /节点path
查看命令帮助    help
断开连接        quit

创建临时节点 create -e /节点path value
创建顺序节点 create -s /节点path value
查询节点详细信息 ls -s /节点path

                czxid:节点被创建的事务ID
                ctime:创建时间
                mzxid:最后一次被更新的事务ID
                mtime:修改时间
                pzxid:子节点列表最后一次被更新的事务ID
                cversion:子节点的版本号
                dataversion:数据版本号
                aclversion:权限版本号
                ephemeralOwner:用于临时节点，代表临时节点的事务ID，如果为持久节点则为0
                dataLength:节点存储的数据的长度
                numChildren:当前节点的子节点个数

3.Zookeeper Courator介绍

Curator 介绍
Curator是 Apache ZooKeeper 的Java客户端库。

Curator项目的目标是简化ZooKeeper客户端的使用。
Curator最初是Netfix研发的,后来捐献了Apache基金会，目前是Apache的顶级项目。

4.Zookeeper 建立连接

//建立连接 
public void testCourator() {

   //建立连接
    @Before
    public void linkCourator() {
        
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
        // 第一种
        // CuratorframeworkFactory.newClient("127.0.0.1:2181", 60 * 1000, 15 * 1000, retryPolicy);
        // 第二种
        // Curatorframework client = CuratorframeworkFactory.builder()
        client = CuratorframeworkFactory.builder()
                .connectString("127.0.0.1:2181")
                // .connectString("192.168.89.181:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy).namespace("jie").build();

        // 开启连接
        client.start();
    }

    @After
    public void close() {
        if (client != null){
            client.close();
        }
    }
}

5.Zookeeper 创建节点

创建节点：create 持久临时顺序数据
        1. 基本创建：create().forPath("")
        2. 创建节点带有数据:create().forPath("",data)
        3. 设置节点的类型：create().withMode().forPath("",data)
        4. 创建多级节点 /app1/p1 ：create().creatingParentsIfNeeded().forPath("",data)

    //创建节点：持久 临时 顺序 数据
    @Test
    public void testCreate() throws Exception {
        //基本创建
        //如果没有指定数据，则默认讲当前客户端的IP作为数据存储
        String path = client.create().forPath("/app1");
        System.out.println(path);
    }

    //创建节点
    @Test
    public void testCreate2() throws Exception {
        //带数据创建
        String path = client.create().forPath("/app2");
        System.out.println(path);
    }

    @Test
    public void testCreate3() throws Exception {
        //设置节点的类型
        String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3", "data".getBytes());
        System.out.println(path);
    }

    @Test
    public void testCreate4() throws Exception {
        //创建多级节点
        String path = client.create().creatingParentsIfNeeded().forPath("/app4/data");
        System.out.println(path);
    }

6.Zookeeper 查询节点

查询节点：
        1. 查询数据：get: getData().forPath()
        2. 查询子节点： ls: getChildren().forPath()
        3. 查询节点状态信息：ls -s:getData().storingStatIn(状态对象).forPath()

    //查询节点
    @Test
    public void testGet() throws Exception {
        //查询数据
        byte[] data = client.getData().forPath("/app1");
        System.out.println(new String(data));
    }

    @Test
    public void testGet2() throws Exception {
        //查询子节点
        List list = client.getChildren().forPath("/");
        System.out.println(list);
    }

    @Test
    public void testGet3() throws Exception {
        Stat stat = new Stat();
        //查询状态信息
        client.getData().storingStatIn(stat).forPath("/app1");
        System.out.println(stat);
    }

7.Zookeeper 修改节点

修改数据
        1. 基本修改数据：setData().forPath()
        2. 根据版本修改: setData().withVersion().forPath()
        version 是通过查询出来的。目的就是为了让其他客户端或者线程不干扰我。

    //修改节点
    @Test
    public void testSet() throws Exception {
        client.setData().forPath("/app1","new".getBytes());
    }

    @Test
    public void testSet2() throws Exception {

        Stat stat = new Stat();
        //查询状态信息
        client.getData().storingStatIn(stat).forPath("/app1");

        int version = stat.getVersion(); // 查询出来

        client.setData().withVersion(version).forPath("/app1","new_2".getBytes());
        // 错误版本无法修改
        // client.setData().withVersion(100).forPath("/app1","new_3".getBytes());
    }

8.Zookeeper 删除节点

删除节点： delete deleteall
        1. 删除单个节点:delete().forPath("/app1");
        2. 删除带有子节点的节点:delete().deletingChildrenIfNeeded().forPath("/app1");
        3. 必须成功的删除:为了防止网络抖动。本质就是重试。client.delete().guaranteed().forPath("/app2");
        4. 回调：inBackground

    //删除节点
    @Test
    public void testDelete() throws Exception {
        //删除叶子节点
        client.delete().forPath("/app1");
    }

    @Test
    public void testDelete2() throws Exception {
        //删除非叶子节点
        client.delete().deletingChildrenIfNeeded().forPath("/app2/data");
    }

    @Test
    public void testDelete3() throws Exception {
        //必须成功删除 本质就是重试
        client.delete().guaranteed().forPath("/app2");
    }

    @Test
    public void testDelete4() throws Exception {
        //回调
        client.delete().guaranteed().inBackground(new BackgroundCallback() {
            @Override
            public void processResult(Curatorframework curatorframework, CuratorEvent curatorEvent) throws Exception {
                System.out.println("die");
                System.out.println(curatorEvent);
            }
        }).forPath("/app1");
    }

9.Zookeeper Watch监听概述

Curator API常用操作–Watch事件监听
        ZooKeeper允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，ZooKeeper服务端会将事件通知到感兴趣的客户端上去，该机制是ZooKeeper实现分布式协调服务的重要特性。
        ZooKeeper中引入了Watcher机制来实现了发布/订阅功能能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。
        ZooKeeper原生支持通过注册Watcher来进行事件监听，但是其使用并不是特别方便需要开发人员自己反复注册Watcher，比较繁琐。
        Curator引入了Cache来实现对ZooKeeper服务端事件的监听。
        ZooKeeper提供了三种Watcher
                NodeCache:只是监听某一个特定的节点
                PathChildrenCache:监控一个ZNode的子节点.
                TreeCache:可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合

10.Zookeeper NodeCache

NodeCache:只是监听某一个特定的节点

    //演示 NodeCache：给指定节点注册监听器
    @Test
    public void testNodeCache() throws Exception {
        //创建NodeCache
        final NodeCache nodeCache = new NodeCache(client, "/mydubbo");
        //注册监听
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("节点变化了!");
                //获取修改节点后的数据
                byte[] data = nodeCache.getCurrentData().getData();
                System.out.println(new String(data));
            }
        });
        //开启监听，如果设置为true，则开启监听时，加载缓冲数据
        nodeCache.start(true);
        //不让程序直接结束
        while (true) {
        }
    }

11.Zookeeper PathChildrenCache

PathChildrenCache:监控一个ZNode的子节点.

    //演示 PathChildrenCache：监听某个节点的所有子节点
    @Test
    public void testPathChildrenCache() throws Exception {
        //创建监听对象
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/mydubbo", true);
        //绑定监听器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(Curatorframework curatorframework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                System.out.println("节点变化了!");
                System.out.println(pathChildrenCacheEvent);
                //获取修改节点后的数据
                //获取类型
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
                    byte[] data = pathChildrenCacheEvent.getData().getData();
                    System.out.println(new String(data));
                }
            }
        });
        //开启监听，如果设置为true，则开启监听时，加载缓冲数据
        pathChildrenCache.start();
        //不让程序直接结束
        while (true) {
        }
    }

12.Zookeeper TreeCache

TreeCache:可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合

    //演示 TreeCache：监听某个节点自己和所有子节点
    @Test
    public void testTreeCache() throws Exception {
        //创建监听对象
        TreeCache treeCache = new TreeCache(client,"/mydubbo");
        //绑定监听器
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(Curatorframework curatorframework, TreeCacheEvent treeCacheEvent) throws Exception {
                System.out.println("节点变化了!");
                System.out.println(treeCacheEvent);
                //获取修改节点后的数据
                //获取类型
                TreeCacheEvent.Type type = treeCacheEvent.getType();
                if (type.equals(TreeCacheEvent.Type.NODE_UPDATED)){
                    byte[] data = treeCacheEvent.getData().getData();
                    System.out.println(new String(data));
                }
            }
        });
        //开启监听，如果设置为true，则开启监听时，加载缓冲数据
        treeCache.start();
        //不让程序直接结束
        while (true) {
        }
    }

13.分布式锁概念

分布式锁
        在我们进行单机应用开发，涉及并发同步的时候，我们往往采用synchronized或者Lock的方式来解决多线程间的代码同续问题,这时多线程的运行都是在同一个JVM之下，没有任何问题。
        但当我们的应用是分布式集群工作的情况下，属于多 JVM下的工作环境，跨JVM之间已经无法通过多线程的锁解决同步问题。
        那么就需要一种更加高级的锁机制，来处理这种跨机器的进程之间的数据同步问题这就是分布式锁。

14.ZooKeeper 分布式锁原理

核心思想：当客户端要获取锁，则创建节点，使用完锁，则删除该节点

1、客户端获取锁时，在lock节点下创建临时顺序节点

2、然后获取lock下面的所有子节点，客户端获取到所有子节点后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁。使用完锁后，将该节点删除。

3、如果发现自己创建的节点并非lock所有子节点中最小的，说明自己还有没获取到锁，此时客户端需要找到比自己小的那个节点，同时对其注册事件监听器，监听删除事件。

4、如果发现比自己小的那个节点被删除，则客户端的watcher会收到相应的通知，此时再次判断自己创建的节点是否是lock子节点中序号最小的值，如果是则获取到了锁，如果不是则重复以上步骤继续获取到比自己小的一个节点并注册监听。

15.Curator 实现分布式锁

在Curator中有五种锁方案：
        InterProcessSemaphoreMutex：分布式排它锁（非可重入锁）
        InterProcessMutex：分布式可重入排它锁
        InterProcessReadWriteLock：分布式读写锁
        InterProcessMultiLock：将多个锁作为单个实体管理的容
        InterProcessSemaphoreV2：共享信号量

public class Ticket12306 implements Runnable{

    private int tickets = 100;//数据库的票数

    private InterProcessMutex lock ;

    public Ticket12306(){
        //重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
        //2.第二种方式
        //CuratorframeworkFactory.builder();
        Curatorframework client = CuratorframeworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .build();

        //开启连接
        client.start();

        lock = new InterProcessMutex(client,"/lock");
    }

    @Override
    public void run() {
        while(true){
            //获取锁
            try {
                lock.acquire(3, TimeUnit.SECONDS);
                if(tickets > 0){

                    System.out.println(Thread.currentThread()+":"+tickets);
                    tickets--;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                //释放锁
                try {
                    lock.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

    }
}

16.Zookeeper集群

Leader选举：
        Serverid：服务器ID比如有三台服务器，编号分别是1,2,3。编号越大在选择算法中的权重越大。
        Zxid：数据ID服务器中存放的最大数据ID.值越大说明数据越新，在选举算法中数据越新权重越大。
        在Leader选举的过程中，如果某台ZooKeeper获得了超过半数的选票，则此ZooKeeper就可以成为Leader了

Zookeeper 集群角色

在ZooKeeper集群服中务中有三个角色：
Leader 领导者：
        处理事务请求
        集群内部各服务器的调度者

Follower 跟随者：
        处理客户端非事务请求，转发事务请求给Leader服务器
        参与Leader选举投票

Observer 观察者：
        处理客户端非事务请求，转发事务请求给Leader服务器

学习Zookeeper

大数据系统相关栏目本月热门文章