Zookeeper中ZkClient及分布式锁的使用

痛定思痛。 2022-12-15 03:25 114阅读 0赞

## 1.zookeeper安装 ##

直接查看：[原文（windows操作系统下 springboot 集成 kafka）][windows_ springboot _ kafka]windows系统

## 2.zookeeper ##

启动zkCli.cmd操作

![20201015104944877.png][]

2.1   使用 ls 命令来查看当前 ZooKeeper 中所包含的内容

![20201015105323139.png][]

2.2 创建一个新的 znode ，使用 create /username myUserName

![20201015105632270.png][]

2.3  get /username 查看数据

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MzM4MTYy_size_16_color_FFFFFF_t_70][]

2.4 delete /username

![20201015110107648.png][]

## 3.java简单使用zk（转） ##

3.1 引用依赖

<dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
                <version>3.5.5</version>
            </dependency>

3.2 新建ZooKeeperProSync

package com.test.mybatisplus.zk;
    
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.data.Stat;
    
    import java.util.concurrent.CountDownLatch;
    
    /**
     * @Description zookeeper使用
     * @project mybatisplus
     * @author：hf
     * @date：
     * @company:
     */
    public class ZooKeeperProSync implements Watcher {
    
    
        private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
        private static ZooKeeper zk = null;
        private static Stat stat = new Stat();
    
        public static void main(String[] args) throws Exception {
            //zookeeper配置数据存放路径
            String path = "/username";
            //连接zookeeper并且注册一个默认的监听器
            zk = new ZooKeeper("127.0.0.1:2181", 5000, //
                    new ZooKeeperProSync());
            //等待zk连接成功的通知
            connectedSemaphore.await();
            //获取path目录节点的配置数据，并注册默认的监听器
            System.out.println(new String(zk.getData(path, true, stat)));
    
            Thread.sleep(Integer.MAX_VALUE);
        }
       @Override
        public void process(WatchedEvent event) {
            if (Event.KeeperState.SyncConnected == event.getState()) {  //zk连接成功通知事件
                if (Event.EventType.None == event.getType() && null == event.getPath()) {
                    connectedSemaphore.countDown();
                } else if (event.getType() == Event.EventType.NodeDataChanged) {  //zk目录节点数据变化通知事件
                    try {
                        System.out.println("配置已修改，新值为：" + new String(zk.getData(event.getPath(), true, stat)));
                    } catch (Exception e) {
                    }
                }
            }
        }
    
    }

3.3 启动之后，输出username的值

![20201015110532322.png][]

3.4 在启动的zkCli.cmd 中使用命令修改 username的数据

![2020101511072716.png][]

![20201015110715618.png][]

## 4.ZkClient 使用 ##

4.1 引入依赖

<dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
                <version>3.5.5</version>
            </dependency>
            <dependency>
                <groupId>com.101tec</groupId>
                <artifactId>zkclient</artifactId>
                <version>0.10</version>
            </dependency>

4.2 创建ZkClientUtils工具类

package com.test.mybatisplus.util;
    
    
    import org.I0Itec.zkclient.IZkDataListener;
    import org.I0Itec.zkclient.ZkClient;
    import org.I0Itec.zkclient.exception.ZkNodeExistsException;
    import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer;
    import org.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    import org.apache.zookeeper.CreateMode;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    import java.util.Collections;
    import java.util.Comparator;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    
    @Component
    public class ZkClientUtils
    {
    
        private static final Log LOG = LogFactory.getLog(ZkClientUtils.class);
        // 如果是集群直接“，”隔开 CONN_STRING = "127.0.0.1:2181,127.0.0.1:2181,127.0.0.1:2181"
    	//private static String CONN_STRING ="127.0.0.1:2181";
    
    	private static String CONN_STRING ;
    
    	// @Component注入之后 没用statics修饰直接@Value可取到值
    	@Value("${zookeeper.node}")
    	private  String CONN_STRING_TEST ;
    
    
    	// @Component注入之后 用statics修饰 start
    	private static ZkClient zkClient = null;
    
    	@Value("${zookeeper.node}")
    	public  void setConnString(String connString) {
    		CONN_STRING = connString;
    	}
        //end
    
        private static final String LOCK_NODE = "/distributed_lock";
        private static final String CHILDREN_NODE = "/task_";
    
    
    	private ZkClientUtils(){
    
    	}
    
    	public synchronized static ZkClient getInstance(){
    		if(zkClient==null){
    			zkClient = new ZkClient(CONN_STRING, 30000, 30000,
    					new BytesPushThroughSerializer());
    		}
    		return zkClient;
    	}
    	
    	public static String readData(String path)
    	{			
    
    		byte[] readBytes = getInstance().readData(path);
    		
    		return new String(readBytes);
    	}
    
    	public static String create(String path,String data)
    	{
    
    		String result = getInstance().create(path,data.getBytes(), CreateMode.EPHEMERAL);
    		//一但创建node即监听情况
    		subscribeDataChanges(path);
    		return result;
    	}
    
    	public static boolean delete(String path)
    	{
    
    		boolean result = getInstance().delete(path);
    
    		return result;
    	}
    
    	public static void write(String path,String data)
    	{
    		 getInstance().writeData(path,data.getBytes());
    
    	}
    
    	/**
    	 * 监控当前节点情况 path
    	 * @param path
    	 */
    	public static void subscribeDataChanges(String path)
    	{
    		getInstance().subscribeDataChanges(path, new IZkDataListener() {
    
    			@Override
    			public void handleDataDeleted(String dataPath) throws Exception {
    				System.out.println("subscribeDataChanges    handleDataDeleted");
    				System.out.println("节点被删除");
    			}
    
    			@Override
    			public void handleDataChange(String dataPath, Object data) throws Exception {
    				System.out.println("subscribeDataChanges    handleDataChange");
    				System.out.println("变更的节点为:" + dataPath + ", 变更内容为:" + data.toString());
    			}
    		});
    
    	}
    
        //    首先zookeeper中我们可以创建一个/distributed_lock持久化节点
        //    然后再在/distributed_lock节点下创建自己的临时顺序节点，比如：/distributed_lock/task_00000000008
        //    获取所有的/distributed_lock下的所有子节点，并排序
        //    判读自己创建的节点是否最小值（第一位）
        //    如果是，则获取得到锁，执行自己的业务逻辑，最后删除这个临时节点。
        //    如果不是最小值，则需要监听自己创建节点前一位节点的数据变化，并阻塞。
        //    当前一位节点被删除时，我们需要通过递归来判断自己创建的节点是否在是最小的，如果是则执行5）；如果不是则执行6）（就是递归循环的判断）
        /**
         * 分布式锁使用
         * @return
         */
        public static String getLock() {
            try {
                // 判断是否存在父节点 --创建永久节点才能创建子节点
                if(!getInstance().exists(LOCK_NODE)){
                    getInstance().create(LOCK_NODE, "分布式锁节点".getBytes(), CreateMode.PERSISTENT);
                }
                // 1。在Zookeeper指定节点下创建临时顺序节点 --当失去连接或者会话过期就会自动删除
                String lockName = getInstance().createEphemeralSequential(LOCK_NODE + CHILDREN_NODE, "children".getBytes());
                // 尝试获取锁
                acquireLock(lockName);
                return lockName;
            } catch(Exception e) {
                e.printStackTrace();
            }
    
            return null;
        }
    
        /**
         * 获取锁
         * @throws InterruptedException
         */
    //    执行日志：
    //    获取到锁：/distributed_lock/task_0000000063
    //    ZkClientUtils - ...... 未获取到锁，阻塞等待 。。。。。。
    //    ZkClientUtils - ...... 未获取到锁，阻塞等待 。。。。。。
    //    ZkClientUtils - ...... 未获取到锁，阻塞等待 。。。。。。
    //    ZkClientUtils - ...... 未获取到锁，阻塞等待 。。。。。。
    //    ZkClientUtils - 。。。。。。前一个节点被删除（删除/distributed_lock/task_0000000063节点）  。。。。。。
    //    ZkClientUtils - 获取到锁：/distributed_lock/task_0000000064
    //    ZkClientUtils - ...... 未获取到锁，阻塞等待 。。。。。。
    //    ZkClientUtils - 。。。。。。前一个节点被删除  。。。。。。
    //    ZkClientUtils - 获取到锁：/distributed_lock/task_0000000065
        public static Boolean acquireLock(String lockName) throws InterruptedException {
            // 2.获取lock节点下的所有子节点
            List<String> childrenList = getInstance().getChildren(LOCK_NODE);
            // 3.对子节点进行排序，获取最小值
            Collections.sort(childrenList, new Comparator<String>() {
                @Override
                public int compare(String o1, String o2) {
                    return Integer.parseInt(o1.split("_")[1]) - Integer.parseInt(o2.split("_")[1]);
                }
            });
            // 4.判断当前创建的节点是否在第一位
            int lockPostion = childrenList.indexOf(lockName.split("/")[lockName.split("/").length - 1]);
            if(lockPostion < 0) {
                // 不存在该节点
                throw new ZkNodeExistsException("不存在的节点：" + lockName);
            } else if (lockPostion == 0) {
                // 获取到锁
                LOG.info("获取到锁：" + lockName);
                return true;
            } else if (lockPostion > 0) {
                // 未获取到锁，阻塞
                LOG.info("...... 未获取到锁，阻塞等待 。。。。。。");
                // 5.如果未获取得到锁，监听当前创建的节点前一位的节点
                final CountDownLatch latch = new CountDownLatch(1);
                IZkDataListener listener = new IZkDataListener() {
    
                    @Override
                    public void handleDataDeleted(String dataPath) throws Exception {
                        // 6.前一个节点被删除,当不保证轮到自己
                        LOG.info("。。。。。。前一个节点被删除  。。。。。。");
                        acquireLock(lockName);
                        latch.countDown();
                    }
    
                    @Override
                    public void handleDataChange(String dataPath, Object data) throws Exception {
                        // 节点内容被修改
                    }
                };
                try {
                    getInstance().subscribeDataChanges(LOCK_NODE + "/" + childrenList.get(lockPostion - 1), listener);
                    latch.await();
                } finally {
                    getInstance().unsubscribeDataChanges(LOCK_NODE + "/" + childrenList.get(lockPostion - 1), listener);
                }
            }
            return false;
        }
    
    }

4.3 方法调用

@ApiOperation(value = "getZkData",notes = "getZkData")
        @RequestMapping(value ="/getZkData" , method = RequestMethod.GET,produces = {"application/json;charset=UTF-8"})
        public String getZkData(HttpServletRequest request,
                                HttpServletResponse response,
                                @ApiParam(value = "path") @RequestParam(required = true) String path
                               ){
          String result=  ZkClientUtils.readData(path);
          return  result;
        }
    
        @ApiOperation(value = "createZk",notes = "createZk")
        @RequestMapping(value ="/createZk" , method = RequestMethod.GET,produces = {"application/json;charset=UTF-8"})
        public String createZk(HttpServletRequest request,
                                HttpServletResponse response,
                                @ApiParam(value = "path") @RequestParam(required = true) String path,
                                @ApiParam(value = "data") @RequestParam(required = true) String data
        ){
            String result=  ZkClientUtils.create(path,data);
            return  result;
        }

## 3 zoopeeper的应用 ##

### 3.1 共享配置文件 ###

## 3.2 zoopeeper跨进程，跨主机，跨网络的共享资源使用同步锁 ##

ZkClientUtils.getLock()

@ApiOperation(value = "getLock",notes = "getLock")
        @RequestMapping(value ="/getLock" , method = RequestMethod.GET,produces = {"application/json;charset=UTF-8"})
        public String getLock(HttpServletRequest request,
                              HttpServletResponse response
        ){
           String  name = ZkClientUtils.getLock();
           String  result = "获取到锁："+name;
           //业务逻辑
           //获取到 直接删除当前子节点
            //        boolean flag= ZkClientUtils.delete(name);
            //        if(flag){
            //            result+="删除"+name+"成功！";
            //        }else{
            //            result+="删除"+name+"失败！！！！";
            //        }
           return  result;
        }

[windows_ springboot _ kafka]: https://mp.csdn.net/console/editor/html/105006861
[20201015104944877.png]: /images/20221123/5e0d2b89c93347809c8b82ae326fdf93.png
[20201015105323139.png]: /images/20221123/2344a0186ec44cf8a6b53acc895c9f45.png
[20201015105632270.png]: /images/20221123/b677eaa0061640488d0f1f07051ce3cf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MzM4MTYy_size_16_color_FFFFFF_t_70]: /images/20221123/c14e6037cf31499cbffaef5e4488e80b.png
[20201015110107648.png]: /images/20221123/b3b83ce9a4ae4c02abd05f33de37c616.png
[20201015110532322.png]: /images/20221123/6638ea14f5a74397bcd0fc6d218385df.png
[2020101511072716.png]: /images/20221123/67c40f57d53c4c87a528ec6708ea9b32.png
[20201015110715618.png]: /images/20221123/9ed077ac16784deb8467256a93fcb5ce.png