ZooKeeper基础知识学习一

一、zookeeper基本命令操作

##1、服务端基本命令操作
## 查看zookeeper服务运行的状态
[root@ser874514596417 bin]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: standalone

## 启动服务
[root@ser874514596417 bin]# zkServer.sh start

## 停止服务
[root@ser874514596417 bin]# zkServer.sh stop

## 启动客户端连接zookeeper服务
[root@ser874514596417 bin]# zkClient.sh -server localhost:2181

// 2、客户端基本命令操作

## 查看指定节点下的子节点
[zk: localhost:2181(CONNECTED) 4] ls /
[node1, zookeeper]

## 创建 节点
[zk: localhost:2181(CONNECTED) 5] create /node1/app1
Created /node1/app1

## 删除指定节点
[zk: localhost:2181(CONNECTED) 0] delete /node1/app1

## 删除指定节点 （含有子节点）
[zk: localhost:2181(CONNECTED) 6] deleteall /node1

## 给节点设置 数据
[zk: localhost:2181(CONNECTED) 7] create /node1 yangxu
Created /node1

## 获取节点 的数据
[zk: localhost:2181(CONNECTED) 8] get /node1
yangxu

## 给节点设置数据
[zk: localhost:2181(CONNECTED) 9] set /node1 xuzai

## 创建临时节点 （会话级别）
[zk: localhost:2181(CONNECTED) 9] create -e /node1/app3 xuzai

## 创建顺序节点 
[zk: localhost:2181(CONNECTED) 0] create -s /node2
Created /node20000000002

[zk: localhost:2181(CONNECTED) 2] create -s /node2
Created /node20000000003


## 查看节点详细信息
[zk: localhost:2181(CONNECTED) 9] ls -s /node1

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

二、zookeeper Javaapi 操作（curator）

/**
     * 创建节点： create 持久 临时 顺序 数据
     * 1、基本创建
     * 2、创建节点 带有数据
     * 3、设置节点类型
     * 4、创建多级节点
     */
    @Test
    public void testCreate() throws Exception {
        // 如果创建节点没有数据，则默认数据是当前客户端的IP地址
        String str = client2.create().forPath("/app1");
        System.out.println(str);

        String str2 = client2.create().forPath("/app2", "yangxu".getBytes());
        System.out.println(str2);

        // 创建一个临时的节点
        String str3 = client2.create().withMode(CreateMode.EPHEMERAL).forPath("/app3");
        System.out.println(str3);


        String str4 = client2.create().creatingParentsIfNeeded().forPath("/app4/huawei");
        System.out.println(str4);

    }

    /**
     * 1、查询数据
     * 2、查询子节点
     * 3、查询节点的状态信息 ls -s
     * @throws Exception
     */
    @Test
    public void testGet() throws Exception {
        // 查询数据
        byte[] str = client2.getData().forPath("/app1");
        System.out.println(new String(str));

        // 查询子节点
        List<String> str2 = client2.getChildren().forPath("/app2");
        System.out.println(str2);

        // 查询节点的状态信息 ls -s\
        Stat stat = new Stat();
        byte[] str3 = client2.getData().storingStatIn(stat).forPath("/app3");
        System.out.println(new String(str3));
        System.out.println(stat);
    }

    /**
     * 设置数据
     *
     * @throws Exception
     */
    @Test
    public void testSet() throws Exception {
        // 修改数据
        Stat str = client2.setData().forPath("/app1", "xuzai".getBytes());
        System.out.println(str);

        // 根据版本修改
        // 先查询节点的状态信息
        Stat stat = new Stat();
        client2.getData().storingStatIn(stat).forPath("/app2");
        int version = stat.getVersion();
        final Stat stat1 = client2.setData().withVersion(version).forPath("/app2");
        System.out.println(stat1);
    }


    /**
     * 删除单个节点
     * 删除带有子节点的节点
     * 必须成功的删除
     * 回调
     * @throws Exception
     */
    @Test
    public void testDelete() throws Exception {
        // 删除单个节点
        client2.delete().forPath("/app1");

        // 删除带有子节点的节点
        client2.delete().deletingChildrenIfNeeded().forPath("/app2");

        // 必须成功的删除
        client2.delete().guaranteed().forPath("/app2");

        // 回调
       client2.delete().guaranteed().inBackground(new BackgroundCallback() {
            @Override
            public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                System.out.println("我被删除了");

            }
        }).forPath("/app3");
    }

三、Watch事件监听

Zookeeper允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，Zookeeper服务端会将事件通知感兴趣的客户端上去，该机制是Zookeeper实现分布式协调服务的重要特性。

zookeeper中引入了Watcher机制来实现发布/订阅功能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态发生变化时，会通知所有订阅者。

zookeeper原生支持通过注册Watcher来进行事件监听，但是并不是特别方便需要开发人员自己反复注册Watcher，比较繁琐。

curator引入了Cache来实现zookeeper服务端事件的监听。

Zookeeper提供了三种Watcher：
	NodeCache: 只是监听某一特定的节点
	PathChildrenCache: 监控一个ZNode的子节点
	TreeCache： 可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合。

实例代码：

/**
     * 演示nodeCache:给指定的一个节点注册监听器
     *
     */
    public void testNodeCache() throws Exception {
        // 1、创建NodeCache对象
        NodeCache nodeCache = new NodeCache(client2, "/app1");

        // 2、注册监听
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("节点变化了");
                byte[] currentData = nodeCache.getCurrentData().getData();
                System.out.println(new String(currentData));
            }
        });
        // 3、开启监听，如果设置为true，则开启监听，加载缓冲数据
        nodeCache.start(true);
        while (true) {

        }

    }

    /**
     * 演示PathChildrenCache:给某个节点的子节点指定注册监听器
     *
     */
    public void testPathChildrenCache() throws Exception {
        // 1、创建NodeCache对象
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client2, "/app2", true);


        // 2、注册监听
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                System.out.println("子节点变化了");
                // 监听子节点的数据变更，并且拿到变更后的数据
                // 1、获取类型
                PathChildrenCacheEvent.Type type = event.getType();
                // 2、判断类型是否是update
                if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {
                    byte[] data = event.getData().getData();
                    System.out.println(new String(data));

                }
            }
        });
        // 3、开启监听
        pathChildrenCache.start();
        while (true) {}

    }

    /**
     * 演示testTreeCache:给某个节点和他的子节点指定注册监听器
     *
     */
    public void testTreeCache() throws Exception {
        // 1、创建NodeCache对象
        TreeCache treeCache = new TreeCache(client2, "/app2");


        // 2、注册监听
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
                System.out.println("节点变化了");
            }
        });

        // 3、开启监听
        treeCache.start();
        while (true) {}

    }

四、分布式锁

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

zookeeper分布式锁的原理

核心思想：当客户端想要获取锁，则创建节点，使用完锁，则删除节点
1、客户端获取锁时，在lock节点下创建临时顺序节点。
2、然后获取lock下面的所有子节点，客户端获取到所有的子节点之后，如果发现自己创建的子节点最小，那么就任务该客户端获取到锁了，使用完后，将该节点删除。
3、如果发现自己创建的节点并非lock所有子节点中最小的，说明自己还没有
获取到锁，此时客户端需要找到比自己小的哪个节点，同时对其注册事件监听器，监听删除事件。
4、如果发现比自己小的节点被删除了，则客户端的Watcher会收到相应的通知，此时在判断自己创建的节点是否是lock子节点中序号最小的，如果是则获取到锁，如果不是则重复以上步骤继续获取到比自己小的一个节点并注册监听。

五、zookeeper集群搭建

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

1.1 搭建要求

真实的集群是需要部署在不同的服务器上的，但是在我们测试时同时启动很多个虚拟机内存会吃不消，所以我们通常会搭建伪集群，也就是把所有的服务都搭建在一台虚拟机上，用端口进行区分。

我们这里要求搭建一个三个节点的Zookeeper集群（伪集群）。

1.2 准备工作

重新部署一台虚拟机作为我们搭建集群的测试服务器。

（1）安装JDK 【此步骤省略】。

（2）Zookeeper压缩包上传到服务器（3）将Zookeeper解压，建立/usr/local/zookeeper-cluster目录，将解压后的Zookeeper复制到以下三个目录

/usr/local/zookeeper-cluster/zookeeper-1

/usr/local/zookeeper-cluster/zookeeper-2

/usr/local/zookeeper-cluster/zookeeper-3

[root@localhost ~]# mkdir /usr/local/zookeeper-cluster
[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-1
[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-2
[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-3

（4）创建data目录，并且将 conf下zoo_sample.cfg 文件改名为 zoo.cfg

mkdir /usr/local/zookeeper-cluster/zookeeper-1/data
mkdir /usr/local/zookeeper-cluster/zookeeper-2/data
mkdir /usr/local/zookeeper-cluster/zookeeper-3/data

mv  /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg
mv  /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg
mv  /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

（5）配置每一个Zookeeper 的dataDir 和 clientPort 分别为2181 2182 2183

修改/usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg

clientPort=2181
dataDir=/usr/local/zookeeper-cluster/zookeeper-1/data

修改/usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg

clientPort=2182
dataDir=/usr/local/zookeeper-cluster/zookeeper-2/data

修改/usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

clientPort=2183
dataDir=/usr/local/zookeeper-cluster/zookeeper-3/data

1.3 配置集群

（1）在每个zookeeper的 data 目录下创建一个 myid 文件，内容分别是1、2、3 。这个文件就是记录每个服务器的ID

echo 1 >/usr/local/zookeeper-cluster/zookeeper-1/data/myid
echo 2 >/usr/local/zookeeper-cluster/zookeeper-2/data/myid
echo 3 >/usr/local/zookeeper-cluster/zookeeper-3/data/myid

（2）在每一个zookeeper 的 zoo.cfg配置客户端访问端口（clientPort）和集群服务器IP列表。

集群服务器IP列表如下

vim /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg
vim /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg
vim /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

server.1=192.168.149.135:2881:3881
server.2=192.168.149.135:2882:3882
server.3=192.168.149.135:2883:3883

解释：server.服务器ID=服务器IP地址：服务器之间通信端口：服务器之间投票选举端口

1.4 启动集群

启动集群就是分别启动每个实例。

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh start

启动后我们查询一下每个实例的运行状态

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

先查询第一个服务

Mode为follower表示是跟随者（从）

再查询第二个服务Mod 为leader表示是领导者（主）

查询第三个为跟随者（从）

1.5 模拟集群异常

（1）首先我们先测试如果是从服务器挂掉，会怎么样

把3号服务器停掉，观察1号和2号，发现状态并没有变化

/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh stop

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

由此得出结论，3个节点的集群，从服务器挂掉，集群正常

（2）我们再把1号服务器（从服务器）也停掉，查看2号（主服务器）的状态，发现已经停止运行了。

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh stop

/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

由此得出结论，3个节点的集群，2个从服务器都挂掉，主服务器也无法运行。因为可运行的机器没有超过集群总数量的半数。

（3）我们再次把1号服务器启动起来，发现2号服务器又开始正常工作了。而且依然是领导者。

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start

/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

（4）我们把3号服务器也启动起来，把2号服务器停掉,停掉后观察1号和3号的状态。

/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh stop

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

发现新的leader产生了~

由此我们得出结论，当集群中的主服务器挂了，集群中的其他服务器会自动进行选举状态，然后产生新得leader

（5）我们再次测试，当我们把2号服务器重新启动起来启动后，会发生什么？2号服务器会再次成为新的领导吗？我们看结果

/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh start

/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

我们会发现，2号服务器启动后依然是跟随者（从服务器），3号服务器依然是领导者（主服务器），没有撼动3号服务器的领导地位。

由此我们得出结论，当领导者产生后，再次有新服务器加入集群，不会影响到现任领导者。