Zookeeper学习总结2

主要介绍下原生zookeeper客户端API使用、zkClient工具包和分布式锁的简单实现。

1.原生的zookeeper客户端API

导入依赖jar

<dependency>
	<groupId>org.apache.zookeeper</groupId>
	<artifactId>zookeeper</artifactId>
	<version>3.4.13</version>
</dependency>复制代码

实现对zookeeper的基本操作

复制代码

private static String connection = "127.0.0.1:2181";//连接信息
    private static String rootPath ="/study";//目录
    /*创建一个Watcher*/
    private static Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("path:"+event.getPath()+",type:"+event.getType());
            if(event.getType().equals(Event.EventType.NodeChildrenChanged)){//子节点变化触发
                System.out.println("子节点变化触发");
            }else if(event.getType().equals(Event.EventType.NodeCreated)){//创建节点触发
                System.out.println("创建节点触发");
            }else if(event.getType().equals(Event.EventType.NodeDataChanged)){//节点数据变化
                System.out.println("节点数据变化");
            }else if(event.getType().equals(Event.EventType.NodeDeleted)){//节点删除
                System.out.println("节点删除");
            }else {
                System.out.println("其他");
            }

        }
    };
    public static void main(String[] args) throws Exception {

        ZooKeeper zooKeeper = new ZooKeeper(connection,2000,watcher);
        //没有创建父节点时抛出异常 KeeperException$NoNodeException
        /*zooKeeper.create(firstPath+"/test","this is my first create".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);*/
        String firstPath = rootPath;
        //判断节点是否存在，两个参数：1、节点路径；2、是否监控（Watcher即初始化ZooKeeper时传入的Watcher）
        if(zooKeeper.exists(firstPath,true) != null){
            //删除节点
            zooKeeper.delete(firstPath,-1);
        }
        if(zooKeeper.exists(firstPath,true) == null){
            //创建一个持久节点节点 ，四个参数：1、节点路径；2、节点数据；3、节点权限；4、创建模式
            String path = zooKeeper.create(firstPath,"this is my first create".getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

        Stat stat = new Stat();
        //获取该节点下的数据，三个参数：1、节点路径；2、书否监控该节点；3、版本等信息可以通过一个Stat对象来指定
        String data = new String(zooKeeper.getData(firstPath,false,stat));
        System.out.println("data:"+data);
        System.out.println(stat.toString());

        //注册watcher
        zooKeeper.register(watcher);
        //修改节点数据 ，version -1 匹配所有版本
        zooKeeper.setData(firstPath,"update date".getBytes(),-1);
        System.out.println("修改后："+new String(zooKeeper.getData(firstPath,watcher,null)));
        
    //创建子节点，并设置watcher
       String childrenPath = firstPath+"/first";
        //获取孩子节点,并注册watcher监听Event.EventType.NodeChildrenChanged事件
       zooKeeper.getChildren(firstPath,true);
       zooKeeper.exists(childrenPath,watcher);
       zooKeeper.create(childrenPath,"this is a child".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL);//触发连个watcher

        //获取孩子节点,并注册watcher监听Event.EventType.NodeChildrenChanged事件
        List<String> children = zooKeeper.getChildren(firstPath,true);
        System.out.println(children);

        zooKeeper.exists(childrenPath,watcher);
        zooKeeper.setData(childrenPath,"modify child data".getBytes(),-1);
        //判断子节点是否存在，并在子节点注册watcher
        zooKeeper.exists(childrenPath,watcher);
        zooKeeper.delete(childrenPath,-1);//触发了了两个watcher，
        //关闭连接
        zooKeeper.close();

    }复制代码

需要注意的是原生的API创建节点只能在父目录存在时才能创建，删除时也不能递归删除。由于watch机制只触发一次的特性，如果想一直监听节点变化就要不断的注册watch，这对于开发者而言是不太友好的。

2.zkClient API使用

导入依赖jar

<dependency>
	<groupId>com.101tec</groupId>
	<artifactId>zkclient</artifactId>
	<version>0.10</version>
</dependency>复制代码

要使用zkClient必须实现它的序列化接口ZkSerializer

public class MyZkSerializer implements ZkSerializer {

	@Override
	public byte[] serialize(Object data) throws ZkMarshallingError {
		String d = (String) data;
		try {
			return d.getBytes("UTF-8");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
		return null;
	}

	@Override
	public Object deserialize(byte[] bytes) throws ZkMarshallingError {
		try {
			return new String(bytes, "UTF-8");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
		return null;
	}

} 复制代码

代码示例:

public static void main(String[] args) {
	
        // 创建一个zk客户端
        ZkClient client = new ZkClient("localhost:2181");
        String path = "/zkClient";
        client.setZkSerializer(new MyZkSerializer());
        //创建节点
        client.create(path, "123", CreateMode.PERSISTENT);

        // 递归创建节点(持久节点)
        client.createPersistent("/key1/key2/key3",true);
        
        //创建对子节点的监听
        IZkChildListener iZkChildListener = new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                System.out.println(parentPath+"子节点发生变化："+currentChilds);

            }
        };
        //注册watcher,每次触发了watcher事件之后会自动注册，
        client.subscribeChildChanges(path,iZkChildListener);
        
        //取消注册
//        client.unsubscribeChildChanges(path,iZkChildListener);
        IZkDataListener iZkDataListener = new IZkDataListener() {
            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                System.out.println(dataPath+"节点被删除");
            }

            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
                System.out.println(dataPath+"发生变化："+data);
            }
        };
        client.subscribeDataChanges(path, iZkDataListener);
//        if(client.exists(path)){//当前节点是否存在
//            client.deleteRecursive(path);//删除节点，可以递归删除目录下所有子节点
//        }
        //获取数据
        Object data = client.readData(path);
        System.out.println(data);
        //获取子节点列表
        List<String>  children = client.getChildren(path);
        System.out.println(children);

        try {
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            e.printStackTrace();

        }
}复制代码

3.分布式锁的简单实现

分布式锁实现的基于客户端断开连接临时节点自动删除这一特性上实现。

实现思路有两种：

1.创建同一个临时节点，创建成功的获取锁，其他服务对这一个临时节点注册watcher,一旦节点被删除所有注册了watcher的服务都将受到通知进行抢锁。

2.利用临时顺序节点，每一个服务创建一个临时顺序节点，抢锁时由当前 排序 最小的获取锁。其他的注册前一个节点的watcher事件，监听节点删除事件，前一个节点删除则再次发起抢锁操作。

下面一段代码利用临时顺序节点实现的锁

package com.top.learn;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class ZKDistributed implements Lock {
    private String lockPath;
    private ZkClient client;
    /*
     * 利用临时顺序节点来实现分布式锁
     * 获取锁：取排队号（创建自己的临时顺序节点），然后判断自己是否是最小号，如是，则获得锁；不是，则注册前一节点的watcher,阻塞等待
     * 释放锁：删除自己创建的临时顺序节点
     */

    //存放当前节点路径
    ThreadLocal<String> currentPath = new ThreadLocal<>();
    //存放前一个节点路劲
    ThreadLocal<String> beforePath = new ThreadLocal<>();
    public ZKDistributed(String connection,String path){
        //这里可以做一些参数校验
        this.lockPath = path;
        client = new ZkClient(connection);
        if(!client.exists(path)){
            client.createPersistent(path,true);
        }
    }
    @Override
    public boolean tryLock() {
        String path = currentPath.get();
        if(path == null || !client.exists(path)){//未创建过节点
            //创建一个临时顺序节点
            path = client.createEphemeralSequential(lockPath + "/","locked");
            currentPath.set(path);
        }
        //获取所有的自节点
        List<String> children = client.getChildren(lockPath);
        //排序
        Collections.sort(children);
        if(currentPath.get().equals(lockPath + "/" + children.get(0))){
            //当前是最小节点，获取锁
            return true;
        }else {
            int index = children.indexOf(currentPath);
            beforePath.set(children.get(index-1)); //获取上一个节点路径
        }
        return false;
    }
    @Override
    public void lock() {
        while(!tryLock()){
            // 阻塞等待
            waitForLock();
            // 再次尝试加锁
            lock();
        }
    }

    private void waitForLock() {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        IZkDataListener dataListener = new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
            }
            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                //节点删除，取消阻塞
                countDownLatch.countDown();
            }
        };
        //注册对前一个节点的监听
        client.subscribeDataChanges(beforePath.get(),dataListener);
        try {
            if(client.exists(beforePath.get())){//前一个节点存在才等待
                countDownLatch.await();
            }
        }catch (Exception e){

        }
        //取消watcher
        client.unsubscribeDataChanges(beforePath.get(),dataListener);
    }

    @Override
    public void unlock() {
        client.deleteRecursive(currentPath.get());
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }


    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }



    @Override
    public Condition newCondition() {
        return null;
    }
} 复制代码

上面一段代码没有对锁进行可重入实现。