Zookeeper详解（最详细的zookeeper解析+项目实例）

zookeeper

在学习zookeeper之前，我们还是先了解它的概念：

一、什么是zookeeper?

1、简介

zookeeper 是一个开放源码的分布式协调服务，主要为了解决分布式架构下数据一致性问题， 它是集群的管理者， 监视着集群中各个节点的状态，根据节点提交的反馈进行下一步合理操作。 最终， 将简单易用的接口和性能高效、 功能稳定的系统提供给用户

2、应用场景

分布式配置中心、分布式注册中心、分布式锁、分布式队列、集群选举、分布式屏障、发布/订阅等场景。

二、zookeeper数据结构

1、简介

zookeeper的数据模型类似于文件系统，是树状结构，每个树节点（目录）对应一个znode节点。这些目录节点和我们普通的目录一样可以新建、删除、修改。

2、常用的数据格式有：

我们常用的主要有四种类型的znode。

1、持久化目录节点：persistent 客户端与zookeeper断开连接后，该节点依旧存在，只要不手动删除该节点，他将永远存在。

2、持久化顺序编号目录节点：persistent_sequential : -s 客户端与zookeeper断开连接后，该节点依旧存在，只是zookeeper给该节点名称进行顺序编号。

3、临时目录节点：ephemeral : -e 客户端与zookeeper断开连接后，该节点被删除。

4、临时顺序编号目录节点： ephemeral_sequential : -es 客户端与zookeeper断开连接后，该节点被删除，只是zookeeper给该节点名称进行顺序编号。

3、zookeeper客户端常用命令：

1）、连接zookeeper服务端（linux）： ./zkcli.sh -server ip:port

2）、断开zookeeper服务端的连接： quit

3）、查看帮助： help

4）、查询所有的目录节点： ls /

5）、创建目录节点： create /节点名 值（可写可不写）

6）、设置目录节点的值（修改时也可以）： set /节点名 值

7）、删除单个目录节点： delete /节点名

8）、删除带有子节点的目录： deleteall /节点名

9）、创建临时目录节点： create -e /节点名 值（可写可不写）

10）、创建持久化目录节点： create -s /节点名 值（可写可不写）

11）、查询目录节点的详情信息： ls -s /节点名

三、zookeeper javaapi（curator）

1、简介：

apache curator 是一个用于apache zookeeper 的java 客户端框架。 curator 提供了一组易于使用的api和工具，简化了与zookeeper 的交互，同时提供了更高级别的抽象和功能。

2、搭建和使用curator（以下环境使用的是spring boot）

1）、引入curator支持

<!-- zookeeper支持 -->
<dependency>
    <groupid>org.apache.zookeeper</groupid>
    <artifactid>zookeeper</artifactid>
    <version>3.6.4</version>
</dependency>
<!-- curator-recipes -->
<dependency>
    <groupid>org.apache.curator</groupid>
    <artifactid>curator-recipes</artifactid>
    <version>5.5.0</version>
</dependency>
<!-- curator-framework -->
<dependency>
    <groupid>org.apache.curator</groupid>
    <artifactid>curator-framework</artifactid>
    <version>5.5.0</version>
</dependency>

2）、连接zookeeper客户端

//超时重试（连接间隔时间和超时连接次数）
retrypolicy retrypolicy = new exponentialbackoffretry(1000, 5);
//连接zookeeper对象
client = curatorframeworkfactory.newclient(
        "ip:port",
        1000,
        60*1000,
        retrypolicy);
//开始连接
client.start();

3）、创建节点

//1、创建节点并赋值
string path = client.create().forpath("/zuxia","helloworld".getbytes());
system.out.println("创建节点："+path);
​
//2、创建节点带子节点（如果不给子节点赋值，子节点的值默认为当前系统的ip地址）
string path = client.create().creatingparentsifneeded().forpath("/zuxia/abc");
system.out.println("创建节点："+path);
​
//3、创建临时节点（当断开连接时临时节点会自动删除,withmode中的属性可选择）
string path =client.create().withmode(createmode.ephemeral).forpath("/a","helloworld".getbytes());
system.out.println("创建节点："+path);

4）、查询节点

//1、查询节点的数据
byte[] bytes = client.getdata().forpath("/zuxia");
system.out.println(new string(bytes));
​
//2、查询节点的数据（详情信息）
stat stats=new stat();
system.out.println(stats);//为了区分两个结果的不同
byte[] be = client.getdata().storingstatin(stats).forpath("/zuxia");
system.out.println(stats);

5）、更新节点

//给节点赋值（返回值为stat,可写可不写）
client.setdata().forpath("/ab", "hello".getbytes());

6）、删除节点

//1、删除节点
system.out.println("删除节点："+client.delete().forpath("/wjh"));
​
//2、删除带有子节点的目录节点
system.out.println("删除子节点："+client.delete().deletingchildrenifneeded().forpath("/zuxia"));

7）、watch事件监听

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

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

•zookeeper提供了三种watcher：

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

pathchildrencache : 监控一个znode的子节点.

treecache : 可以监控整个树上的所有节点，类似于pathchildrencache和nodecache的组合

1、nodecache 监听事件

@test
void testnodecache() throws exception {
// 1. 创建nodecache
nodecache nodecache = new nodecache(client, "/ab");
// 2. 注册监听
nodecache.getlistenable().addlistener(new nodecachelistener() {
    @override
    public void nodechanged() throws exception {
        system.out.println("/ab节点发生变更");
        byte[] databytes = nodecache.getcurrentdata().getdata();
        system.out.println("节点修改后的数据：" + new string(databytes));
    }
});
// 3. 开启监听，如果设置为true，则开启监听时，加载缓冲数据
nodecache.start(true);
while(true){}
}

2、pathchildrencache 监听事件

@test
void testpathchildrencache() throws exception {
//创建监听对象(监听指定节点下的)
pathchildrencache pathchildrencache= new pathchildrencache(client, "/zuxia", true);
//注册监听事件
pathchildrencache.getlistenable().addlistener(new pathchildrencachelistener() {
    @override
    public void childevent(curatorframework cf, pathchildrencacheevent event) throws exception {
        system.out.println("节点发生变化了");
        pathchildrencacheevent.type type = event.gettype();
        //当前判断的是当节点发生更新时进入改方法，可以选择添加或者删除的方法
        if (type.equals(pathchildrencacheevent.type.child_updated)){
            byte[] bytes = event.getdata().getdata();
            system.out.println("节点修改后的数据"+new string(bytes));
        }
    }
});
//开启监听
pathchildrencache.start();
while (true){}
}

3、treecache 监听事件

@test
void testtreecache() throws exception {
    //创建监听对象
    treecache treecache = new treecache(client, "/zuxia");
    //注册监听
    treecache.getlistenable().addlistener(new treecachelistener() {
        @override
        public void childevent(curatorframework curatorframework, treecacheevent treecacheevent) throws exception {
            system.out.println("节点发生变化了");
            treecacheevent.type type = treecacheevent.gettype();
            if (type.equals(treecacheevent.type.node_added)){
                system.out.println("节点添加了");
            }
        }
    });
    //开启监听
    treecache.start();
    while (true){}
}

8）、分布式锁实现

• 首先我们要了解什么是分布式锁？

  在我们进行单机应用开发，涉及并发同步的时候，我们往往采用synchronized或者lock的方式来解决多线程间的代码同步问题，这时多线程的运行都是在同一个jvm之下，没有任何问题。

  但当我们的应用是分布式集群工作的情况下，属于多jvm下的工作环境，跨jvm之间已经无法通过多线程的锁解决同步问题。

  那么就需要一种更加高级的锁机制，来处理种跨机器的进程之间的数据同步问题——这就是分布式锁。

• 

• 其次也要悉知分布式锁的原理：

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

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

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

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

  4.如果发现比自己小的那个节点被删除，则客户端的

  watcher会收到相应通知，此时再次判断自己创建的节点

  是否是lock子节点中序号最小的，如果是则获取到了锁，

  如果不是则重复以上步骤继续获取到比自己小的一个节点

  并注册监听。

• 

• 案例操作----模拟12306售票：

  •在curator中有五种锁方案：

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

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

  •interprocessreadwritelock：分布式读写锁

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

  •interprocesssemaphorev2：共享信号量

  方法类：

  package com.wjh;
  ​
  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;
  ​
  public class ticktest implements runnable{
      private int x=10;//票数
      //创建分布式可重入排它锁对象
      private interprocessmutex lock;
      curatorframework client;
  ​
      //当前方法的构造方法
      public ticktest() {
          //超时重试（连接间隔时间和超时连接次数）
          retrypolicy retrypolicy = new exponentialbackoffretry(1000, 5);
          //连接zookeeper对象
          client = curatorframeworkfactory.newclient(
                  "ip:port",
                  1000,
                  60*1000,
                  retrypolicy);
          //开始连接
          client.start();
          //创建分布式可重入排它锁对象连接zookeeper注册中心客户端
          //客户端中不用创建，这里会自动创建
          lock = new interprocessmutex(client, "/lock");
      }
  ​
      @override
      public void run() {
          try {
              //设置锁
              lock.acquire(3, timeunit.seconds);
              while (true) {
                  if(x>0){
                      //输出的调用线程的对象以及票数的数量
                      system.out.println(thread.currentthread()+"票数:" + x);
                      //间隔200毫秒输出一次
                      thread.sleep(200);
                      x--;
                  }
              }
          } catch (exception e) {
              throw new runtimeexception(e);
          }finally {
              try {
                  //释放锁
                  lock.release();
              } catch (exception e) {
                  throw new runtimeexception(e);
  ​
              }
          }
      }
  }
  ​

测试类：

package com.wjh;
​
public class maitest {
​
    //使用main方法调用
    public static void main(string[] args) {
        //实现线程方法
        ticktest tick = new ticktest();
        //创建线程对象
        thread t1 = new thread(tick,"携程");
        thread t2 = new thread(tick,"飞猪");
        //启动线程
        t1.start();
        t2.start();
    }
}
​

三、zookeeper集群搭建

1、zookeeper集群介绍

leader选举：

          比如有三台服务器，编号分别是1,2,3。

          编号越大在选择算法中的权重越大。

          服务器中存放的最大数据id.值越大说明数据  越新，在选举算法中数据越新权重越大。

            获得了超过半数的选票，

            则此zookeeper就可以成为leader了

2、zookeeper集群搭建

附录文件上有详细搭建步骤.....

四、zookeeper核心理论

在zookeeper集群服中务中有三个角色：

•leader 领导者 ：

1. 处理事务请求

2. 集群内部各服务器的调度者

•follower 跟随者 ：

1. 处理客户端非事务请求，转发事务请求给leader服务器

2. 参与leader选举投票

•observer 观察者：

1. 处理客户端非事务请求，转发事务请求给leader服务器

最后附注：

        以上的集群搭建如过有需要的可以关注私信要文件。我一直都在~~