zoukankan      html  css  js  c++  java
  • Zookeeper--0300--java操作Zookeeper,临时节点实现分布式锁原理

    删除Zookeeper的java客户端有  :

       1,Zookeeper官方提供的原生API,

       2,zkClient,在原生api上进行扩展的开源java客户端

      3,

    一、Zookeeper原生API

    1,建立连接

    创建会话方法:客户端可以通过创建一个Zookeeper实例来连接zookeeper服务器

    Zookeeper(arguments) 一共4个构造方法,根据参数不同

    connectString :连接服务器列表,用“,”逗号隔开

    sessionTimeout :心跳检测时间周期(毫秒)

    watcher :事件处理通知器

    canBeReadOnly:标识当前会话是否支持只读

     sessionId、sessionPasswd:提供连接zookeeper的sessionId和密码,通过这两个确定唯一一台客户端,目的是提供重复会话

    注意:zookeeper客户端和服务器端会话的建立是一个异步的过程

    代码:

    package com.lhy.zookeeper.base;
    
    import java.util.concurrent.CountDownLatch;
    
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.EventType;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.ZooKeeper;
    
    /**
     * Zookeeper base学习笔记
     * @since 2015-6-13
     */
    public class ZookeeperBase {
    
        /** zookeeper地址 ,连接集群中的一个或多个都行,多个用,隔开*/
        static final String CONNECT_ADDR = "192.168.75.3:2181";
        
        /** session超时时间 */
        static final int SESSION_OUTTIME = 2000;//ms 
        
        /** 
         * 
         * 信号量,阻塞程序执行,用于等待zookeeper连接成功,发送成功信号
         * 
         **/
        static final CountDownLatch connectedSemaphore = new CountDownLatch(1);
        
        
        public static void main(String[] args) throws Exception{
            
            //构造器参数:1:zookeeper连接地址,2:超时时间,3:Watcher:观察者,
            ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher(){
                @Override
                public void process(WatchedEvent event) {
                    //获取事件的状态
                    KeeperState keeperState = event.getState();
                    EventType eventType = event.getType();
                    //SyncConnected表示 连接成功状态
                    if(KeeperState.SyncConnected == keeperState){
                        //刚连接成功什么事件也没有,所以是none
                        if(EventType.None == eventType){
                            //如果建立连接成功,则发送信号量,让后续阻塞程序向下执行
                            connectedSemaphore.countDown();
                            System.out.println("zk 建立连接");
                        }
                    }
                }
            });
    
            //主线程 进行阻塞,等待Zookeeper对象 初始化
            connectedSemaphore.await();
            
            System.out.println("连接zookeeper成功");
            //创建父节点
        //    zk.create("/lhy", "niubei".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            
            //创建子节点
            //zk.create("/lhy/children", "children data".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            
            //获取节点洗信息
            byte[] data = zk.getData("/lhy", false, null);
            System.out.println(new String(data));
            System.out.println(zk.getChildren("/lhy", false));
            
            //修改节点的值
    //        zk.setData("/testRoot", "modify data root".getBytes(), -1);
    //        byte[] data = zk.getData("/testRoot", false, null);
    //        System.out.println(new String(data));        
            
            //判断节点是否存在
    //        System.out.println(zk.exists("/testRoot/children", false));
            //删除节点
    //        zk.delete("/testRoot/children", -1);
    //        System.out.println(zk.exists("/testRoot/children", false));
            
            zk.close();
                
            
        }
        
    }

    主要是通过实例化 Zookeeper 对象进行连接的,看Zookeeper 源码:

    1,connectString 表示连接地址,
    2,sessionTimeout 会话超时时间
     public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
            throws IOException
        {
            this(connectString, sessionTimeout, watcher, false);
        }

    3, Watcher 观察者,查看其源码

    英语的大概意思是,这接口是在客户端连接 Zookeeper 服务端时需要实现的接口,其中注册可很多事件,相当于连接时候的一个回调函数。这个很像swing里的监听事件

    /**
     * This interface specifies the public interface an event handler class must
     * implement. A ZooKeeper client will get various events from the ZooKeeper
     * server it connects to. An application using such a client handles these
     * events by registering a callback object with the client. The callback object
     * is expected to be an instance of a class that implements Watcher interface.
     * 
     */
    public interface Watcher {
      ......      
    }

     需要注意的是:

    实例化 Zookeeper 的过程是异步的,也就是说,在new  Zookeeper对象的同时,代码是继续往下走的,而这样就会出现问题,可能Zookeeper对象还没有初始化,下边的代码就开始调用创建节点、等操作,会出现空指针,所以用到了一个多线程的工具类,CountDownLatch ,可以让程序阻塞等待Zookeeper实例化完成代码再往下走。

    +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

    CountDownLatch用法简单介绍

    CountDownLatch是通过一个计数器来实现的,计数器的初始值为线程的数量。每当一个线程完成了自己的任务后,计数器的值就会减1。当计数器值到达0时,它表示所有的线程已经完成了任务,然后在闭锁上等待的线程就可以恢复执行任务。
    构造器中的计数值(count)实际上就是闭锁需要等待的线程数量。这个值只能被设置一次,而且CountDownLatch没有提供任何机制去重新设置这个计数值。

    与CountDownLatch的第一次交互是主线程等待其他线程。主线程必须在启动其他线程后立即调用CountDownLatch.await()方法。这样主线程的操作就会在这个方法上阻塞,直到其他线程完成各自的任务。

    其他N个线程必须引用闭锁对象,因为他们需要通知CountDownLatch对象,他们已经完成了各自的任务。这种通知机制是通过 CountDownLatch.countDown()方法来完成的;每调用一次这个方法,在构造函数中初始化的count值就减1。所以当N个线程都调 用了这个方法,count的值等于0,然后主线程就能通过await()方法,恢复执行自己的任务。

    CountDownLatch类只提供了一个构造器:

    public CountDownLatch(int count) {  };//参数count为计数值

    然后下面这3个方法是CountDownLatch类中最重要的方法:

    1. public void await() throws InterruptedException { };//调用await()方法的线程会被挂起,它会等待直到count值为0才继续执行
    2. public boolean await( longtimeout, TimeUnit unit)throws InterruptedException { };//和await()类似,只不过等待一定的时间后count值还没变为0的话就会继续执行
    3. public void countDown() { };//将count值减1

    +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

    2,操作节点

    zookeeper增删改查都提供了同步异步 两套方法,来操作节点,同步的等待操作执行完响应结果,异步的方法另起线程执行。异步的方法提供了回调函数来获取一些参数。

    创建:

    读取:

    修改:

     

    删除:

    举例1,同步方式创建:

    参数1,节点路径(名称): /nodeName    --(不允许递归创建节点,父节点不存在不允许创建子节点)

    参数2:节点内容,要求类型是字节数组,(也就是说不支持序列化方式,如果需要实现序列化,可使用java相关序列化框架,如Hessian、Kryo框架)

    参数3,节点权限,使用Ids.OPEN_ACL_UNSAFE 开放权限即可。(这个参数一般在权限没有太高要求的场景下,没必要关注)

    参数4,节点类型:创建节点的类型,CreateMode.*  提供4种节点类型,临时节点本次会话有效,连接关闭就删除

        PERSISTENT   (持久节点)

        PERSISTENT_SEQUENTIAL(持久序列节点) 

        EPHEMERAL(临时节点)

        EPHEMERAL_SEQUENTIAL(临时序列节点)

      创建节点:不能递归创建

           //创建父节点
            String result = zk.create("/lhy", "niubei".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.err.println(result);
            //创建子节点
            zk.create("/lhy/children", "children data".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            

    举例2,异步方式删除

    异步方式在同步方式的基础上加了两个参数

    参数5:注册一个异步回调函数,要实现AsyncCallback.VoidCallback 接口,重写processResult(int rc, String path, Object ctx) 方法,当节点创建完毕后执行此方法。

    回调函数方法参数:

    rc:为服务端的响应码, 0表示调用成功,-4表示端口连接,-110表示指定节点存在,-112表示会话已过期,

    path:接口调用时传入API的数据节点的路径参数

    ctx:为调用接口传入API的ctx值

    name: 实际在服务器端创建节点的名称

     参数6:传递给回调函数的参数,一般为上下文Context信息

     举例3:读取

     读取某个节点

    读取某个节点的第一级子节点,原生api不支持递归读取

    上述代码只能打印aaa bbb  ccc节点数据, 而不能打印aaa1

     举例4:修改节点

    举例5,节点是否存在

    会返回如下版本等信息

    举例6;同步删除节点

     


     小插曲

    +++++++++++++临时节点 实现分布式锁原理++++++++++++

    如下图所示,场景:

    应用集群部署,nginx做负载均衡,集群模拟为tomcat1,tomcat2;数据库是主备,tomcat1、2分别操作DB1,DB2,zookeeper来实现分布式锁

    高并发下,多个请求(模拟2个)想要修改id=1的记录的年龄25-->26,

    第一步:请求1经nginx转发到tomcat2,

    第二步:tomcat2中app中zk客户端会连接zookeeper,创建/user/1 1临时节点(主要是节点名称,值随便)

    第三步:tomcat2修改DB2中uid=1的年龄25改为26,

    第四步:BD2和BD1进行数据同步

    第五步:同步完成,释放zookeeper连接,临时节点清除

    这期间任一时刻,转发到tomcat2的请求,也想修改uid=1的年龄,tomcat2在修改前,会先get("/user/1")一 下,看Zookeeper中有没有/user/1 1的临时节点,因为zookeeper的get性能是很高的,这一步骤很快,如果zoookeeper中有/user/1 1的临时节点,就等待,因为这说明有别的线程正在操作这条数据,直到get不到/user/1 1的节点。说明别的线程都释放锁了,此时也create   /user/1 1的节点,抱着锁,进行数据写操作。

    这样就能实现分布式锁了

    ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

  • 相关阅读:
    Java 零基础跑起第一个程序
    Xcode6 引入第三方静态库project的方法
    Cocos2d-x3.0 RenderTexture(三)
    POJ 题目3461 Oulipo(KMP)
    unity3D游戏开发实战原创视频讲座系列9之塔防类游戏开发第一季
    android 虚拟按键是通过哪种机制上报的?
    深入理解 C 指针阅读笔记 -- 第五章
    【1】按照Django官网,编写一个web app 创建project/配置数据库
    [Swift]LeetCode442. 数组中重复的数据 | Find All Duplicates in an Array
    [Swift]LeetCode441. 排列硬币 | Arranging Coins
  • 原文地址:https://www.cnblogs.com/lihaoyang/p/9227768.html
Copyright © 2011-2022 走看看