当前位置:   article > 正文

Zookeeper详解(最详细的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支持
  1. <!-- zookeeper支持 -->
  2. <dependency>
  3.    <groupId>org.apache.zookeeper</groupId>
  4.    <artifactId>zookeeper</artifactId>
  5.    <version>3.6.4</version>
  6. </dependency>
  7. <!-- curator-recipes -->
  8. <dependency>
  9.    <groupId>org.apache.curator</groupId>
  10.    <artifactId>curator-recipes</artifactId>
  11.    <version>5.5.0</version>
  12. </dependency>
  13. <!-- curator-framework -->
  14. <dependency>
  15.    <groupId>org.apache.curator</groupId>
  16.    <artifactId>curator-framework</artifactId>
  17.    <version>5.5.0</version>
  18. </dependency>
 
2)、连接zookeeper客户端
  1. //超时重试(连接间隔时间和超时连接次数)
  2. RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
  3. //连接zookeeper对象
  4. client = CuratorFrameworkFactory.newClient(
  5.        "ip:port",
  6.        1000,
  7.        60*1000,
  8.        retryPolicy);
  9. //开始连接
  10. client.start();

3)、创建节点
  1. //1、创建节点并赋值
  2. String path = client.create().forPath("/zuxia","helloworld".getBytes());
  3. System.out.println("创建节点:"+path);
  4. //2、创建节点带子节点(如果不给子节点赋值,子节点的值默认为当前系统的IP地址)
  5. String path = client.create().creatingParentsIfNeeded().forPath("/zuxia/abc");
  6. System.out.println("创建节点:"+path);
  7. //3、创建临时节点(当断开连接时临时节点会自动删除,withMode中的属性可选择)
  8. String path =client.create().withMode(CreateMode.EPHEMERAL).forPath("/a","helloworld".getBytes());
  9. System.out.println("创建节点:"+path);

4)、查询节点
  1. //1、查询节点的数据
  2. byte[] bytes = client.getData().forPath("/zuxia");
  3. System.out.println(new String(bytes));
  4. //2、查询节点的数据(详情信息)
  5. Stat stats=new Stat();
  6. System.out.println(stats);//为了区分两个结果的不同
  7. byte[] be = client.getData().storingStatIn(stats).forPath("/zuxia");
  8. System.out.println(stats);

5)、更新节点
  1. //给节点赋值(返回值为Stat,可写可不写)
  2. client.setData().forPath("/ab", "hello".getBytes());

6)、删除节点
  1. //1、删除节点
  2. System.out.println("删除节点:"+client.delete().forPath("/wjh"));
  3. //2、删除带有子节点的目录节点
  4. 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 监听事件

  1. @Test
  2. void testNodeCache() throws Exception {
  3. // 1. 创建NodeCache
  4. NodeCache nodeCache = new NodeCache(client, "/ab");
  5. // 2. 注册监听
  6. nodeCache.getListenable().addListener(new NodeCacheListener() {
  7.    @Override
  8.    public void nodeChanged() throws Exception {
  9.        System.out.println("/ab节点发生变更");
  10.        byte[] dataBytes = nodeCache.getCurrentData().getData();
  11.        System.out.println("节点修改后的数据:" + new String(dataBytes));
  12.   }
  13. });
  14. // 3. 开启监听,如果设置为true,则开启监听时,加载缓冲数据
  15. nodeCache.start(true);
  16. while(true){}
  17. }

2、PathChildrenCache 监听事件

  1. @Test
  2. void testPathChildrenCache() throws Exception {
  3. //创建监听对象(监听指定节点下的)
  4. PathChildrenCache pathChildrenCache= new PathChildrenCache(client, "/zuxia", true);
  5. //注册监听事件
  6. pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
  7.    @Override
  8.    public void childEvent(CuratorFramework cf, PathChildrenCacheEvent event) throws Exception {
  9.        System.out.println("节点发生变化了");
  10.        PathChildrenCacheEvent.Type type = event.getType();
  11.        //当前判断的是当节点发生更新时进入改方法,可以选择添加或者删除的方法
  12.        if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
  13.            byte[] bytes = event.getData().getData();
  14.            System.out.println("节点修改后的数据"+new String(bytes));
  15.       }
  16.   }
  17. });
  18. //开启监听
  19. pathChildrenCache.start();
  20. while (true){}
  21. }

3、TreeCache 监听事件

  1. @Test
  2. void testTreeCache() throws Exception {
  3.    //创建监听对象
  4.    TreeCache treeCache = new TreeCache(client, "/zuxia");
  5.    //注册监听
  6.    treeCache.getListenable().addListener(new TreeCacheListener() {
  7.        @Override
  8.        public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
  9.            System.out.println("节点发生变化了");
  10.            TreeCacheEvent.Type type = treeCacheEvent.getType();
  11.            if (type.equals(TreeCacheEvent.Type.NODE_ADDED)){
  12.                System.out.println("节点添加了");
  13.           }
  14.       }
  15.   });
  16.    //开启监听
  17.    treeCache.start();
  18.    while (true){}
  19. }

8)、分布式锁实现
  • 首先我们要了解什么是分布式锁?

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

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

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

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

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

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

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

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

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

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

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

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

    并注册监听。

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

    •在Curator中有五种锁方案:

    •InterProcessSemaphoreMutex:分布式排它锁(非可重入锁)

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

    •InterProcessReadWriteLock:分布式读写锁

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

    •InterProcessSemaphoreV2:共享信号量

    方法类:

    1. package com.wjh;
    2. import org.apache.curator.RetryPolicy;
    3. import org.apache.curator.framework.CuratorFramework;
    4. import org.apache.curator.framework.CuratorFrameworkFactory;
    5. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
    6. import org.apache.curator.retry.ExponentialBackoffRetry;
    7. import java.util.concurrent.TimeUnit;
    8. public class TickTest implements Runnable{
    9.    private int x=10;//票数
    10.    //创建分布式可重入排它锁对象
    11.    private InterProcessMutex lock;
    12.    CuratorFramework client;
    13.    //当前方法的构造方法
    14.    public TickTest() {
    15.        //超时重试(连接间隔时间和超时连接次数)
    16.        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
    17.        //连接zookeeper对象
    18.        client = CuratorFrameworkFactory.newClient(
    19.                "ip:port",
    20.                1000,
    21.                60*1000,
    22.                retryPolicy);
    23.        //开始连接
    24.        client.start();
    25.        //创建分布式可重入排它锁对象连接zookeeper注册中心客户端
    26.        //客户端中不用创建,这里会自动创建
    27.        lock = new InterProcessMutex(client, "/lock");
    28.   }
    29.    @Override
    30.    public void run() {
    31.        try {
    32.            //设置锁
    33.            lock.acquire(3, TimeUnit.SECONDS);
    34.            while (true) {
    35.                if(x>0){
    36.                    //输出的调用线程的对象以及票数的数量
    37.                    System.out.println(Thread.currentThread()+"票数:" + x);
    38.                    //间隔200毫秒输出一次
    39.                    Thread.sleep(200);
    40.                    x--;
    41.               }
    42.           }
    43.       } catch (Exception e) {
    44.            throw new RuntimeException(e);
    45.       }finally {
    46.            try {
    47.                //释放锁
    48.                lock.release();
    49.           } catch (Exception e) {
    50.                throw new RuntimeException(e);
    51.           }
    52.       }
    53.   }
    54. }
测试类:
  1. package com.wjh;
  2. public class MaiTest {
  3.    //使用main方法调用
  4.    public static void main(String[] args) {
  5.        //实现线程方法
  6.        TickTest tick = new TickTest();
  7.        //创建线程对象
  8.        Thread t1 = new Thread(tick,"携程");
  9.        Thread t2 = new Thread(tick,"飞猪");
  10.        //启动线程
  11.        t1.start();
  12.        t2.start();
  13.   }
  14. }

三、Zookeeper集群搭建

1、zookeeper集群介绍

Leader选举:

•  Serverid :服务器 ID

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

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

•  Zxid :数据 ID

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

•  Leader 选举的过程中,如果某台 ZooKeeper

            获得了超过半数的选票,

            则此ZooKeeper就可以成为Leader

2、zookeeper集群搭建

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

四、Zookeeper核心理论

在ZooKeeper集群服中务中有三个角色:

•Leader 领导者 :

  1. 处理事务请求

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

•Follower 跟随者 :

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

  2. 参与Leader选举投票

•Observer 观察者:

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

最后附注:

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

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/不正经/article/detail/94727
推荐阅读
相关标签
  

闽ICP备14008679号