ZooKeeper 客户端API操作

文章目录

  • 一、节点信息
    • 1、创建节点
    • 2、获取子节点并监听节点变化
    • 3、判断节点是否存在
    • 4、客户端向服务端写入数据
      • 写入请求直接发给 Leader 节点
      • 写入请求直接发给 follow 节点
  • 二、服务器动态上下线监听
    • 1、监听过程
    • 2、代码
  • 三、分布式锁
    • 1、什么是分布式锁?
    • 2、Curator 框架实现分布式锁

一、节点信息

前提:centos102、centos103、centos104 服务器都已经开启

pom.xml 依赖

<dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.13.2</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-core</artifactId><version>2.17.1</version></dependency><dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.5.7</version></dependency><dependency><groupId>org.junit.jupiter</groupId><artifactId>junit-jupiter</artifactId><version>RELEASE</version><scope>compile</scope></dependency>
</dependencies>

log4j.properties 配置

# 设置全局的日志记录级别为 INFO
log4j.rootLogger=INFO, stdout# 控制台输出
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n# 文件输出
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

1、创建节点

zkClient.java 代码

// 注意:逗号后面不能有空格
private String connectString = "centos102:2181,centos103:2181,centos104:2181";
private int sessionTimeout = 2000;
private ZooKeeper zkClient;// 创建客户端
@Before
public void init() throws IOException {zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent event) {}});
}// 创建子节点
@Test
public void create() throws InterruptedException, KeeperException {String nodeCreated = zkClient.create("/frost", "cat".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}

运行创建子节点,看看是否创建了该节点

在这里插入图片描述

2、获取子节点并监听节点变化

@Test
public void getChildren() throws InterruptedException, KeeperException {List<String> children = zkClient.getChildren("/", true);for (String child : children) {System.out.println(child);}
}

那如果此时我再创建一个节点,此时控制台没有任何变化,我想要创建一个节点控制台能够看到相关变化怎么办?此时只需要将程序保持不结束,然后将客户端查看子节点函数放入监听器中。

3、判断节点是否存在

@Test
public void exit() throws InterruptedException, KeeperException {Stat stat = zkClient.exists("/frost", false);System.out.println(stat == null ? "not exits" : "exits");
}

4、客户端向服务端写入数据

写入请求直接发给 Leader 节点

  1. 客户端发送写入请求,leader节点执行写入操作
  2. leader通知follow1执行写入操作
  3. folllow1写入完毕给leader返回确认ack
  4. 现在半数以上服务器完成写入,leader给客户端发送确认ack
  5. leader通知follow2写入
  6. follow2写入完毕给leader发送确认ack
    在这里插入图片描述

写入请求直接发给 follow 节点

  1. 客户端发送写入请求,
  2. follow1 将写入请求发送给leader
  3. leader节点执行写入操作,然后leader通知follow1执行写入操作
  4. folllow1写入完毕给leader返回确认ack
  5. 现在半数以上服务器完成写入,leader给follow1发送确认ack
  6. follow1给客户端发送确认ack
  7. leader通知follow2写入
  8. follow2写入完毕给leader发送确认ack

在这里插入图片描述

二、服务器动态上下线监听

1、监听过程

在这里插入图片描述

以下红色字体写错,应该是下线则通知注册监听器的客户端
在这里插入图片描述

对于ZooKeeper集群来说,客户端和服务器都相当于客户端,区别在于:服务器在ZooKeeper集群中是创建节点,客户端在ZooKeeper是监听信息。

在这里插入图片描述

2、代码

服务器注册到zk集群

import org.apache.zookeeper.*;
import java.io.IOException;public class DistributeServer {private String connectString = "centos102:2181,centos103:2181,centos104:2181";private int sessionTimeout = 2000;ZooKeeper zk;public static void main(String[] args) throws IOException, InterruptedException, KeeperException {DistributeServer server = new DistributeServer();// 1. 获取zk连接server.getConnect();// 2. 注册服务器到 zk 集群server.regist(args[0]);// 3. 启动业务逻辑(睡觉)server.business();}private void business() throws InterruptedException {Thread.sleep(Long.MAX_VALUE);}private void regist(String hostname) throws InterruptedException, KeeperException {String create = zk.create("/servers", hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); // 临时带序号的节点System.out.println(hostname + "is online");}private void getConnect() throws IOException {zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent event) {}});}
}

客户端进行监听

import org.apache.zookeeper.*;import java.io.IOException;
import java.util.ArrayList;
import java.util.List;public class DistributeClient {private String connectString = "centos102:2181,centos103:2181,centos104:2181";private int sessionTimeout = 2000;ZooKeeper zk;public static void main(String[] args) throws IOException, InterruptedException, KeeperException {DistributeClient client = new DistributeClient();// 1. 获取zk连接client.getConnect();// 2. 监听/servers下子节点的增加和删除client.getServerList();// 3. 启动业务逻辑(睡觉)client.business();}private void business() throws InterruptedException {Thread.sleep(Long.MAX_VALUE);}private void getServerList() throws InterruptedException, KeeperException {List<String> children = zk.getChildren("/servers", true);ArrayList<String> servers = new ArrayList<>();for (String child : children) {byte[] data = zk.getData("/servers/" + child, false, null);servers.add(new String(data));}System.out.println(servers);}private void getConnect() throws IOException {zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent event) {try {getServerList();} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();}}});}
}

启动客户端,然后在服务器上进行增加节点监听
在这里插入图片描述

删除节点监听
在这里插入图片描述

因为我们服务端的代码传参了,所以我们需要设置一下这个参数:
在这里插入图片描述

下图代表服务端启动的是hadoop102节点
在这里插入图片描述

先把客户端启动起来,发现有一个节点hadoop101:
在这里插入图片描述

在启动服务端,hadoop102上线:
在这里插入图片描述

然后返回看客户端的监听,发现节点有变化,打印出所有节点 [hadoop102, hadoop101]
在这里插入图片描述

此时我们修改一下再此启动服务端让 hadoop103 上线:
在这里插入图片描述

返回客户端查看发现 hadoop102 下线,hadoop103 上线
在这里插入图片描述

三、分布式锁

1、什么是分布式锁?

比如说"进程1"在使用该资源的时候,会先去获得锁,"进程1"获得锁以后会对该资源保持独占,这样其他进程就无法访问该资源,"进程1"用完该资源以后就将锁释放掉,让其他进程来获得锁,那么通过这个锁机制,我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

在这里插入图片描述

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;public class DistributedLock {private final String connectString = "centos102:2181,centos103:2181,centos104:2181";private final int sessionTimeout = 2000;ZooKeeper zk;private CountDownLatch connectLatch = new CountDownLatch(1);private CountDownLatch waitLatch = new CountDownLatch(1);// 前一个节点private String waitPath;// 当前节点String currentMode;public DistributedLock() throws IOException, InterruptedException, KeeperException {// 1. 获取连接zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent event) {// connectLatch,如果连接上zk,可以释放if (event.getState() == Event.KeeperState.SyncConnected) {connectLatch.countDown();}// waitLatch,需要释放if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {waitLatch.countDown();}}});// 等待zk正常连接后往下走connectLatch.await();// 2. 判断根节点/lock是否存在Stat stat = zk.exists("/locks", false);if (stat == null) {// 创建根节点(永久节点)zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}}// 对 zk 加锁public void zkLock() {// 创建对应的临时带序号的节点try {currentMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);// 判断创建的节点是否是最小的序号节点,如果是,获取到锁;如果不是,监听前一个节点List<String> children = zk.getChildren("/locks", false);// 如果 children 只有一个节点,直接获取锁;如果有多个节点,需要判断,谁最小if (children.size() == 1) {return;}else {// 排序Collections.sort(children);// 获取节点名称seq00000001String thisNode = currentMode.substring("/locks/".length());// 通过seq00000001获取该节点在children当中的位置int index = children.indexOf(thisNode);if (index == -1) {System.out.println("数据异常");}else if (index == 0) {// 该节点为第一个,获取锁直接返回return;}else {// 不是第一个,监听前一个节点waitPath = "/locks/" + children.get(index - 1);zk.getData(waitPath, true, null);// 等待监听结束waitLatch.await();return;}}} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}// 解锁public void unZkLock() throws InterruptedException, KeeperException {// 删除节点zk.delete(currentMode, -1);}
}

测试

import org.apache.zookeeper.KeeperException;import java.io.IOException;public class DistributedLockTest {public static void main(String[] args) throws IOException, InterruptedException, KeeperException {final DistributedLock lock1 = new DistributedLock();final DistributedLock lock2 = new DistributedLock();new Thread(new Runnable() {@Overridepublic void run() {try {lock1.zkLock();System.out.println("线程1启动,获取到锁");Thread.sleep(5000);lock1.unZkLock();System.out.println("线程1释放锁");} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();}}}).start();new Thread(new Runnable() {@Overridepublic void run() {try {lock2.zkLock();System.out.println("线程2启动,获取到锁");Thread.sleep(5000);lock2.unZkLock();System.out.println("线程2释放锁");} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();}}}).start();}}

2、Curator 框架实现分布式锁

原生JAVA API出现的问题:
(1)会话是异步的,需要自己去连接
(2)Watch需要重复注册,不然就不能生效
(3)开发的复杂性还是比较高
(4)不支持多节点的删除和创建,需要自己去递归

Curator 是一个专门解决分布式锁的框架,解决了原生JAVA API开发分布式遇到的的问题
curator 官方文档:https://curator.apache.org/index.html

pom.xml 文件添加依赖

<dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>4.3.0</version>
</dependency>
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>4.3.0</version>
</dependency>
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-client</artifactId><version>4.3.0</version>
</dependency>
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;public class CuratorLockTest {public static void main(String[] args) {// 创建分布式锁1InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");// 创建分布式锁2InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");new Thread(new Runnable() {@Overridepublic void run() {try {lock1.acquire();System.out.println("线程1获取到锁");lock1.acquire();System.out.println("线程1获取到锁");Thread.sleep(5 * 1000);lock1.release();System.out.println("线程1释放锁");lock1.release();System.out.println("线程1再次释放锁");} catch (Exception e) {e.printStackTrace();}}}).start();new Thread(new Runnable() {@Overridepublic void run() {try {lock2.acquire();System.out.println("线程2获取到锁");lock2.acquire();System.out.println("线程2获取到锁");Thread.sleep(5 * 1000);lock2.release();System.out.println("线程2释放锁");lock2.release();System.out.println("线程2再次释放锁");} catch (Exception e) {e.printStackTrace();}}}).start();}private static CuratorFramework getCuratorFramework() {ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);CuratorFramework client = CuratorFrameworkFactory.builder().connectString("centos102:2181,centos103:2181,centos104:2181").connectionTimeoutMs(2000).sessionTimeoutMs(2000).retryPolicy(policy).build();// 启动客户端client.start();System.out.println("zookeeper 启动成功");return client;}
}

在这里插入图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/462684.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

江协科技STM32学习- P30 FlyMCU串口下载STLink Utility

&#x1f680;write in front&#x1f680; &#x1f50e;大家好&#xff0c;我是黄桃罐头&#xff0c;希望你看完之后&#xff0c;能对你有所帮助&#xff0c;不足请指正&#xff01;共同学习交流 &#x1f381;欢迎各位→点赞&#x1f44d; 收藏⭐️ 留言&#x1f4dd;​…

Java中的日期与时间对象:LocalDate类、LocalTime类、LocalDateTime类、DateTimeFormatter类

在 Java 中&#xff0c;LocalDate、LocalTime 和 LocalDateTime 是 java.time 包中的类&#xff0c;用于表示日期、时间和日期时间。这些类提供了不可变的日期与时间对象&#xff0c;是 Java 8 及以后版本中引入的一部分&#xff0c;用于替代旧的 java.util.Date 和 java.util.…

Java基于微信小程序的美食推荐系统(附源码,文档)

博主介绍&#xff1a;✌程序猿徐师兄、8年大厂程序员经历。全网粉丝15w、csdn博客专家、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java技术领域和毕业项目实战✌ &#x1f345;文末获取源码联系&#x1f345; &#x1f447;&#x1f3fb; 精彩专栏推荐订阅&#x1f447;…

杨辉三角形

大家好&#xff0c;今天给大家分享一下杨辉三角形是如何打印的&#xff0c;首先我们来看看它的原理。 我们先来看结果 1.如果把它看为一个二维数组&#xff08;包括后面的空格&#xff09;&#xff0c;那么它数字的这边是一个直角三角形&#xff0c;它的第一列和对角线都为1&a…

C语言进阶之我与指针的爱恨情仇(1)

一.前言 我们在初阶《指针》初阶C语言-指针-CSDN博客已经讲过了一些基础知识&#xff0c;知道了关于指针的一些概念-> 1.指针就是个变量&#xff0c;用来存放地址&#xff0c;地址唯一标识一块内存空间 2.指针的大小是固定的4/8个字节&#xff08;32位平台/64位平台&#xf…

构建灵活、高效的HTTP/1.1应用:探索h11库

文章目录 构建灵活、高效的HTTP/1.1应用&#xff1a;探索h11库背景这个库是什么&#xff1f;如何安装这个库&#xff1f;库函数使用方法使用场景常见的Bug及解决方案总结 构建灵活、高效的HTTP/1.1应用&#xff1a;探索h11库 背景 在现代网络应用中&#xff0c;HTTP协议是基础…

基于语音信号的说话人识别

基于语音信号的说话人识别 摘 要 语音是人类相互交流和通信最方便快捷的手段。如何高效地实现语音传输存储或通过 语音实现人机交互&#xff0c;是语音信号处理领域中的重要研究课题。语音信号处理涉及数字信号处理、语音学、语言学、生理学、心理学、计算机科学以及模式识别…

车载软件架构 --- 智能汽车软件

我是穿拖鞋的汉子&#xff0c;魔都中坚持长期主义的汽车电子工程师。 老规矩&#xff0c;分享一段喜欢的文字&#xff0c;避免自己成为高知识低文化的工程师&#xff1a; 所有人的看法和评价都是暂时的&#xff0c;只有自己的经历是伴随一生的&#xff0c;几乎所有的担忧和畏惧…

实际案例说明用基于FPGA的原型来测试、验证和确认IP——如何做到鱼与熊掌兼得?

作者&#xff1a;Philipp Jacobsohn&#xff0c;SmartDV首席应用工程师 Sunil Kumar&#xff0c;SmartDV FPGA设计总监 本系列文章从数字芯片设计项目技术总监的角度出发&#xff0c;介绍了如何将芯片的产品定义与设计和验证规划进行结合&#xff0c;详细讲述了在FPGA上使用硅…

UiPath调用Python脚本的完整示例

一、主要步骤&#xff1a; 1、创建Python脚本 2、安装UiPath.Python.Activities库 3、使用方法&#xff1a; a、添加python作用域 b、加载python脚本 c、调用python方法 d、获取python对象 e、显示Python结果的消息对话框 二、详细步骤 1、安装UiPath.Python.Activities库 …

【简易进度条的实现】

独夜无伴守灯下&#xff0c;清风对面吹............................................................................................. 文章目录 前言 一、【行缓冲区的引入】 1、【问题提出】 2、【\r和\n】 3、【简易倒计时程序】 二、【简易进度条的实现】 process_bar.…

【已解决】cra 配置路径别名 @ 后,出现 ts 报错:找不到模块“@/App”或其相应的类型声明。ts(2307)

cra 配置路径别名 后&#xff0c;出现 ts 报错&#xff1a;找不到模块“/App”或其相应的类型声明。ts(2307) 然后可以在 tsconfig.json 中配置 baseUrl 和 paths &#xff1a; {"compilerOptions": {"target": "es5","lib": [&quo…

es拼音分词器(仅供自己参考)

github地址&#xff1a;https://github.com/infinilabs/analysis-pinyin&#xff08;各种版本&#xff0c;对接es版本&#xff09; 拼音分词器存在的问题&#xff1a; 1、是直接将每个字的拼音返回和一段话的拼音首字母返回&#xff0c;不能很好的分词。 2、不会保留中文&am…

为什么大家都在学数字孪生呢?

随着物联网&#xff0c;大数据、人工智能等技术的发展&#xff0c;新一代信息技术与制造业正在深度融合&#xff0c;人们与物理世界的交互方式正在发生转折性的变化。数字化转型正在成为企业的重要战略&#xff0c;而数字孪生则成为全新的焦点。 当下&#xff0c;在数字技术和…

【英特尔IA-32架构软件开发者开发手册第3卷:系统编程指南】2001年版翻译,2-11

文件下载与邀请翻译者 学习英特尔开发手册&#xff0c;最好手里这个手册文件。原版是PDF文件。点击下方链接了解下载方法。 讲解下载英特尔开发手册的文章 翻译英特尔开发手册&#xff0c;会是一件耗时费力的工作。如果有愿意和我一起来做这件事的&#xff0c;那么&#xff…

LLM Observability: Azure OpenAI (一)

作者&#xff1a;来自 Elastic Vinay Chandrasekhar•Andres Rodriguez 我们很高兴地宣布 Azure OpenAI 集成现已全面上市&#xff0c;它提供了对 Azure OpenAI 服务性能和使用的全面可观察性&#xff01;另请参阅本博客的第 2 部分 虽然我们已经提供了对 LLM 环境的可视性一段…

HTML 基础标签——表格标签<table>

文章目录 1. `<table>` 标签:定义表格2. `<tr>` 标签:定义表格行3. `<th>` 标签:定义表头单元格4. `<td>` 标签:定义表格单元格5. `<caption>` 标签:为表格添加标题6. `<thead>` 标签:定义表格头部7. `<tbody>` 标签:定义表格…

第7章 内容共享

第 7 章 内容共享 bilibili学习地址 github代码地址 本章介绍Android不同应用之间共享内容的具体方式&#xff0c;主要包括&#xff1a;如何利用内容组件在应用之间共享数据&#xff0c;如何使用内容组件获取系统的通讯信息&#xff0c;如何借助文件提供器在应用之间共享文件…

基于 Python 的 Django 框架开发的电影推荐系统

项目简介&#xff1a;本项目是基于 Python 的 Django 框架开发的电影推荐系统&#xff0c;主要功能包括&#xff1a; 电影信息爬取&#xff1a;获取并更新电影数据。数据展示&#xff1a;提供电影数据的列表展示。推荐系统&#xff1a;基于协同过滤算法实现个性化推荐。用户系…

【高等数学】3-2多元函数积分学

1. 二重积分 可以想象你有一块不规则的平面薄板,它在一个平面区域上。二重积分就是用来求这个薄板的质量(假设薄板的面密度函数是)。 把区域划分成许多非常小的小方块(类似于把一块地划分成很多小格子),在每个小方块上,密度近似看成是一个常数,然后把每个小方块的质量加…