Zookeeper-JavaApI操作

JavaApI操作

  • JavaApI操作
    • 1) Curator 介绍
    • 2) Curator API 常用操作
      • a) 建立连接与CRUD基本操作
      • b) Watch事件监听
      • c) 分布式锁
        • c.1) 介绍
        • c.2) Zookeeper分布式锁原理
        • c.3) 案例：模拟12306售票

JavaApI操作

1) Curator 介绍

Curator 是 Apache ZooKeeper 的Java客户端库。

常见的ZooKeeper Java API ：

  • 原生Java API
  • ZkClient
  • Curator

Curator 项目的目标是简化 ZooKeeper 客户端的使用。

Curator 最初是 Netfix 研发的,后来捐献了 Apache 基金会,目前是 Apache 的顶级项目。

官网：http://curator.apache.org/

2) Curator API 常用操作

a) 建立连接与CRUD基本操作

Curator API 常用操作:

  • 建立连接
    • 第一种方法：CuratorFrameworkFactory.newClient
    • 第二种方法：CuratorFrameworkFactory.builder() 推荐
/**
* 建立连接
*/
@Test
public void testConnect() {/*** connectString – 连接字符串 zk server 地址和端口 "192.168.200.130:2181"* sessionTimeoutMs – 会话超时时间 单位ms* connectionTimeoutMs – 连接超时时间 单位ms* retryPolicy – 重试策略*/RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);// 1.第一种方式CuratorFramework client1 = CuratorFrameworkFactory.newClient("192.168.200.130:2181", 60 * 1000, 15 * 1000, retryPolicy);client1.start();// 2.第二种方式CuratorFramework client2 = CuratorFrameworkFactory.builder().connectString("192.168.200.130:2181").sessionTimeoutMs(60 * 1000).connectionTimeoutMs(15 * 1000).retryPolicy(retryPolicy).namespace("dcy") // 名称空间.build();client2.start();
}
  • 添加节点
    • 如果创建节点，没有指定数据，则默认将当前客户端的ip作为数据存储
    • 默认类型：持久化
    • creatingParentsIfNeeded 如果父节点不存在，则创建父节点
/**
* 创建节点: create 持久 临时 顺序 数据
*/
@Test
public void testCreate1() throws Exception {// 1.基本创建// 如果创建节点，没有指定数据，则默认将当前客户端的ip作为数据存储String rs = client2.create().forPath("/app1");System.out.println("rs = " + rs);
}@Test
public void testCreate2() throws Exception {// 2.创建节点 带有数据String rs = client2.create().forPath("/app2", "hehe".getBytes());System.out.println("rs = " + rs);
}@Test
public void testCreate3() throws Exception {// 3.设置节点的类型// 默认类型：持久化String rs = client2.create().withMode(CreateMode.EPHEMERAL).forPath("/app3");System.out.println("rs = " + rs);
}@Test
public void testCreate4() throws Exception {// 4.创建多级节点// creatingParentsIfNeeded 如果父节点不存在，则创建父节点String rs = client2.create().creatingParentsIfNeeded().forPath("/app4/p1");System.out.println("rs = " + rs);
}
  • 删除节点
    • 1.删除单个节点 delete().forPath()
    • 2.删除带有子节点的节点 delete().deletingChildrenIfNeeded().forPath()
    • 3.必须成功的删除(为了防止网络抖动，本质是重试) delete().guaranteed().forPath()
    • 4.回调 inBackground
/**
* 删除节点: delete, deleteall
*/
@Test
public void testDelete1() throws Exception {// 1.删除单个节点client2.delete().forPath("/app1");
}@Test
public void testDelete2() throws Exception {// 2.删除带有子节点的节点client2.delete().deletingChildrenIfNeeded().forPath("/app4");
}@Test
public void testDelete3() throws Exception {// 3.必须成功的删除 (可能会因为网络抖动等原因，操作超时)client2.delete().guaranteed().forPath("/app2");
}@Test
public void testDelete4() throws Exception {// 4.回调client2.delete().guaranteed().inBackground(new BackgroundCallback() {@Overridepublic void processResult(CuratorFramework client, CuratorEvent event) throws Exception {System.out.println("我被删除了~");System.out.println(event);}}).forPath("/app1");
}
  • 修改节点
    • 1.修改数据 setData().forPath()
    • 2.根据版本修改 setData().withVersion().forPath()
      • version 是通过查询出来的，目的是为了让其他客户端和线程不干扰我
/**
* 修改数据
*/
@Test
public void testSet() throws Exception {client2.setData().forPath("/app1", "dcy".getBytes());
}@Test
public void testSetForVersion() throws Exception {Stat stat = new Stat();client2.getData().storingStatIn(stat).forPath("/app1");int version = stat.getVersion(); // 查询出来client2.setData().withVersion(version).forPath("/app1", "haha".getBytes());
}
  • 查询节点
    • 1.查询数据：get: getData().forPath()
    • 2.查询子节点：ls: getChildren().forPath()
    • 3.查询节点状态信息：ls -s: getData().storingStatIn(状态对象).forPath()
/**
* 查询节点：
*  1.查询数据：get: getData().forPath()
*  2.查询子节点：ls: getChildren().forPath()
*  3.查询节点状态信息：ls -s: getData().storingStatIn(状态对象).forPath()
*/
@Test
public void testGet1() throws Exception {// 1.查询数据：getbyte[] bytes = client2.getData().forPath("/app1");System.out.println(new String(bytes));
}@Test
public void testGet2() throws Exception {// 2.查询子节点：lsList<String> path = client2.getChildren().forPath("/");System.out.println(path);
}@Test
public void testGet3() throws Exception {// 3.查询节点状态信息：ls -sStat status = new Stat();client2.getData().storingStatIn(status).forPath("/app1");System.out.println("status = " + status);
}

b) Watch事件监听

ZooKeeper 允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，ZooKeeper 服务端会将事件通知到感兴趣的客户端上去，该机制是 ZooKeeper 实现分布式协调服务的重要特性。

ZooKeeper 中引入了Watcher机制来实现了发布/订阅功能能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。

ZooKeeper 原生支持通过注册Watcher来进行事件监听，但是其使用并不是特别方便 需要开发人员自己反复注册Watcher，比较繁琐。

Curator引入了 Cache 来实现对 ZooKeeper 服务端事件的监听。

ZooKeeper提供了三种Watcher：

  • NodeCache : 只是监听某一个特定的节点
  • PathChildrenCache : 监控一个ZNode的子节点.
  • TreeCache : 可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合

1.NodeCache : 监听一个特定的节点

/**
* 演示NodeCache
*/
@Test
public void testNodeCache() throws Exception {// 1.创建NodeCache监听对象final NodeCache nodeCache = new NodeCache(client, "/app1");// 2.注册监听nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {System.out.println("节点变化~~~");// 获取修改节点后的数据byte[] data = nodeCache.getCurrentData().getData();System.out.println(new String(data));}});// 3.开启监听，如果设置true，则开启监听是，加载缓存数据nodeCache.start(true);while (true) {}
}

2.PathChildrenCache：监听某个节点的所有子节点

/**
* 演示PathChildrenCache：监听某个节点的所有子节点
*/
@Test
public void testPathChildrenCache () throws Exception {// 1.创建监听对象PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/app2", true);// 2.绑定监听器pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {System.out.println("子节点变化~~~");System.out.println("event = " + event);// 监听子节点的数据变更，并且拿到变更后的数据// 1.获取类型PathChildrenCacheEvent.Type type = event.getType();// 2.判断类型是否是updateif (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {byte[] data = event.getData().getData();System.out.println(new String(data));}}});// 3.开启监听pathChildrenCache.start();while (true) {}
}

3.TreeCache：监听某个节点自己和所有的子节点

/**
* 演示TreeCache：监听某个节点自己和所有的子节点
*/
@Test
public void testTreeCache () throws Exception {// 1.创建监听器TreeCache treeCache = new TreeCache(client, "/app2");// 2.注册监听treeCache.getListenable().addListener(new TreeCacheListener() {@Overridepublic void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {System.out.println("节点变化了");System.out.println("event = " + event);}});// 3.开启监听treeCache.start();while (true) {}
}

c) 分布式锁

c.1) 介绍

在我们进行单机应用开发，涉及并发同步的时候，我们往往采用synchronized或者Lock的方式来解决多线程间的代码同步问题，这时多线程的运行都是在同一个JVM之下，没有任何问题。

但当我们的应用是分布式集群工作的情况下，属于多JVM下的工作环境，跨JVM之间已经无法通过多线程的锁解决同步问题。

在这里插入图片描述

那么就需要一种更加高级的锁机制，来处理种跨机器的进程之间的数据同步问题——这就是分布式锁。

在这里插入图片描述

c.2) Zookeeper分布式锁原理

核心思想：当客户端要获取锁，则创建节点，使用完锁，则删除节点

  • 1.客户端获取锁时，在lock节点下创建临时顺序节点
  • 2.然后获取lock下面的所有子节点，客户端获取到所有的子节点后。如果发现自己创建的节点顺序最小，那就认为该客户端获取到了锁。使用完锁后，将该节点删除
  • 3.如果发现自己创建的节点并发lock所有子节点中最小的，说明自己还没有获取到锁。此时客户端需要找到比自己小的那个节点，同时对其注册事件监听器，监听删除事件
  • 4.如果发现比自己小的那个节点被删除，则客户端的Watcher会收到相应通知，此时再次判断自己创建的节点 是否是lock子节点中序号最小的。如果是则获取到了锁，如果不是则重复以上步骤继续获取到比自己小的一个节点，并注册监听。
c.3) 案例：模拟12306售票

Curator实现分布式锁API

在Curator中有五种锁方案：

  • InterProcessSemaphoreMutex：分布式排它锁（非可重入锁）
  • InterProcessMutex：分布式可重入排它锁
  • InterProcessReadWriteLock：分布式读写锁
  • InterProcessMultiLock：将多个锁作为单个实体管理的容器
  • InterProcessSemaphoreV2：共享信号量

在这里插入图片描述

// 测试类
public class LockTest {public static void main(String[] args) {Ticket12306 ticket12306 = new Ticket12306();// 创建客户端Thread t1 = new Thread(ticket12306, "携程");Thread t2 = new Thread(ticket12306, "飞猪");t1.start();t2.start();}
}// Ticket12306
public class Ticket12306 implements Runnable{private int tickets = 10; // 数据库的票数private InterProcessMutex lock; // 创建锁public Ticket12306() {RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);CuratorFramework client = CuratorFrameworkFactory.builder().connectString("192.168.200.130:2181").sessionTimeoutMs(60 * 1000).connectionTimeoutMs(15 * 1000).retryPolicy(retryPolicy).build();client.start();this.lock = new InterProcessMutex(client, "/lock");}@Overridepublic void run() {while (true) {try {// 获取锁lock.acquire(3, TimeUnit.SECONDS);if (tickets > 0) {System.out.println(Thread.currentThread().getName() + ":" + tickets);tickets--;}} catch (Exception e) {e.printStackTrace();} finally {// 释放锁try {lock.release();} catch (Exception e) {e.printStackTrace();}}}}
}

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

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

相关文章

通过商品ID获取淘宝天猫商品评论数据，淘宝商品评论接口，淘宝商品评论api接口

通过商品ID获取淘宝天猫商品评论数据，淘宝商品评论接口，淘宝商品评论api接口

淘宝商品评论内容数据接口可以通过以下步骤获取&#xff1a; 登录淘宝开放平台&#xff0c;进入API管理控制台。在API管理控制台中创建一个应用&#xff0c;获取到应用的App Key和Secret Key。构造请求URL&#xff0c;请求URL由App Key和Secret Key拼接而成&#xff0c;请求UR…
阅读更多...
GPU 基础知识整理

GPU 基础知识整理

萌新&#xff1a; 在接触一款硬件时我会&#xff1a;基础硬件结构&#xff0c;线程结构&#xff0c;内存布局&#xff0c;数据吞吐量&#xff0c;等方面进行学习 首先GPU的特点: 并行性能&#xff1a;GPU 是专门设计用于并行计算的硬件&#xff0c;通常具有大量的处理单元&am…
阅读更多...
【已编译资料】基于正点原子alpha开发板的第三篇系统移植

【已编译资料】基于正点原子alpha开发板的第三篇系统移植

系统移植的三大步骤如下&#xff1a; 系统uboot移植系统linux移植系统rootfs制作 一言难尽&#xff0c;踩了不少坑&#xff0c;当时只是想学习驱动开发&#xff0c;发现必须要将第三篇系统移植弄好才可以学习后面驱动&#xff0c;现将移植好的文件分享出来&#xff1a; 仓库&…
阅读更多...
关于vue3启动的一些错误总结

关于vue3启动的一些错误总结

一、成功解决 npm ERR! ERESOLVE could not resolve 解决办法&#xff1a; npm i --legacy-peer-deps –legacy-peer-deps 作用&#xff1a; 在NPM v7中&#xff0c;现在默认安装peerDependencies。 在很多情况下&#xff0c;这会导致版本冲突&#xff0c;从而中断安装过程。 …
阅读更多...
如何通过Express和React处理SSE

如何通过Express和React处理SSE

本文作者为360奇舞团前端开发工程师 最近AIGC技术的大热&#xff0c;市面上也出现了许多类似生产的AI工具&#xff0c;其中有一大特色就是对话的输出结果是类似真人的打字效果出现&#xff0c;要呈现出这种效果&#xff0c;最主要的就是要利用SSE技术&#xff08;Server-Sent E…
阅读更多...
本地vscode安装GPU版本PyTorch

本地vscode安装GPU版本PyTorch

操作系统 windows, IDE环境vscode&#xff0c;本地GPU 可以新建一个jupyter文件&#xff0c;运行一些测试代码 确保装好显卡驱动 在底下调出终端窗口&#xff0c;默认是power shell&#xff0c;我喜欢用cmd窗口 激活自己的虚拟环境&#xff0c;输入命令 nvidia-smi 确保自己…
阅读更多...
Generative AI 新世界 | 文生图领域动手实践：预训练模型的微调

Generative AI 新世界 | 文生图领域动手实践：预训练模型的微调

在上期文章&#xff0c;我们探讨了预训练模型的部署和推理&#xff0c;包括运行环境准备、角色权限配置、支持的主要推理参数、图像的压缩输出、提示工程 (Prompt Engineering)、反向提示 (Negative Prompting) 等内容。 亚马逊云科技开发者社区为开发者们提供全球的开发技术资…
阅读更多...
Lumen/Laravel - 数据库读写分离原理 - 探究

Lumen/Laravel - 数据库读写分离原理 - 探究

1.应用场景 主要用于学习与探究Lumen/Laravel的数据库读写分离原理。 2.学习/操作 1.文档阅读 chatgpt & 其他资料 数据库入门 | 数据库操作 | Laravel 8 中文文档 入门篇&#xff08;一&#xff09;&#xff1a;数据库连接配置和读写分离 | 数据库与 Eloquent 模型 | La…
阅读更多...
分布式数据库HBase（林子雨慕课课程）

分布式数据库HBase（林子雨慕课课程）

文章目录 4. 分布式数据库HBase4.1 HBase简介4.2 HBase数据模型4.3 HBase的实现原理4.4 HBase运行机制4.5 HBase的应用方案4.6 HBase安装和编程实战 4. 分布式数据库HBase 4.1 HBase简介 HBase是BigTable的开源实现 对于网页搜索主要分为两个阶段 1.建立整个网页索引&#xf…
阅读更多...
【JVM】初步认识Java虚拟机

【JVM】初步认识Java虚拟机

&#x1f40c;个人主页&#xff1a; &#x1f40c; 叶落闲庭 &#x1f4a8;我的专栏&#xff1a;&#x1f4a8; c语言 数据结构 javaEE 操作系统 Redis 石可破也&#xff0c;而不可夺坚&#xff1b;丹可磨也&#xff0c;而不可夺赤。 JVM 一、初识JVM1.1 什么是JVM1.2 JVM的功能…
阅读更多...
苹果安卓网页的H5封装成App的应用和原生开发的应用有什么不一样？

苹果安卓网页的H5封装成App的应用和原生开发的应用有什么不一样？

老哥在么&#xff1f;H5封装的app和原生开发的app有什么不一样&#xff1f;&#xff0c;不懂就要问&#xff0c;我能理解哈&#xff0c;虽然这个问题有点小白&#xff0c;但是我还是得认真回答&#xff0c;以防止我回答的不是很好&#xff0c;所以我科技了一下&#xff0c;所以…
阅读更多...
input时间控件选择时禁用某个日期之前或之后

input时间控件选择时禁用某个日期之前或之后

【版权所有&#xff0c;文章允许转载&#xff0c;但须以链接方式注明源地址&#xff0c;否则追究法律责任】【创作不易&#xff0c;点个赞就是对我最大的支持】 前言 仅作为学习笔记&#xff0c;供大家参考 总结的不错的话&#xff0c;记得点赞收藏关注哦&#xff01; 目录 …
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023