1.zookeeper主要功能 1.1 配置管理

分布式项目的配置总管理处

1.2 分布式锁

对于分布式项目修改共享数据时加入锁管理(同一时间只能有一个服务对数据进更改)

1.3 集群管理

最常见的功能,作为注册中心使用.

2.zookeeper命令操作 2.1 zookeeper数据模型

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

2.2 zookeeper 服务端常用命令

启动Zookeeper服务 : ./zkServer.sh start
查看Zookeeper服务状态 : ./zkServer.sh statu
停止Zookeeper服务 : ./zkServer.sh stop
重启Zookeeper服务 : ./zkServer.sh restart

2.3 zookeeper 客户端常用命令

连接ZooKeeper服务端
- ./zkCli.sh -server ip:port
断开客户端连接
- quit
设置节点的值
- set /节点path value
查看帮助命令
- help
删除单个节点
- delete /节点path
显示指定目录下的节点
- ls 目录
删除带有子节点的节点
- deleteall /节点path
创建节点
- create /节点path value
获取节点的值
- get /节点path
创建临时节点
- create -e /节点path value
创建顺序节点
- create -s /节点path value
查询节点详细信息
- ls -s /节点path

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

3.Java API-Cuator

Curator是Apache ZooKeeper的Java客户端库

4.Java API 常用操作

建立连接
添加节点
删除节点
修改节点
查询节点
Watch时间监听
分布式锁实现

4.1 建立连接 4.1.1 方式一

        
        //重试策略1 每间隔x一共重试x次  每隔3秒连接一次一共连接10次
        RetryPolicy r = new ExponentialBackoffRetry(3000, 10);
        //第一种方式
        Curatorframework zookeeperClient = CuratorframeworkFactory.newClient("116.62.71.234:2181",
                60 * 1000,
                15 * 1000,
                r);
        //开启连接
        zookeeperClient.start();

4.1.2 方式二 : 链式编程

    //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.234:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

namespace : 创建根节点itheima(为了方便,不用之后每次进行客户端操作都写/根节点path)

4.2 添加节点 4.2.1 基本创建

 public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.235:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

        
        
        String path = zookeeperClient.create().forPath("/app1");
        System.out.println(path);
        //关闭连接
        zookeeperClient.close();
    }

4.2.2 创建节点带有数据

如果没有节点信息,则信息默认为客户端ip

public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.235:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();
        //开启连接
        zookeeperClient.start();
        
        
        String path = zookeeperClient.create().forPath("/app2","hehe".getBytes());
        System.out.println(path);
        //关闭连接
        zookeeperClient.close();
    }

4.2.3 创建节点,设置节点类型

默认为持久化类型

public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.235:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

        
        
        String path = zookeeperClient.create().withMode(CreateMode.EPHEMERAL).forPath("/app3","hehe".getBytes());
        System.out.println(path);


        //关闭连接  app3就会被释放
        zookeeperClient.close();
    }

4.2.4 创建多级节点

ZooKeeper如果父节点不存在无法创建子节点

  public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.235:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

        
        
        String path = zookeeperClient.create().creatingParentsIfNeeded().forPath("/app4/p1","hehe".getBytes());
        System.out.println(path);


        //关闭连接  app3就会被释放
        zookeeperClient.close();
    }

4.3 查询节点 4.3.1 查询节点数据

public static void main(String[] args) throws Exception {
            //第二种连接方式:链式编程
            Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                    connectString("116.62.71.235:2181").
                    sessionTimeoutMs(60 * 1000).
                    connectionTimeoutMs(15 * 1000).
                    retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                    namespace("itheima").build();

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

            
            byte[] bytes = zookeeperClient.getData().forPath("/app1");
            System.out.println(new String(bytes));
            //关闭连接
            zookeeperClient.close();
        }

4.3.2 查询节点子节点

public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.235:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

        
        List strings = zookeeperClient.getChildren().forPath("/app1");
        //关闭连接
        zookeeperClient.close();
    }

4.3.3 查询节点详细信息 2.3详见详细信息

返回的结果需要封装到一个JavaBean中

 public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.235:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

        
        
        Stat status = new Stat();
        zookeeperClient.getData().storingStatIn(status).forPath("/app1");
        //关闭连接
        zookeeperClient.close();
    }

4.4 修改节点信息 4.4.1 基本修改

  public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.235:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

        
        zookeeperClient.setData().forPath("/app1","itcast".getBytes());
        //关闭连接
        zookeeperClient.close();
    }

4.4.2 根据版本修改节点信息(推荐)

 public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.234:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

        
        
        Stat status = new Stat();
        zookeeperClient.getData().storingStatIn(status).forPath("/app1");
        //查询出版本
        int version = status.getVersion();
        System.out.println(version);
        zookeeperClient.setData().withVersion(3).forPath("/app1","itcast".getBytes());
        //关闭连接
        zookeeperClient.close();
    }

4.5 删除节点 4.5.1 基本删除

 public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.234:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

        
        zookeeperClient.delete().forPath("/app1");
        
        //关闭连接
        zookeeperClient.close();
    }

4.5.2 带有子节点的递归删除

 public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.234:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

        zookeeperClient.delete().deletingChildrenIfNeeded().forPath("/app1");

        //关闭连接
        zookeeperClient.close();
    }

4.5.3 必须删除成功

guaranteed();

public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.234:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

        //开启连接
        zookeeperClient.start();
        
        zookeeperClient.delete().guaranteed().forPath("/app1");

        //关闭连接
        zookeeperClient.close();
    }

4.5.4 删除回调

inBackground(()->{})

public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.234:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

        //开启连接
        zookeeperClient.start();
        
        zookeeperClient.delete().guaranteed().inBackground(new BackgroundCallback(){
            @Override
            public void processResult(Curatorframework client, CuratorEvent event) throws Exception {
                System.out.println("我被删除了");
                System.out.println(event);
            }
        }).forPath("/app1");

        //关闭连接
        zookeeperClient.close();
    }

5.Java API 高级操作 5.1 Watch事件监听 5.1.1 基本概念

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

5.1.2 NodeCache

   public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.234:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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

        
        //1.创建NodeCache对象
        NodeCache nodeCache = new NodeCache(zookeeperClient, "/app1");
        //2.注册监听
        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));
            }
        });
        //3.开启监听,如果设置为true,则开启监听时,加载缓存数据
        nodeCache.start();;

        //while (true){} //保证线程不结束 , web环境不需要加,这里是主方法演示
       
       //关闭连接
        zookeeperClient.close();
    }

5.1.3 PathChildrenCache

参数event可以获得修改的详细信息

 public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.234:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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


        
        //1.创建PathChildrenCache对象
        PathChildrenCache p = new PathChildrenCache(zookeeperClient,"/app2",true);
        //2.注册监听
        p.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(Curatorframework client, PathChildrenCacheEvent event) throws Exception {
                System.out.println("子节点发送变化了");
                System.out.println(event);
                //监听子节点的数据变更,并且拿到变更后的数据
                //1.获取类型
                PathChildrenCacheEvent.Type type = event.getType();
                //2.判断类型是否为update
                if(type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
                    System.out.println("数据变了");
                    byte[] data = event.getData().getData();
                    System.out.println(new String(data));
                }
            }
        });
        //3.开启监听,如果设置为true,则开启监听时,加载缓存数据
        p.start();

        //while (true){} //保证线程不结束 , web环境不需要加,这里是主方法演示
        //关闭连接
        zookeeperClient.close();
    }

5.1.4 TreeCache

public static void main(String[] args) throws Exception {
        //第二种连接方式:链式编程
        Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
                connectString("116.62.71.234:2181").
                sessionTimeoutMs(60 * 1000).
                connectionTimeoutMs(15 * 1000).
                retryPolicy(new ExponentialBackoffRetry(3000, 10)).
                namespace("itheima").build();

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


        
        //1.创建NodeCache对象
        TreeCache treeCache = new TreeCache(zookeeperClient, "/app1");
        //2.注册监听
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(Curatorframework client, TreeCacheEvent event) throws Exception {
                System.out.println("结点变化了");
                System.out.println(event);
            }
        });
        //3.开启监听,如果设置为true,则开启监听时,加载缓存数据
        treeCache.start();
        //关闭连接
        zookeeperClient.close();

        //while (true){} //保证线程不结束 , web环境不需要加,这里是主方法演示

    }

6.分布式锁

6.1 分布式锁

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

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

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

6.2 分布式锁原理

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

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

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

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

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

6.3 模拟12306售票案例

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

Runnable

public class Ticket12306 implements Runnable {
    private int tickets = 10;


    //第二种连接方式:链式编程
   private static Curatorframework zookeeperClient = CuratorframeworkFactory.builder().
            connectString("116.62.71.235:2181").
            sessionTimeoutMs(60 * 1000).
            connectionTimeoutMs(15 * 1000).
            retryPolicy(new ExponentialBackoffRetry(3000, 10)).
            build();


    private static InterProcessMutex lock = new InterProcessMutex(zookeeperClient,"/lock") ;


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


        }
    }
}

Main

public class Lock12306 {
    public static void main(String[] args) {
        Ticket12306 t = new Ticket12306();
        Thread t1 = new Thread(t,"携程");
        Thread t2 = new Thread(t,"飞猪");
        t1.start();
        t2.start();
    }
}

7 ZooKeeper集群 7.1 集群简介

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

7.2 Zookeer集群搭建

与创建单机环境类似
修改conf目录下的zoo_sample.cfg为zoo.cfg
修改data目录为指定目录,每个集群成员分别配置
在data目录下创建myid文件,内容分别为1,2,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地址：服务器之间通信端口：服务器之间投票选举端口,搭建伪集群,端口可以写成127.0.0.1

7.3 Zookeeper集群角色

Leader领导者
- 处理事务请求(增删改)
- 集群内部各个服务器的调度者
Follower跟随者
- 处理客户端非事务请求(查),转发事务请求给Leader服务器.
- 参数Leader选举投票
Observer观察者
- 处理客户端非事务请求,转发事务请求给Leader服务器.

ZooKeeper-B站黑马程序员,多多三连

6.分布式锁

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