zookeeper3.5.9的安装与使用 Curator Watcher监听 分布式锁

zookeeper是Apache Hadoop的子项目，是一个树形的目录服务。

在官网下载想要的版本： http://zookeeper.apache.org/releases.html   如这里使用的是 apache-zookeeper-3.5.9-bin.tar.gz（该版本是windows和Linux通用的版本）

 

*安装 （温馨提示，zookeeper是java语言编写的，它依赖于JDK7及以上版本）.

 

1， 上传到Linux服务器并解压到指定路径

> mkdir zookeeper

> tar -zxvf apache-zookeeper-3.5.9-bin.tar.gz

 

2， zookeeper的启动必须有zoo.cfg配置文件，默认是没有的

> cp zoo_sample.cfg zoo.cfg

> vi zoo.cfg  编辑该文件，主要是指定zookeeper的数据保存路径

     dataDir=/app/zookeeper/zkdata

 

3， 启动 (默认是 2181端口 )

> cd bin

> ./zkServer.sh start 启动，，，当然还有 status stop restart 参数可用.

 

zookeeper中每个节点被称为 ZNode，每个节点上都会保存自己的数据和节点信息, 节点下可以有子节点;

通常节点分为 持久化节点、临时节点 -e 、 持久化顺序节点 -s  临时顺序节点 -es

 

*客户端命令

> ./zkCli.sh  [-server localhost:2181]     如果客户端和服务端在一台，客户端就是要连这一台可以省略参数 连接成功后，退出用 quit

 *JavaApi调用zkServer

    Curator，它是Apache提供的Java客户端库连接zookeeper，最初是Netflix研发的捐献给Apache基金会，目前是Apache的顶级项目。

    官网： http://curator.apache.org

 

   0. 准备工程  pom.xml

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>

        <!--curator start-->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.3.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.3.0</version>
        </dependency>
        <!--curator end-->

        <!--日志-->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.30</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.30</version>
        </dependency>

    </dependencies>


    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

   1. 连接到zkServer

   2. 创建节点

   3. 查询节点

    4. 修改节点

 

  5. 删除节点

 

      6. Watcher监听

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

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

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

     •Curator引入了 Cache 来实现对 ZooKeeper 服务端事件的监听。

    •ZooKeeper提供了三种Watcher：

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

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

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

     6.1  NodeCache 示例

      6.2  PathChildrenCache 示例

 6.3  TreeCache 示例

 7. zookeeper分布式锁原理

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

          1.客户端获取锁时，在lock节点下创建临时顺序节点。（lock节点是举例的节点名）

            临时：保证任何时候哪怕是宕机节点也能被删除, 顺序： 因为要找序号最小的节点获取到锁，其它节点依次监听比它小的那一个节点的删除事件

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

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

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

 

Curator实现分布式锁API

• 在Curator中有五种锁方案：

• InterProcessSemaphoreMutex：分布式排它锁（非可重入锁）

• InterProcessMutex：分布式可重入排它锁

• InterProcessReadWriteLock：分布式读写锁

• InterProcessMultiLock：将多个锁作为单个实体管理的容器

• InterProcessSemaphoreV2：共享信号量

资源在谁手上，分布式锁就加到谁手上。

package com.laoyang.curator;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.concurrent.TimeUnit;

//这里用多线程的方式模拟多个12306的服务
public class Ticket12306 implements Runnable {

    private int tickets = 10; //假定有10张票
    //这里使用curator提供的5种锁当中的InterProcessMutex
    InterProcessMutex lock ;
    CuratorFramework client = null;
    public Ticket12306(){

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,5);
        //调用CuratorFrameworkFactory工厂的builder()方法创建客户端对象
        client = CuratorFrameworkFactory.builder()
                .connectString("192.168.239.135:2181") //设置连接串 ip:port,ip:port...
                .sessionTimeoutMs(60000)       //设置会话超时毫秒数，默认值是60*1000
                .connectionTimeoutMs(15000)   //设置连接超时毫秒数,默认值是 15*1000
                .retryPolicy(retryPolicy)      //重试策略
                .build();
        client.start();         //开启连接

        lock = new InterProcessMutex(client,"/lock");
    }
    @Override
    public void run() {

        while (true){
            //获取锁
            try {
                lock.acquire(1, TimeUnit.SECONDS);
                if (tickets >0){
                    System.out.println(Thread.currentThread()+":"+tickets--);
                    Thread.sleep(200);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //释放锁
                try {
                    lock.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

package com.laoyang.curator;

import org.junit.Test;

public class LockTest {


    public static void main(String[] args){

        Ticket12306 ticket12306 = new Ticket12306();
        Thread t1 = new Thread(ticket12306,"飞猪");
        Thread t2 = new Thread(ticket12306,"携程");
        t1.start();
        t2.start();
    }
}