Zookeeper 客户端Curator （分布式锁删除ZNode节点)

曾经终败给现在 2022-04-25 10:26 246阅读 0赞

### Zookeeper=文件系统\+通知机制 ###

Curator是Netflix公司开源的一套zookeeper客户端框架，解决了很多Zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册Watcher和NodeExistsException异常等等。Patrixck Hunt（Zookeeper）以一句“Guava is to Java that Curator to Zookeeper”给Curator予高度评价。

Curator包含了几个包：

*  curator-framework：对zookeeper的底层api的一些封装。
 *  curator-client：提供一些客户端的操作，例如重试策略等。
 *  curator-recipes：封装了一些高级特性，如：Cache事件监听、选举、分布式锁、分布式计数器、分布式Barrier等

> ZooKeeper 分布式锁实现 后有些问题需要删除ZNode节点的空文件夹

进程需要访问共享数据时, `就在"/locks"节点下创建一个sequence类型的子节点, 称为thisPath`. **`当thisPath在所有子节点中最小时, 说明该进程获得了锁. 进程获得锁之后, 就可以访问共享资源了`**

### **数据结构** ###

ZooKeeper数据模型的结构与Unix文件系统很类似，整体上可以看作是一棵树，每个节点称做一个ZNode。

很显然zookeeper集群自身维护了一套数据结构。这个存储结构是一个树形结构，其上的每一个节点，我们称之为"znode"，每一个znode默认能够存储1MB的数据，每个ZNode都可以通过其路径唯一标识，leaf 称为叶子节点 如图所示

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z3azE5ODQwMzAx_size_16_color_FFFFFF_t_70][]

现在的项目中使用  来生成全局的唯一ID和对象锁

现在的问题是znode的存储有限制，当容量大于1MB后容易造成zookeeper宕机，所以需要删除节点数据

删除数据节点
    
    删除一个节点
    
    client.delete().forPath("path");
    
    注意，此方法只能删除叶子节点，否则会抛出异常。
    
    删除一个节点，并且递归删除其所有的子节点
    
    client.delete().deletingChildrenIfNeeded().forPath("path");
    
    删除一个节点，强制指定版本进行删除
    
    client.delete().withVersion(10086).forPath("path");
    
    删除一个节点，强制保证删除
    
    client.delete().guaranteed().forPath("path");
    
    guaranteed()接口是一个保障措施，只要客户端会话有效，那么Curator会在后台持续进行删除操作，直到删除节点成功。
    
    注意：上面的多个流式接口是可以自由组合的，例如：
    
    client.delete().guaranteed().deletingChildrenIfNeeded().withVersion(10086).forPath("path");
    
    读取数据节点数据
    
    读取一个节点的数据内容
    
    client.getData().forPath("path");
    
    注意，此方法返的返回值是byte[ ];
    
    读取一个节点的数据内容，同时获取到该节点的stat
    
    Stat stat = new Stat();
    client.getData().storingStatIn(stat).forPath("path");
    
    更新数据节点数据
    
    更新一个节点的数据内容
    
    client.setData().forPath("path","data".getBytes());
    
    注意：该接口会返回一个Stat实例
    
    更新一个节点的数据内容，强制指定版本进行更新
    
    client.setData().withVersion(10086).forPath("path","data".getBytes());
    
    检查节点是否存在
    
    client.checkExists().forPath("path");
    
    注意：该方法返回一个Stat实例，用于检查ZNode是否存在的操作. 可以调用额外的方法(监控或者后台处理)
    并在最后调用forPath()指定要操作的ZNode获取某个节点的所有子节点路径
    
    client.getChildren().forPath("path");
    
    注意：该方法的返回值为List,获得ZNode的子节点Path列表。 可以调用额外的方法(监控、后台处理或者获
    取状态watch, background or get stat) 并在最后调用forPath()指定要操作的父ZNode

TestController

@RestController
    @Slf4j
    @Api(value="test", description="测试")
    public class TestController {
    
    	private final CuratorFramework curator;
    	private final LockNodeCleanService lockNodeCleanService;
    
    
    
    	public TestController(CuratorFramework curator,LockNodeCleanService lockNodeCleanService) {
    		this.curator = curator;
    		this.lockNodeCleanService =lockNodeCleanService;
    	}
    
    
    	@GetMapping("/internal/zookeeper/listData")
    	public void listData(@RequestParam String zkPath) throws Exception {
    //		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    
    //		Map<String, Long> result = new HashMap<>();
    		for (String string : curator.getChildren().forPath(zkPath)) {
    			String tempPath = zkPath + "/" + string;
    			log.info("***********:"+tempPath);
    		}
    	}
    
    	@GetMapping("/internal/zookeeper/remove")
    	public void listData() throws Exception {
    		lockNodeCleanService.deleteEmptyLockNodes();
    	}
    
    }

LockNodeCleanService

import lombok.AllArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.CollectionUtils;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.framework.recipes.leader.LeaderLatch;
    import org.apache.curator.framework.recipes.locks.InterProcessMutex;
    import org.apache.curator.retry.RetryNTimes;
    import org.apache.zookeeper.data.Stat;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Service;
    
    import java.util.List;
    
    @Service
    @Slf4j
    @AllArgsConstructor
    public class LockNodeCleanService {
        private final CuratorFramework curator;
        private final LeaderLatch leaderLatch;
        private static final String root = "/locks/test";
    
    
        public static void main(String[] args) throws Exception {
            CuratorFramework curator = curator(args[0]);
            InterProcessMutex lock = new InterProcessMutex(curator, "");
            lock.acquire();
            lock.release();
            LockNodeCleanService service = new LockNodeCleanService(curator, null);
            service.deleteEmptyLockNodes();
        }
    
        private static CuratorFramework curator(String zkAddress) {
            CuratorFramework curator = CuratorFrameworkFactory.
                    builder().
                    connectString(zkAddress).
                    sessionTimeoutMs(1000).
                    retryPolicy(new RetryNTimes(3, 1000)).
                    build();
            curator.start();
            return curator;
        }
    
        /**
         * 删除所有空的节点
         */
        @Scheduled(cron = "0 0 4 * * ?")
        public void deleteEmptyLockNodes() {
            if (!leaderLatch.hasLeadership()) {
                log.info("deleteEmptyLockNodes not leader");
                return;
            }
            try {
                //二级目录是否存在
                Stat stat = curator.checkExists().forPath(root);
                if (stat == null) {
                    return;
                }
    
                List<String> folders = curator.getChildren().forPath(root);
                if (CollectionUtils.isEmpty(folders)) {
                    return;
                }
                for (String folder : folders) {
                    //三级路径
                    String folderFullPath = root + "/" + folder;
                    try {
                        clearOneFolder(folderFullPath);
                    } catch (Exception e) {
                        log.info("error when clear folder: {}", folderFullPath, e);
                    }
                }
            } catch (Exception e) {
                log.error("x", e);
    
            }
        }
    
    
        /**
         * 删除一个文件夹下的空节点
         * 一个文件夹就是一种业务对象
         *
         * @param folder
         * @throws Exception
         */
        private void clearOneFolder(String folder) throws Exception {
            for (String nodePath : curator.getChildren().forPath(folder)) {
                //四级路径
                String nodeFullPath = folder + "/" + nodePath;
                List<String> folders = curator.getChildren().forPath(nodeFullPath);
                if (CollectionUtils.isEmpty(folders)) {
                    try {
                        //没有子目录直接删除
                        deleteNode(nodeFullPath, curator);
                    } catch (Exception e) {
                        log.info("fail to delete: {}", nodeFullPath, e);
                    }
                    continue;
                }
                for (String path : folders) {
                    //删除第五级目录
                    String pathFive = nodeFullPath + "/" + path;
                    try {
                        deleteNode(pathFive, curator);
                    } catch (Exception e) {
                        log.info("fail to delete: {}", pathFive, e);
                    }
                }
                try {//删除第四级目录
                    deleteNode(nodeFullPath, curator);
                } catch (Exception e) {
                    log.info("fail to delete: {}", nodeFullPath, e);
                }
    
    
    
    
            }
        }
    
    
        /**
         * 安全删除由于锁造成的空节点
         * 确保节点存在、没有子节点、没有数据、最近修改时间在一小时前
         *
         * @param path
         * @param curator
         * @throws Exception
         */
        public static void deleteNode(String path, CuratorFramework curator) throws Exception {
            Stat stat = curator.checkExists().forPath(path);
            if (stat == null) {
                log.info(path + "  not exist");
                return;
            }
            if (stat.getNumChildren() > 0) {
                log.info(path + "  has children");
                return;
            }
            if (curator.getData().forPath(path).length > 0) {
                log.info(path + "  has data");
                return;
            }
            if (System.currentTimeMillis() - stat.getMtime() < 3600 * 1000) {
                log.info(path + "  very new");
                return;
            }
            curator.delete().inBackground().forPath(path);
            log.info(path + "  is deleted");
        }
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z3azE5ODQwMzAx_size_16_color_FFFFFF_t_70]: /images/20220125/3a8aeb7e4ffa4055b65f9ff20cf3c50a.png