完整域名展示网站源码,网络推广公司哪家好,网站换网址了怎么找,小程序有哪些开发平台项目介绍
ZooKeeper本质上是一个分布式的小文件存储系统。原本是Apache Hadoop的一个组件#xff0c;现在被拆分为一个Hadoop的独立子项目。
Zookeeper 作为一个分布式的服务框架#xff0c;主要用来解决分布式集群中应用系统的一致性问题#xff0c;它能提供基于类似于文…项目介绍
ZooKeeper本质上是一个分布式的小文件存储系统。原本是Apache Hadoop的一个组件现在被拆分为一个Hadoop的独立子项目。
Zookeeper 作为一个分布式的服务框架主要用来解决分布式集群中应用系统的一致性问题它能提供基于类似于文件系统的目录节点树方式的数据存储但是 Zookeeper 并不是用来专门存储数据的它的作用主要是用来维护和监控你存储的数据的状态变化。通过监控这些数据状态的变化从而可以达到基于数据的集群管理。
Zookeeper 在Windows安装和使用可参考http://www.cnblogs.com/shanyou/p/3221990.html
ZookeeperClient是在https://github.com/shayhatsor/zookeeper基础上的再次封装使开发者更方便使用ZooKeeper相关的功能。
ZookeeperClient实现了断线重连会话过期重连永久监听子节点数据变化的监听。并且加入了常用功能例如分布式锁Leader选举分布式队列
项目地址https://github.com/milanyangbo/ZooKeeper.Net
支持的平台
.NET 4及以上目前尚不支持.NET Core, .NET Core推荐用另外一个项目 支持断线重连、永久watcher、递归操作并且能跨平台.NET Core的ZooKeeper异步客户端 使用说明
下面列一下常用的使用方法不仅限于此哦
一、创建ZKClient对象
创建ZKClient对象 有两种方式可以方便的创建ZKClient对象 使用构造函数创建 string address localhost:2181; ZKClient zkClient1 new ZKClient(address); ZKClient zkClient2 new ZKClient(address, TimeSpan.FromMilliseconds(10000)); ZKClient zkClient3 new ZKClient(address, TimeSpan.FromMilliseconds(10000), TimeSpan.FromMilliseconds(10000)); ZKClient zkClient4 new ZKClient(address, TimeSpan.FromMilliseconds(30000), TimeSpan.FromMilliseconds(10000), new SerializableSerializer()); ZKClient zkClient5 new ZKClient(address, TimeSpan.FromMilliseconds(30000), TimeSpan.FromMilliseconds(10000), new SerializableSerializer(), TimeSpan.FromMilliseconds(60000)); 使用辅助类创建
string address localhost:2181; ZKClient zkClient ZKClientBuilder.NewZKClient(address) .SessionTimeout(30000)//可选 .Serializer(new SerializableSerializer())//可选 .RetryTimeout(60000)//可选 .ConnectionTimeout(10000)//可选 .Build(); //创建实例 二、节点的新增、更新、删除和获取
新增节点
常规新增节点父节点不存在会抛出异常 await zkClient.CreateAsync(/test1, 123, CreateMode.EPHEMERAL); await zkClient.CreateAsync(/test1-1, 123, CreateMode.EPHEMERAL_SEQUENTIAL); await zkClient.CreateAsync(/test1-2, 123, CreateMode.PERSISTENT); await zkClient.CreateAsync(/test1-3, 123, CreateMode.PERSISTENT_SEQUENTIAL);
递归新增节点新增节点及其父节点如果父节点不存在会被一并创建。对于PERSISTENT类型的节点递归创建父节点和子节点都创建为PERSISTENT。对于EPHEMERAL类型的节点递归创建父节点都是PERSISTENT类型而最后一级节点才是EPHEMERAL类型。因为EPHEMERAL不能拥有子节点注意第二个参数为节点的值指的的最后一级节点的值。 string path /test8/1/2/3; //递归创建节点及父节点 await zkClient.CreateRecursiveAsync(path, abc, CreateMode.PERSISTENT); await zkClient.CreateRecursiveAsync(path, 123, ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
特殊的EPHEMERAL类型节点特殊类型的EPHEMERAL节点该节点在会话失效被删除后重新连接会被自动创建。 更新节点数据 string path /test; await zkClient.SetDataAsync(path, 456); //带期望版本号的更新如果真实的版本号与期望版本号不一致会更新失败抛出异常await zkClient.SetDataAsync(path, 123, 2); 删除节点
常规删除 bool flag await zkClient.DeleteAsync(/test);//删除任意版本bool flag await zkClient.DeleteAsync(/test,1);//删除指定版本
递归删除删除节点及子节点 string path /test; await zkClient.DeleteRecursiveAsync(path);//如果/test下有多个子节点会被一并删除 获取节点数据 string path /test; await zkClient.GetDataAsyncstring(path); //如果节点不存在抛出异常await zkClient.GetDataAsyncstring(path, true); //如果节点不存在返回nullStat stat (await zkClient.GetZKDataAsyncstring(path)).stat; //获得数据以及stat信息 等待节点创建 string path /test; //等待直到超时或者节点创建成功。await zkClient.WaitUntilExistsAsync(path, TimeSpan.FromMilliseconds(5000)); 三、权限管理
ZooKeeper的权限管理亦即ACL控制功能通过Server、Client两端协调完成Server端一个ZooKeeper的节点znode存储两部分内容数据和状态状态中包含ACL信息。创建一个znode会产生一个ACL列表列表中每个ACL包括 验证模式(scheme)具体内容(Id)当scheme“digest”时Id为用户名密码例如“rootJ0sTy9BCUKubtK1y8pkbL7qoxSw”权限(perms) ZooKeeper提供了如下几种验证模式scheme digestClient端由用户名和密码验证譬如user:passworddigest的密码生成方式是Sha1摘要的base64形式auth不使用任何id代表任何已确认用户。ipClient端由IP地址验证譬如172.2.0.0/24world固定用户为anyone为所有Client端开放权限super在这种scheme情况下对应的id拥有超级权限可以做任何事情(cdrwa 注意的是exists操作和getAcl操作并不受ACL许可控制因此任何客户端可以查询节点的状态和节点的ACL。节点的权限perms主要有以下几种 Create 允许对子节点Create操作Read 允许对本节点GetChildren和GetData操作Write 允许对本节点SetData操作Delete 允许对子节点Delete操作Admin 允许对本节点setAcl操作Znode ACL权限用一个int型数字perms表示perms的5个二进制位分别表示setacl、delete、create、write、read。比如0x1fadcwr0x1----r0x15a-c-r。 四、监听相关
注意对于断开连接时间过长造成的会话过期由于服务器端在会话过期后会删除客户端设置的监听。
即便客户端在会话过期后自动连接成功但是在会话过期到会话重建这段时间客户端监听的节点仍可能发生了改变
而具体哪些变了或是没变客户端是无法感知到的。
为了避免丢掉任何数据改变的事件所有的监听器的都有一个回调方法SessionExpiredHandler用来处理会话过期这种特殊情况。 SessionExpiredHandler async (path) { await Task.Run(() { Console.WriteLine(path);});}; 订阅节点的信息改变创建节点删除节点添加子节点 IZKChildListener childListener new ZKChildListener(); //子节点内容变化childListener.ChildChangeHandler async (parentPath, currentChilds) { await Task.Run(() { Console.WriteLine(parentPath); Console.WriteLine(string.Join(., currentChilds));});}; //子节点数量变化childListener.ChildCountChangedHandler async (parentPath, currentChilds) { await Task.Run(() { Console.WriteLine(parentPath); Console.WriteLine(string.Join(., currentChilds));});}; ///testUserNode 监听的节点可以是现在存在的也可以是不存在的 zkClient.SubscribeChildChanges(/testUserNode3, childListener); 订阅节点的数据内容的变化 IZKDataListener dataListener new ZKDataListener(); // 节点创建和节点内容变化dataListener.DataCreatedOrChangeHandler async (dataPath, data) { await Task.Run(() { Console.WriteLine(dataPath : Convert.ToString(data));});}; // 节点删除dataListener.DataDeletedHandler async (dataPath) { await Task.Run(() { Console.WriteLine(dataPath);});}; // 节点创建dataListener.DataCreatedHandler async (dataPath, data) { await Task.Run(() { Console.WriteLine(dataPath : Convert.ToString(data));});}; // 节点内容变化dataListener.DataChangeHandler async (dataPath, data) { await Task.Run(() { Console.WriteLine(dataPath);});}; zkClient.SubscribeDataChanges(/testUserNode, dataListener); 客户端状态监听 IZKStateListener stateListener new ZKStateListener(); //状态改变stateListener.StateChangedHandler async (state) { await Task.Run(() { Console.WriteLine(state.ToString());});}; //会话失效stateListener.SessionExpiredHandler async (path) { await Task.Run(() { Console.WriteLine(path);});}; //创建会话stateListener.NewSessionHandler async () { await Task.Run(() {});}; //会话失败stateListener.SessionEstablishmentErrorHandler async (ex) { await Task.Run(() { Console.WriteLine(ex.Message);});}; zkClient.SubscribeStateChanges(stateListener); 五、扩展功能 分布式锁 using (var zkClient new ZKClient(TestUtil.zkServers)){ await zkClient.CreateRecursiveAsync(/zk/lock, null, CreateMode.Persistent); //创建分布式锁 非线程安全类每个线程请创建单独实例。var _lock new ZKDistributedLock(zkClient, /zk/lock); await _lock.LockAsync(); //获得锁//do sometingawait _lock.UnLockAsync();//释放锁} Leader选举 using (var zkClient new ZKClient(TestUtil.zkServers)){ await zkClient.CreateRecursiveAsync(/zk/leader, null, CreateMode.Persistent); var listener new ZKLeaderSelectorListener(); listener.takeLeadership async (client, selector) { Console.WriteLine(I am the leader- await selector.GetLeaderAsync()); selector.Close();}; var selector new ZKLeaderSelector(id, true, zkClient, /zk/leader, listener); //启动并参与Leader选举selector.Start(); //获得当前主服务的IDawait selector.GetLeaderAsync(); //如果要退出Leader选举selector.Close();
} 分布式队列
using (var zkClient new ZKClient(TestUtil.zkServers)){ await zkClient.CreateRecursiveAsync(/zk/queue, null, CreateMode.PERSISTENT); var queue new ZKDistributedQueuelong(new ZKClient(TestUtil.zkServers), /zk/queue) await queue.OfferAsync(123);//放入元素var value await queue.PollAsync();//删除并获取顶部元素var value await queue.PeekAsync(); //获取顶部元素不会删除 }
