大家好欢迎来到我的技术博客 在这里我会分享学习笔记、实战经验与技术思考力求用简单的方式讲清楚复杂的问题。 本文将围绕Zookeeper这个话题展开希望能为你带来一些启发或实用的参考。 无论你是刚入门的新手还是正在进阶的开发者希望你都能有所收获文章目录Zookeeper - 基于 Curator 客户端的入门与基础使用 ZooKeeper 与 Curator 简介 环境准备添加 Maven 依赖 Curator 基本使用创建客户端节点操作创建节点创建带数据的节点创建临时节点获取节点数据更新节点数据删除节点节点监听机制使用 NodeCache 监听节点变化使用 PathChildrenCache 监听子节点变化分布式锁的实现使用 InterProcessMutex 实现分布式锁 Curator 操作流程图Mermaid 高级功能与应用场景分布式队列Leader 选举分布式计数器 小结与建议Zookeeper - 基于 Curator 客户端的入门与基础使用 在分布式系统的世界中协调多个节点之间的状态和操作是至关重要的。ZooKeeper 作为一个高性能的协调服务被广泛应用于如 Hadoop、Kafka、HBase 等大型分布式系统中。而 Curator 是 Netflix 公司为简化 ZooKeeper 的使用而开发的一套高级客户端库它封装了 ZooKeeper 的原生 API提供了更加友好、安全和易于使用的接口。本文将带你深入了解如何使用 Curator 客户端与 ZooKeeper 进行交互从基础连接到节点操作再到监听机制和分布式锁的实现逐步构建你对 ZooKeeper 的实战能力。同时文章中将穿插 Java 示例代码帮助你更好地理解每个知识点的实际应用。 ZooKeeper 与 Curator 简介ZooKeeper 是一个开源的分布式协调服务由 Apache 提供其核心功能包括维护节点ZNode树形结构提供节点的创建、删除、读取和写入操作提供 Watcher 监听机制实现分布式锁、选举、配置管理等功能而 Curator 是一个基于 ZooKeeper 的客户端封装库提供了一系列增强功能如自动连接管理重试机制高级抽象如分布式锁、Leader 选举等更简洁的 API 接口Curator 官方文档地址https://curator.apache.org/ 环境准备在开始使用 Curator 之前需要确保以下环境已经准备就绪Java 开发环境JDK 1.8 或更高版本ZooKeeper 服务可以是本地安装的单机版也可以是集群环境Maven 项目用于引入 Curator 依赖添加 Maven 依赖dependenciesdependencygroupIdorg.apache.curator/groupIdartifactIdcurator-framework/artifactIdversion5.7.0/version/dependencydependencygroupIdorg.apache.curator/groupIdartifactIdcurator-recipes/artifactIdversion5.7.0/version/dependency/dependenciescurator-framework是核心库curator-recipes提供了高级功能如分布式锁、选举等。 Curator 基本使用创建客户端使用 Curator 首先要创建一个客户端实例。Curator 提供了CuratorFrameworkFactory来创建客户端。importorg.apache.curator.framework.CuratorFramework;importorg.apache.curator.framework.CuratorFrameworkFactory;importorg.apache.curator.retry.ExponentialBackoffRetry;publicclassCuratorDemo{publicstaticvoidmain(String[]args)throwsException{StringconnectStringlocalhost:2181;intsessionTimeoutMs5000;intconnectionTimeoutMs3000;CuratorFrameworkclientCuratorFrameworkFactory.builder().connectString(connectString).sessionTimeoutMs(sessionTimeoutMs).connectionTimeoutMs(connectionTimeoutMs).retryPolicy(newExponentialBackoffRetry(1000,3)).build();client.start();// 后续操作代码client.close();}}节点操作Curator 提供了丰富的 API 来操作 ZooKeeper 节点包括创建、删除、读取、更新等。创建节点Stringpath/curator-demo;client.create().forPath(path);创建带数据的节点StringdataHello ZooKeeper;client.create().forPath(path,data.getBytes());创建临时节点client.create().withMode(CreateMode.EPHEMERAL).forPath(/temp-node);获取节点数据byte[]dataclient.getData().forPath(path);System.out.println(newString(data));更新节点数据StringnewDataUpdated data;client.setData().forPath(path,newData.getBytes());删除节点client.delete().forPath(path);节点监听机制Curator 提供了多种监听机制包括PathChildrenCache、NodeCache和TreeCache分别用于监听子节点变化、节点数据变化和整个树的变化。使用 NodeCache 监听节点变化importorg.apache.curator.framework.recipes.cache.NodeCache;importorg.apache.curator.framework.recipes.cache.NodeCacheListener;NodeCachenodeCachenewNodeCache(client,path);nodeCache.getListenable().addListener(newNodeCacheListener(){OverridepublicvoidnodeChanged()throwsException{byte[]datanodeCache.getCurrentData().getData();System.out.println(Node changed: newString(data));}});nodeCache.start();使用 PathChildrenCache 监听子节点变化importorg.apache.curator.framework.recipes.cache.PathChildrenCache;importorg.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;importorg.apache.curator.framework.recipes.cache.PathChildrenCacheListener;PathChildrenCachechildrenCachenewPathChildrenCache(client,/parent-node,true);childrenCache.getListenable().addListener(newPathChildrenCacheListener(){OverridepublicvoidchildEvent(CuratorFrameworkclient,PathChildrenCacheEventevent)throwsException{switch(event.getType()){caseCHILD_ADDED:System.out.println(Child added: event.getData().getPath());break;caseCHILD_UPDATED:System.out.println(Child updated: event.getData().getPath());break;caseCHILD_REMOVED:System.out.println(Child removed: event.getData().getPath());break;}}});childrenCache.start();分布式锁的实现Curator 提供了多种分布式锁实现如InterProcessMutex、InterProcessSemaphoreMutex等。使用 InterProcessMutex 实现分布式锁importorg.apache.curator.framework.recipes.locks.InterProcessMutex;StringlockPath/locks/my-lock;InterProcessMutexlocknewInterProcessMutex(client,lockPath);try{if(lock.acquire(10,TimeUnit.SECONDS)){System.out.println(Lock acquired);// 执行业务逻辑}}finally{lock.release();}这个锁是可重入的多个线程可以多次获取同一个锁。 Curator 操作流程图Mermaid下面是一个使用 Mermaid 描述的 Curator 客户端操作流程图渲染错误:Mermaid 渲染失败: Parse error on line 5: ...建节点| E[client.create()] D --|读取节点| -----------------------^ Expecting SQE, DOUBLECIRCLEEND, PE, -), STADIUMEND, SUBROUTINEEND, PIPE, CYLINDEREND, DIAMOND_STOP, TAGEND, TRAPEND, INVTRAPEND, UNICODE_TEXT, TEXT, TAGSTART, got PS 高级功能与应用场景Curator 提供了许多高级功能可以帮助开发者快速构建分布式系统中的协调逻辑。以下是一些常见的高级功能分布式队列Curator 提供了DistributedQueue和DistributedIdQueue来实现分布式队列。importorg.apache.curator.framework.recipes.queue.DistributedQueue;importorg.apache.curator.framework.recipes.queue.QueueBuilder;importorg.apache.curator.framework.recipes.queue.QueueConsumer;importorg.apache.curator.framework.recipes.queue.SimpleQueueConsumer;QueueConsumerStringconsumernewSimpleQueueConsumerString(){OverridepublicvoidconsumeMessage(Stringmessage)throwsException{System.out.println(Consumed: message);}};DistributedQueueStringqueueQueueBuilder.builder(client,consumer,newStringSerializer(),/queue-path).buildQueue();queue.start();// 生产者queue.put(message1);Leader 选举Curator 提供了LeaderSelector来实现简单的 Leader 选举机制。importorg.apache.curator.framework.recipes.leader.LeaderSelector;importorg.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;LeaderSelectorleaderSelectornewLeaderSelector(client,/leader-election,newLeaderSelectorListenerAdapter(){OverridepublicvoidtakeLeadership(CuratorFrameworkclient)throwsException{System.out.println(I am the leader now!);Thread.sleep(5000);// 模拟执行任务}});leaderSelector.autoRequeue();// 自动重新加入选举leaderSelector.start();分布式计数器Curator 提供了DistributedAtomicInteger来实现分布式计数器。importorg.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;DistributedAtomicIntegercounternewDistributedAtomicInteger(client,/counter-path,newRetryNTimes(3,1000));counter.increment();System.out.println(Counter value: counter.get().postValue()); 小结与建议Curator 作为 ZooKeeper 的高级客户端极大地简化了开发者在构建分布式系统时的复杂度。通过本文的学习你应该已经掌握了如何创建和管理 Curator 客户端如何进行节点的基本操作创建、读取、更新、删除如何使用监听机制监听节点和子节点的变化如何实现分布式锁、队列、选举和计数器等高级功能在实际开发中建议使用 Curator 提供的重试机制和连接管理功能以提高系统的稳定性和健壮性。同时合理使用监听机制可以减少不必要的轮询提高系统性能。Curator 官方文档地址https://curator.apache.org/你可以在这里查阅更多 API 和使用示例。在分布式系统中良好的协调机制是系统稳定运行的关键。Curator 为我们提供了一套强大而简洁的工具来实现这一目标。希望本文能帮助你更好地理解和使用 Curator为你的分布式系统开发之路添砖加瓦。如果你对 ZooKeeper 或 Curator 有任何疑问欢迎留言交流 感谢你读到这里 技术之路没有捷径但每一次阅读、思考和实践都在悄悄拉近你与目标的距离。 如果本文对你有帮助不妨 点赞、收藏、分享给更多需要的朋友 欢迎在评论区留下你的想法、疑问或建议我会一一回复我们一起交流、共同成长 关注我不错过下一篇干货我们下期再见✨ Zookeeper - 基于 Curator 客户端的入门与基础使用
大家好欢迎来到我的技术博客 在这里我会分享学习笔记、实战经验与技术思考力求用简单的方式讲清楚复杂的问题。 本文将围绕Zookeeper这个话题展开希望能为你带来一些启发或实用的参考。 无论你是刚入门的新手还是正在进阶的开发者希望你都能有所收获文章目录Zookeeper - 基于 Curator 客户端的入门与基础使用 ZooKeeper 与 Curator 简介 环境准备添加 Maven 依赖 Curator 基本使用创建客户端节点操作创建节点创建带数据的节点创建临时节点获取节点数据更新节点数据删除节点节点监听机制使用 NodeCache 监听节点变化使用 PathChildrenCache 监听子节点变化分布式锁的实现使用 InterProcessMutex 实现分布式锁 Curator 操作流程图Mermaid 高级功能与应用场景分布式队列Leader 选举分布式计数器 小结与建议Zookeeper - 基于 Curator 客户端的入门与基础使用 在分布式系统的世界中协调多个节点之间的状态和操作是至关重要的。ZooKeeper 作为一个高性能的协调服务被广泛应用于如 Hadoop、Kafka、HBase 等大型分布式系统中。而 Curator 是 Netflix 公司为简化 ZooKeeper 的使用而开发的一套高级客户端库它封装了 ZooKeeper 的原生 API提供了更加友好、安全和易于使用的接口。本文将带你深入了解如何使用 Curator 客户端与 ZooKeeper 进行交互从基础连接到节点操作再到监听机制和分布式锁的实现逐步构建你对 ZooKeeper 的实战能力。同时文章中将穿插 Java 示例代码帮助你更好地理解每个知识点的实际应用。 ZooKeeper 与 Curator 简介ZooKeeper 是一个开源的分布式协调服务由 Apache 提供其核心功能包括维护节点ZNode树形结构提供节点的创建、删除、读取和写入操作提供 Watcher 监听机制实现分布式锁、选举、配置管理等功能而 Curator 是一个基于 ZooKeeper 的客户端封装库提供了一系列增强功能如自动连接管理重试机制高级抽象如分布式锁、Leader 选举等更简洁的 API 接口Curator 官方文档地址https://curator.apache.org/ 环境准备在开始使用 Curator 之前需要确保以下环境已经准备就绪Java 开发环境JDK 1.8 或更高版本ZooKeeper 服务可以是本地安装的单机版也可以是集群环境Maven 项目用于引入 Curator 依赖添加 Maven 依赖dependenciesdependencygroupIdorg.apache.curator/groupIdartifactIdcurator-framework/artifactIdversion5.7.0/version/dependencydependencygroupIdorg.apache.curator/groupIdartifactIdcurator-recipes/artifactIdversion5.7.0/version/dependency/dependenciescurator-framework是核心库curator-recipes提供了高级功能如分布式锁、选举等。 Curator 基本使用创建客户端使用 Curator 首先要创建一个客户端实例。Curator 提供了CuratorFrameworkFactory来创建客户端。importorg.apache.curator.framework.CuratorFramework;importorg.apache.curator.framework.CuratorFrameworkFactory;importorg.apache.curator.retry.ExponentialBackoffRetry;publicclassCuratorDemo{publicstaticvoidmain(String[]args)throwsException{StringconnectStringlocalhost:2181;intsessionTimeoutMs5000;intconnectionTimeoutMs3000;CuratorFrameworkclientCuratorFrameworkFactory.builder().connectString(connectString).sessionTimeoutMs(sessionTimeoutMs).connectionTimeoutMs(connectionTimeoutMs).retryPolicy(newExponentialBackoffRetry(1000,3)).build();client.start();// 后续操作代码client.close();}}节点操作Curator 提供了丰富的 API 来操作 ZooKeeper 节点包括创建、删除、读取、更新等。创建节点Stringpath/curator-demo;client.create().forPath(path);创建带数据的节点StringdataHello ZooKeeper;client.create().forPath(path,data.getBytes());创建临时节点client.create().withMode(CreateMode.EPHEMERAL).forPath(/temp-node);获取节点数据byte[]dataclient.getData().forPath(path);System.out.println(newString(data));更新节点数据StringnewDataUpdated data;client.setData().forPath(path,newData.getBytes());删除节点client.delete().forPath(path);节点监听机制Curator 提供了多种监听机制包括PathChildrenCache、NodeCache和TreeCache分别用于监听子节点变化、节点数据变化和整个树的变化。使用 NodeCache 监听节点变化importorg.apache.curator.framework.recipes.cache.NodeCache;importorg.apache.curator.framework.recipes.cache.NodeCacheListener;NodeCachenodeCachenewNodeCache(client,path);nodeCache.getListenable().addListener(newNodeCacheListener(){OverridepublicvoidnodeChanged()throwsException{byte[]datanodeCache.getCurrentData().getData();System.out.println(Node changed: newString(data));}});nodeCache.start();使用 PathChildrenCache 监听子节点变化importorg.apache.curator.framework.recipes.cache.PathChildrenCache;importorg.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;importorg.apache.curator.framework.recipes.cache.PathChildrenCacheListener;PathChildrenCachechildrenCachenewPathChildrenCache(client,/parent-node,true);childrenCache.getListenable().addListener(newPathChildrenCacheListener(){OverridepublicvoidchildEvent(CuratorFrameworkclient,PathChildrenCacheEventevent)throwsException{switch(event.getType()){caseCHILD_ADDED:System.out.println(Child added: event.getData().getPath());break;caseCHILD_UPDATED:System.out.println(Child updated: event.getData().getPath());break;caseCHILD_REMOVED:System.out.println(Child removed: event.getData().getPath());break;}}});childrenCache.start();分布式锁的实现Curator 提供了多种分布式锁实现如InterProcessMutex、InterProcessSemaphoreMutex等。使用 InterProcessMutex 实现分布式锁importorg.apache.curator.framework.recipes.locks.InterProcessMutex;StringlockPath/locks/my-lock;InterProcessMutexlocknewInterProcessMutex(client,lockPath);try{if(lock.acquire(10,TimeUnit.SECONDS)){System.out.println(Lock acquired);// 执行业务逻辑}}finally{lock.release();}这个锁是可重入的多个线程可以多次获取同一个锁。 Curator 操作流程图Mermaid下面是一个使用 Mermaid 描述的 Curator 客户端操作流程图渲染错误:Mermaid 渲染失败: Parse error on line 5: ...建节点| E[client.create()] D --|读取节点| -----------------------^ Expecting SQE, DOUBLECIRCLEEND, PE, -), STADIUMEND, SUBROUTINEEND, PIPE, CYLINDEREND, DIAMOND_STOP, TAGEND, TRAPEND, INVTRAPEND, UNICODE_TEXT, TEXT, TAGSTART, got PS 高级功能与应用场景Curator 提供了许多高级功能可以帮助开发者快速构建分布式系统中的协调逻辑。以下是一些常见的高级功能分布式队列Curator 提供了DistributedQueue和DistributedIdQueue来实现分布式队列。importorg.apache.curator.framework.recipes.queue.DistributedQueue;importorg.apache.curator.framework.recipes.queue.QueueBuilder;importorg.apache.curator.framework.recipes.queue.QueueConsumer;importorg.apache.curator.framework.recipes.queue.SimpleQueueConsumer;QueueConsumerStringconsumernewSimpleQueueConsumerString(){OverridepublicvoidconsumeMessage(Stringmessage)throwsException{System.out.println(Consumed: message);}};DistributedQueueStringqueueQueueBuilder.builder(client,consumer,newStringSerializer(),/queue-path).buildQueue();queue.start();// 生产者queue.put(message1);Leader 选举Curator 提供了LeaderSelector来实现简单的 Leader 选举机制。importorg.apache.curator.framework.recipes.leader.LeaderSelector;importorg.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;LeaderSelectorleaderSelectornewLeaderSelector(client,/leader-election,newLeaderSelectorListenerAdapter(){OverridepublicvoidtakeLeadership(CuratorFrameworkclient)throwsException{System.out.println(I am the leader now!);Thread.sleep(5000);// 模拟执行任务}});leaderSelector.autoRequeue();// 自动重新加入选举leaderSelector.start();分布式计数器Curator 提供了DistributedAtomicInteger来实现分布式计数器。importorg.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;DistributedAtomicIntegercounternewDistributedAtomicInteger(client,/counter-path,newRetryNTimes(3,1000));counter.increment();System.out.println(Counter value: counter.get().postValue()); 小结与建议Curator 作为 ZooKeeper 的高级客户端极大地简化了开发者在构建分布式系统时的复杂度。通过本文的学习你应该已经掌握了如何创建和管理 Curator 客户端如何进行节点的基本操作创建、读取、更新、删除如何使用监听机制监听节点和子节点的变化如何实现分布式锁、队列、选举和计数器等高级功能在实际开发中建议使用 Curator 提供的重试机制和连接管理功能以提高系统的稳定性和健壮性。同时合理使用监听机制可以减少不必要的轮询提高系统性能。Curator 官方文档地址https://curator.apache.org/你可以在这里查阅更多 API 和使用示例。在分布式系统中良好的协调机制是系统稳定运行的关键。Curator 为我们提供了一套强大而简洁的工具来实现这一目标。希望本文能帮助你更好地理解和使用 Curator为你的分布式系统开发之路添砖加瓦。如果你对 ZooKeeper 或 Curator 有任何疑问欢迎留言交流 感谢你读到这里 技术之路没有捷径但每一次阅读、思考和实践都在悄悄拉近你与目标的距离。 如果本文对你有帮助不妨 点赞、收藏、分享给更多需要的朋友 欢迎在评论区留下你的想法、疑问或建议我会一一回复我们一起交流、共同成长 关注我不错过下一篇干货我们下期再见✨