C#如何连接使用Zookeeper C#如何连接使用Zookeeper

Zookeeper作为分布式的服务框架，虽然是java写的，但是强大的C#也可以连接使用。
C#要连接使用Zookeeper，需要借助第三方插件，而现在主要有两个插件可供使用，分别是ZooKeeperNetEx和Zookeeper.Net
Zookeeper.Net好像是是Apache官方提供的，但是5年没更新了，也就是说他依赖于.net framework，因此无法在.net core项目中使用
ZooKeeperNetEx是从java改过来的，因此里面的一些习惯是java风格的，但是好像有人在提供更新维护，支持最新的Zookeeper特性，而且摆脱了对.net framework的依赖，所以个人推荐使用ZooKeeperNetEx做开发，本文也已介绍ZooKeeperNetEx为主
新建一个控制台项目，在nuget中搜索ZooKeeperNetEx，并安装最新版

文章图片

在Program的Main方法：

using org.apache.zookeeper; using org.apache.zookeeper.data; using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; namespace AspNetCore.ZookeeperConsole{class Program{static void Main(string[] args){//Zookeeper连接字符串，采用host:port格式，多个地址之间使用逗号（,）隔开string connectionString = "192.168.209.133:2181,192.168.209.133:2181,192.168.209.133:2181"; //会话超时时间,单位毫秒int sessionTimeOut = 10000; //异步监听var watcher = new MyWatcher("ConnectWatcher"); //连接ZooKeeper zooKeeper = new ZooKeeper(connectionString, sessionTimeOut, watcher); Thread.Sleep(1000); //停一秒，等待连接完成while (zooKeeper.getState() == ZooKeeper.States.CONNECTING){Console.WriteLine("等待连接完成..."); Thread.Sleep(1000); }var state = zooKeeper.getState(); if (state != ZooKeeper.States.CONNECTED && state != ZooKeeper.States.CONNECTEDREADONLY){Console.WriteLine("连接失败：" + state); Console.ReadKey(); return; }//创建znode节点{var data = https://www.it610.com/article/Encoding.UTF8.GetBytes("hello world"); List acl = ZooDefs.Ids.OPEN_ACL_UNSAFE; //创建节点时的acl权限，也可以使用下面的自定义权限//List acl = new List() {//new ACL((int)ZooDefs.Perms.READ, new Id("ip", "127.0.0.1")),//new ACL((int)(ZooDefs.Perms.READ | ZooDefs.Perms.WRITE), new Id("auth", "id:pass"))//}; CreateMode createMode = CreateMode.PERSISTENT; zooKeeper.createAsync("/mynode", data, acl, createMode).Wait(); Console.WriteLine("完成创建节点"); }//节点是否存在{var exists = zooKeeper.existsAsync("/mynode", new MyWatcher("ExistsWatcher")).GetAwaiter().GetResult(); Console.WriteLine("节点是否存在：" + exists); }//获取节点数据{var dataResult = zooKeeper.getDataAsync("/mynode", new MyWatcher("GetWatcher")).GetAwaiter().GetResult(); var value = https://www.it610.com/article/Encoding.UTF8.GetString(dataResult.Data); Console.WriteLine("完成读取节点：" + value); }//设置节点数据{var data = https://www.it610.com/article/Encoding.UTF8.GetBytes("hello world again"); zooKeeper.setDataAsync("/mynode", data); Console.WriteLine("设置节点数据"); }//重新获取节点数据{var dataResult = zooKeeper.getDataAsync("/mynode", new MyWatcher("GetWatcher")).GetAwaiter().GetResult(); var value = https://www.it610.com/article/Encoding.UTF8.GetString(dataResult.Data); Console.WriteLine("重新获取节点数据：" + value); }//移除节点{zooKeeper.deleteAsync("/mynode").Wait(); Console.WriteLine("移除节点"); }Console.WriteLine("完成"); Console.ReadKey(); }}class MyWatcher : Watcher{public string Name { get; private set; }public MyWatcher(string name){this.Name = name; }public override Task process(WatchedEvent @event){var path = @event.getPath(); var state = @event.getState(); Console.WriteLine($"{Name} recieve: Path-{path}State-{@event.getState()}Type-{@event.get_Type()}"); return Task.FromResult(0); }}}

运行后显示结果：

文章图片

这个简单的例子是使用ZooKeeperNetEx操作的简单例子，下面具体介绍
ZooKeeperNetEx连接Zookeeper只需要实例化ZooKeeper对象即可

//Zookeeper连接字符串，采用host:port格式，多个地址之间使用逗号（,）隔开string connectionString = "192.168.209.133:2181,192.168.209.133:2181,192.168.209.133:2181"; //会话超时时间,单位毫秒int sessionTimeOut = 10000; //异步监听var watcher = new MyWatcher("ConnectWatcher"); //连接ZooKeeper zooKeeper = new ZooKeeper(connectionString, sessionTimeOut, watcher);

实例化过程中至少需要三个参数
连接字符串(connectstring)：host:port形式，多个地址之间使用英文逗号隔开
会话超时时间(sessionTimeout)：当会话中，Zookeeper超过此时间未响应，则表示会话超时
监听器(watcher)：这个连接过程中可以注册一个监听器，当连接过程中出现状态改变时，会通知到监听器
ZooKeeper对象实例化过程中会异步的去连接Zookeeper，所以例子中才有一个while循环来判断状态

Thread.Sleep(1000); //停一秒，等待连接完成while (zooKeeper.getState() == ZooKeeper.States.CONNECTING){Console.WriteLine("等待连接完成..."); Thread.Sleep(1000); }

而Zookeeper的连接状态有6种：

//ZooKeeper.States的枚举CONNECTING = 0,//连接中CONNECTED = 1,//已连接CONNECTEDREADONLY = 2,//已连接，但是只能只读访问CLOSED = 3,//已关闭连接AUTH_FAILED = 4,//认证失败NOT_CONNECTED = 5//未连接

当应用连接到Zookeeper时，一般都是读取数据，所以主需要只读连接就可以满足的，不过具体还是要看需求。
当在指定的会话时间内未成功连接时，则会导致连接超时，因为这个过程是异步的，所以需要一个监听器来接收。
监听器其实是org.apache.zookeeper.Watcher的一个子类，这个需要开发者去继承实现它的process方法，比如上面的例子中我们就简单的实现

class MyWatcher : Watcher{public string Name { get; private set; }public MyWatcher(string name){this.Name = name; }public override Task process(WatchedEvent @event){var path = @event.getPath(); var state = @event.getState(); Console.WriteLine($"{Name} recieve: Path-{path}State-{@event.getState()}Type-{@event.get_Type()}"); return Task.FromResult(0); }}

这里仅仅只是简单的输出节点路径、监听事件响应状态和监听事件类型

//监听事件响应状态，Watcher.Event.KeeperState的枚举Expired = -112,//连接超时Disconnected = 0,//连接断开SyncConnected = 3,//已同步连接AuthFailed = 4,//认证失败ConnectedReadOnly = 5//只读连接//监听事件类型，Watcher.Event.EventType的枚举None = -1,//非节点操作事件NodeCreated = 1,//创建节点事件NodeDeleted = 2,//删除节点事件NodeDataChanged = 3,//节点数据改变NodeChildrenChanged = 4//子节点发生改变

为什么要有监听器？监听器就类似一个回调，当发生某个事件时，我们的应用可能需要进行相应的处理，如当连接断开时，由于监听器的存在，我们可以让我们的应用程序重新与Zookeeper建立连接。
ZooKeeperNetEx创建znode节点使用的是createAsync异步方法，传入4个参数，分别是
节点路径(path):：创建的节点路径
节点数据(data)：节点数据，它是一个字节数组，可以通过编码将字符串转化为字符数组
ACL权限(acl)：ACL权限，可以使用已定义好的，也可以使用自定义，如：

//已经定义好的，ZooDefs.Ids的枚举OPEN_ACL_UNSAFE：完全开放CREATOR_ALL_ACL：创建该znode的连接拥有所有权限READ_ACL_UNSAFE：所有的客户端都可读

自定义方式如：

List acl = new List() {new ACL((int)ZooDefs.Perms.READ, new Id("ip", "127.0.0.1")),new ACL((int)(ZooDefs.Perms.READ | ZooDefs.Perms.WRITE), new Id("auth", "id:pass"))};

节点类型(createMode)：节点类型有4种，分别是CreateMode类的4个静态字段

PERSISTENT：持久化节点PERSISTENT_SEQUENTIAL：持久化有序节点EPHEMERAL：临时节点（连接断开自动删除）EPHEMERAL_SEQUENTIAL：临时有序节点（连接断开自动删除）

createAsync异步方法会返回实际创建的znode路径，貌似没什么用（在创建顺序节点时会用到，当一个新的znode被创建为一个顺序节点时，ZooKeeper通过将10位的序列号附加到原始名称来设置znode的路径）
上面这个是ZooKeeperNetEx创建znode节点的方法，而对znode的其他操作的参数就很简单了，这里就不在重述，需要具体操作才能理解，一个简单的介绍如下：

//删除znode节点public Task deleteAsync(string path, int version = -1); //指定的znode节点是否存在public Task existsAsync(string path, Watcher watcher); public Task existsAsync(string path, bool watch = false); //获取znode节点数据public Task getDataAsync(string path, bool watch = false); public Task getDataAsync(string path, Watcher watcher); //设置指定znode节点的数据public Task setDataAsync(string path, byte[] data, int version = -1); //获取指定znode节点的子节点，注意，监听器是注册给当前节点的，而非子节点public Task getChildrenAsync(string path, Watcher watcher); public Task getChildrenAsync(string path, bool watch = false);

可以比较一下上一节介绍的zkCli对znode节点的操作就很容易理解了。
另外，需要注意的是，existsAsync方法、getDataAsync方法和getChildrenAsync方法可以在指定的znode注册一个监听器，setDataAsync方法却没有这个注册功能，这个是因为Zookeeper注册的监听器只会响应一次，当需要再次响应时，需要重新注册，这时就可以调用existsAsync方法或者getDataAsync方法或者getChildrenAsync方法进行重新注册了！
上一节说到ACL权限不仅可以在创建是给予，在创建后也可以修改，ZookeeperNetEx操作znode的ACL权限使用的方法如下：

//获取ACL权限public Task getACLAsync(string path); //设置ACL权限public Task setACLAsync(string path, List acl, int aclVersion = -1);

说到ACL，自然就会认证存在，ZookeeperNetEx添加认证使用的是addAuthInfo方法

public void addAuthInfo(string scheme, byte[] auth);

其中scheme就是我们上一节介绍的那几种：

world：默认模式，所有客户端都拥有指定的权限。world下只有一个id选项，就是anyone，通常组合写法为world:anyone:[permissons]；auth：只有经过认证的用户才拥有指定的权限。通常组合写法为auth:user:password:[permissons]，使用这种模式时，你需要先进行登录，之后采用auth模式设置权限时，user和password都将使用登录的用户名和密码；比如：digest：只有经过认证的用户才拥有指定的权限。通常组合写法为digest:user:BASE64(SHA1(password)):[permissons]，这种形式下的密码必须通过SHA1和BASE64进行双重加密；ip：限制只有特定IP的客户端才拥有指定的权限。通常组成写法为ip:182.168.0.168:[permissions]；super：代表超级管理员，拥有所有的权限，需要修改Zookeeper启动脚本进行配置。

auth是认证数据，如果没有则可以是空的字节数组，如：

//world模式认证zk.addAuthInfo("world",new byte[0]); //auth模式认证byte[] auth=Encoding.UTF8.GetBytes("id:pass")zk.addAuthInfo("auth",new byte[0]); //digest模式认证byte[] auth=Encoding.UTF8.GetBytes("加密后的字符串")zk.addAuthInfo("digest",new byte[0]);

ZookeeperNetEx关闭会话使用的是closeAsync方法，调用这个方法之后，当前连接对象ZooKeeper就不能再访问了

public Task closeAsync();

其他常用方法就不介绍了，一般时候基本上也用不上。
简单封装
真实项目中，我们连接Zookeeper多数只是为了创建znode节点，读取数据等等操作，一般不会去设置ACL等权限，甚至连认证都可能不会用到，为了更好使用ZookeeperNetEx，我做了一层简单的封装，用以满足常见的CRUD操作，同时也让它更符合我们.net开发的一些习惯。

using org.apache.zookeeper; using System; using System.Collections.Generic; using System.IO; using System.Linq; using System.Threading.Tasks; using System.Collections.Concurrent; using System.Threading; using System.Text; using org.apache.zookeeper.data; using org.apache.utils; using System.Diagnostics; namespace AspNetCore.ZookeeperConsole{/// /// Zookeeper辅助类/// public class ZookeeperHelper : Watcher, IDisposable{/// /// Zookeeper路径分隔符/// string sep = "/"; /// /// Zookeeper访问对象/// ZooKeeper zookeeper; /// /// Zookeeper集群地址/// string[] address; /// /// 路径监控节点列表/// ConcurrentDictionary nodeWatchers = new ConcurrentDictionary(); /// /// 节点的默认权限/// List defaultACL = ZooDefs.Ids.OPEN_ACL_UNSAFE; /// /// 默认的监听器/// DefaultWatcher defaultWatcher; /// /// 监控定时器/// System.Timers.Timer timer; /// /// 同步锁/// AutoResetEvent are = new AutoResetEvent(false); /// /// 是否正常关闭/// bool isClose = false; /// /// 回话超时时间/// public int SessionTimeout { get; set; } = 10000; /// /// 当前路径/// public string CurrentPath { get; private set; }/// /// 是否已连接Zookeeper/// public bool Connected { get { return zookeeper != null && (zookeeper.getState() == ZooKeeper.States.CONNECTED || zookeeper.getState() == ZooKeeper.States.CONNECTEDREADONLY); } }/// /// Zookeeper是否有写的权限/// public bool CanWrite { get { return zookeeper != null && zookeeper.getState() == ZooKeeper.States.CONNECTED; } }/// /// 数据编码/// public Encoding Encoding { get; set; } = Encoding.Default; /// /// 释放时发生/// public event Action OnDisposing; /// /// 在重新连接时发生/// public event Action OnConnected; /// /// 构造函数/// /// 集群地址(host:prot)public ZookeeperHelper(params string[] address) : this(address, ""){}/// /// 构造函数/// /// 集群地址(host:prot)/// 初始化根路经public ZookeeperHelper(string[] address, string root){this.address = address.ToArray(); CurrentPath = string.IsNullOrWhiteSpace(root) ? sep : root; SetLogger(new ZookeeperLogger()); timer = new System.Timers.Timer(); timer.Interval = 5000; timer.Elapsed += Timer_Elapsed; }/// /// Zookeeper的日志设置/// /// public static void SetLogger(ZookeeperLogger log){ZooKeeper.LogLevel = log.LogLevel; ZooKeeper.LogToFile = log.LogToFile; ZooKeeper.LogToTrace = log.LogToTrace; ZooKeeper.CustomLogConsumer = log; }#region 私有方法/// /// 定时器/// /// /// private void Timer_Elapsed(object sender, System.Timers.ElapsedEventArgs e){timer.Enabled = false; if (Monitor.TryEnter(timer))//每次只能一个线程进去{if (!isClose){//Thread.Sleep(SessionTimeout); if (!Connected){try{zookeeper?.closeAsync(); are.Reset(); zookeeper = new ZooKeeper(string.Join(",", address), SessionTimeout, defaultWatcher); if (are.WaitOne(SessionTimeout) && Connected)//会话未超时，表示成功连接{//挂载监听器foreach (var key in nodeWatchers.Keys){NodeWatcher watcher; if (nodeWatchers.TryGetValue(key, out watcher)){WatchAsync(key, watcher, true).Wait(); }}OnConnected?.Invoke(); Monitor.Exit(timer); return; }}catch { }timer.Enabled = true; }}Monitor.Exit(timer); }}/// /// 检查连接是否正常/// private void CheckConnection(){if (!Connected){throw new Exception("fail to connect to the server:" + string.Join(",", address)); }}/// /// 检查是否具有写的权限/// private void CheckWriten(){if (!CanWrite){throw new Exception("this connection is in readonly mode"); }}/// /// 连接数据成Zookeeper的路径格式/// /// 路径/// 连接后的路径private string Combine(params string[] paths){List list = new List(); foreach (var path in paths){var ps = path.Split(new string[] { "/", "\\" }, StringSplitOptions.RemoveEmptyEntries); foreach (var p in ps){if (p == ".")//当前路径{continue; }else if (p == "..")//回退{if (list.Count == 0){throw new ArgumentOutOfRangeException("path is out of range"); }list.RemoveAt(list.Count - 1); }else{list.Add(p); }}}return sep + string.Join(sep, list.ToArray()); }/// /// 使用指定的分隔符连接路径/// /// 分隔符/// 路径/// 连接后的路径private string MakePathName(string sep, params string[] paths){List list = new List(); foreach (var path in paths){var ps = path.Split(new string[] { "/", "\\" }, StringSplitOptions.RemoveEmptyEntries); list.AddRange(ps); }return string.Join(sep, list.ToArray()); }/// /// 获取绝对路径/// /// 路径/// 路径是否是绝对路径/// 绝对路径private string GetAbsolutePath(string path, bool isAbsolute){if (!isAbsolute){path = Combine(CurrentPath, path); }else{path = Combine(path); }return path; }#endregion/// /// 连接Zookeeper/// /// 成功连接返回true,否则返回falsepublic bool Connect(){if (Connected){return true; }if (zookeeper == null){lock (this){defaultWatcher = defaultWatcher ?? new DefaultWatcher(this, are); are.Reset(); zookeeper = new ZooKeeper(string.Join(",", address), SessionTimeout, defaultWatcher); are.WaitOne(SessionTimeout); }}if (!Connected){return false; }OnConnected?.Invoke(); return true; }/// /// 关闭连接/// public void Close(){isClose = true; if (Connected){zookeeper.closeAsync().Wait(); }}/// /// 监控回调/// /// 回调事件/// 异步public async override Task process(WatchedEvent @event){ZookeeperEvent ze = new ZookeeperEvent(@event); if (!string.IsNullOrEmpty(ze.Path)){NodeWatcher watcher; if (nodeWatchers.TryGetValue(ze.Path, out watcher)){if (watcher != null){try{watcher.Process(ze); }catch { }await WatchAsync(ze.Path, watcher, true); //重新监控}}}}/// /// 修改当前目录地址/// /// public void ChangeDirectory(string path){this.CurrentPath = Combine(path); }/// /// 切换到相对目录下/// /// public void Goto(string path){this.CurrentPath = Combine(this.CurrentPath, path); }/// /// 使用认证/// /// 认证类型/// 认证数据public void Authorize(AuthScheme scheme, string auth = ""){CheckConnection(); zookeeper.addAuthInfo(scheme.ToString().ToLower(), Encoding.GetBytes(auth)); }#region 监听与取消/// /// 对当前路径添加监控/// /// 监控/// 异步，true表示成功，false表示失败public async Task WatchAsync(WatcherEvent @delegate){return await WatchAsync(CurrentPath, @delegate, true); }/// /// 对当前路径添加监控/// /// 监控/// 异步，true表示成功，false表示失败public async Task WatchAsync(NodeWatcher watcher){return await WatchAsync(CurrentPath, watcher, true); }/// /// 对指定路径添加监控/// /// 节点路径/// 监控/// 是否绝对路径/// 异步，true表示成功，false表示失败public async Task WatchAsync(string path, WatcherEvent @delegate, bool isAbsolutePath = false){var array = await WatchManyAsync(new string[] { path }, @delegate, isAbsolutePath); return array.FirstOrDefault(); }/// /// 对指定路径添加监控/// /// 节点路径/// 监控/// 是否绝对路径/// 异步，true表示成功，false表示失败public async Task WatchAsync(string path, NodeWatcher watcher, bool isAbsolutePath = false){var array = await WatchManyAsync(new string[] { path }, watcher, isAbsolutePath); return array.FirstOrDefault(); }/// /// 监控多个路径，但是不包括子路径/// /// 节点路径/// 监控/// 是否绝对路径/// 异步，true表示成功，false表示失败public async Task WatchManyAsync(string[] paths, WatcherEvent @delegate, bool isAbsolutePath = false){var watcher = new NodeWatcher(); watcher.AllTypeChanged += @delegate; return await WatchManyAsync(paths, watcher, isAbsolutePath); }/// /// 监控多个路径，但是不包括子路径/// /// 节点路径/// 监控/// 是否绝对路径/// 异步，true表示成功，false表示失败public async Task WatchManyAsync(string[] paths, NodeWatcher watcher, bool isAbsolutePath = false){CheckConnection(); List list = new List(); foreach (var path in paths){try{var p = GetAbsolutePath(path, isAbsolutePath); if (await zookeeper.existsAsync(p, this) != null){nodeWatchers[p] = watcher; list.Add(true); }else{nodeWatchers.TryRemove(p, out _); list.Add(false); }}catch{list.Add(false); }}return list.ToArray(); }/// /// 监控当前路径，包括子路径/// /// 监控/// 异步，true表示成功，false表示失败public async Task WatchAllAsync(WatcherEvent @delegate){return await WatchAllAsync(CurrentPath, @delegate, true); }/// /// 监控当前路径，包括子路径/// /// 监控/// 异步，true表示成功，false表示失败public async Task WatchAllAsync(NodeWatcher watcher){return await WatchAllAsync(CurrentPath, watcher, true); }/// /// 监控指定路径，包括子路径/// /// 节点路径/// 监控/// 是否绝对路径/// 异步，true表示成功，false表示失败public async Task WatchAllAsync(string path, WatcherEvent @delegate, bool isAbsolutePath = false){var array = await WatchAllAsync(new string[] { path }, @delegate, isAbsolutePath); return array.FirstOrDefault(); }/// /// 监控指定路径，包括子路径/// /// 节点路径/// 监控/// 是否绝对路径/// 异步，true表示成功，false表示失败public async Task WatchAllAsync(string path, NodeWatcher watcher, bool isAbsolutePath = false){var array = await WatchAllAsync(new string[] { path }, watcher, isAbsolutePath); return array.FirstOrDefault(); }/// /// 监控所有路径，包括子路径/// /// 节点路径/// 监控/// 是否绝对路径/// 异步，true表示成功，false表示失败public async Task WatchAllAsync(string[] paths, WatcherEvent @delegate, bool isAbsolutePath = false){var watcher = new NodeWatcher(); watcher.AllTypeChanged += @delegate; return await WatchAllAsync(paths, watcher, isAbsolutePath); }/// /// 监控所有路径，包括子路径/// /// 节点路径/// 监控/// 是否绝对路径/// 异步，true表示成功，false表示失败public async Task WatchAllAsync(string[] paths, NodeWatcher watcher, bool isAbsolutePath = false){CheckConnection(); List list = new List(); foreach (var path in paths){try{var p = GetAbsolutePath(path, isAbsolutePath); if (await zookeeper.existsAsync(p, this) != null){nodeWatchers[p] = watcher; list.Add(true); var result = await zookeeper.getChildrenAsync(p); await WatchAllAsync(result.Children.Select(c => Combine(p, c)).ToArray(), watcher, true); }else{nodeWatchers.TryRemove(p, out _); list.Add(false); }}catch{list.Add(false); }}return list.ToArray(); }/// /// 取消多个指定路径上的监控/// /// 节点路径/// 是否绝对路径/// 异步public async Task CancelAsync(string path, bool isAbsolutePath = true){await CancelAsync(new string[] { path }, isAbsolutePath); }/// /// 取消多个指定路径上的监控/// /// 节点路径/// 是否绝对路径/// 异步public async Task CancelAsync(string[] paths, bool isAbsolutePath = true){foreach (var path in paths){var p = GetAbsolutePath(path, isAbsolutePath); nodeWatchers.TryRemove(p, out _); await Task.CompletedTask; }}/// /// 获取指定路径上的监控/// /// 节点路径/// 是否绝对路径/// 存在则返回监控对象，否则返回nullpublic NodeWatcher GetWatcher(string path, bool isAbsolutePath = true){path = GetAbsolutePath(path, isAbsolutePath); NodeWatcher watcher; if (nodeWatchers.TryGetValue(path, out watcher)){return watcher; }return null; }#endregion#region 基本数据操作/// /// 当前节点是否存在/// /// 存在返回true，否则返回falsepublic bool Exists(){return ExistsAsync().GetAwaiter().GetResult(); }/// /// 指定节点是否存在（相对当前节点）/// /// 节点路径/// 存在返回true，否则返回falsepublic bool Exists(string path){return ExistsAsync(path).GetAwaiter().GetResult(); }/// /// 指定节点是否存在/// /// 绝对路径/// 存在返回true，否则返回falsepublic bool ExistsByAbsolutePath(string absolutePath){return ExistsByAbsolutePathAsync(absolutePath).GetAwaiter().GetResult(); }/// /// 当前节点是否存在/// /// 异步，存在返回true，否则返回falsepublic async Task ExistsAsync(){return await ExistsByAbsolutePathAsync(CurrentPath); }/// /// 指定节点是否存在（相对当前节点）/// /// 节点路径/// 异步，存在返回true，否则返回falsepublic async Task ExistsAsync(string path){path = Combine(CurrentPath, path); return await ExistsByAbsolutePathAsync(path); }/// /// 指定节点是否存在/// /// 绝对路径/// 异步，存在返回true，否则返回falsepublic async Task ExistsByAbsolutePathAsync(string absolutePath){absolutePath = Combine(absolutePath); return await zookeeper.existsAsync(absolutePath, false) != null; }/// /// 添加或者修改当前路径上的数据/// /// 数据/// 是否持久节点/// 是否顺序节点/// znode节点名，不包含父节点路径public string SetData(string value, bool persistent = false, bool sequential = false){return SetDataAsync(value, persistent, sequential).GetAwaiter().GetResult(); }/// /// 添加或者修改指定相对路径上的数据/// /// 相对路径/// 数据/// 【C#如何连接使用Zookeeper】是否持久节点/// 是否顺序节点/// znode节点名，不包含父节点路径public string SetData(string path, string value, bool persistent = false, bool sequential = false){return SetDataAsync(path, value, persistent, sequential).GetAwaiter().GetResult(); }/// /// 添加或者修改指定绝对路径上的数据/// /// 绝对路径/// 数据/// 是否持久节点/// 是否顺序节点/// znode节点名，不包含父节点路径public string SetDataByAbsolutePath(string absolutePath, string value, bool persistent = false, bool sequential = false){return SetDataByAbsolutePathAsync(absolutePath, value, persistent, sequential).GetAwaiter().GetResult(); }/// /// 添加或者修改当前路径上的数据/// /// 数据/// 是否持久节点/// 是否顺序节点/// znode节点名，不包含父节点路径public async Task SetDataAsync(string value, bool persistent = false, bool sequential = false){return await SetDataByAbsolutePathAsync(CurrentPath, value, persistent, sequential); }/// /// 添加或者修改指定相对路径上的数据/// /// 相对路径/// 数据/// 是否持久节点/// 是否顺序节点/// znode节点名，不包含父节点路径public async Task SetDataAsync(string path, string value, bool persistent = false, bool sequential = false){path = Combine(CurrentPath, path); return await SetDataByAbsolutePathAsync(path, value, persistent, sequential); }/// /// 添加或者修改指定绝对路径上的数据/// /// 绝对路径/// 数据/// 是否持久节点/// 是否顺序节点/// znode节点名，不包含父节点路径public async Task SetDataByAbsolutePathAsync(string absolutePath, string value, bool persistent = false, bool sequential = false){CheckConnection(); CheckWriten(); absolutePath = Combine(absolutePath); if (await zookeeper.existsAsync(absolutePath, false) == null){absolutePath = await zookeeper.createAsync(absolutePath, Encoding.GetBytes(value), defaultACL, persistent ?sequential ? CreateMode.PERSISTENT_SEQUENTIAL : CreateMode.PERSISTENT :sequential ? CreateMode.EPHEMERAL_SEQUENTIAL : CreateMode.EPHEMERAL); }else{await zookeeper.setDataAsync(absolutePath, Encoding.GetBytes(value)); }return absolutePath.Split(new string[] { sep }, StringSplitOptions.RemoveEmptyEntries).LastOrDefault(); }/// /// 获取指定相对路径上的数据/// /// 相对路径/// 相对路径上的数据public string GetData(string path){return GetDataAsync(path).GetAwaiter().GetResult(); }/// /// 获取指定绝对路径上的数据/// /// 绝对路径/// 相对路径上的数据public string GetDataByAbsolutePath(string absolutePath){return GetDataByAbsolutePathAsync(absolutePath).GetAwaiter().GetResult(); }/// /// 获取指定相对路径上的数据/// /// 相对路径/// 相对路径上的数据public async Task GetDataAsync(string path){path = Combine(CurrentPath, path); return await GetDataByAbsolutePathAsync(path); }/// /// 获取指定绝对路径上的数据/// /// 绝对路径/// 绝对路径上的数据public async Task GetDataByAbsolutePathAsync(string absolutePath){CheckConnection(); absolutePath = Combine(absolutePath); if (await zookeeper.existsAsync(absolutePath, false) == null){return ""; }var data = https://www.it610.com/article/await zookeeper.getDataAsync(absolutePath, false); return Encoding.GetString(data.Data); }/// /// 获取指定节点及其字节点的所有值，使用路径做键返回字典型/// /// /// public async Task GetDictionaryAsync(string sep = ":"){CheckConnection(); Dictionary dict = new Dictionary(); async Task action(string path){try{var result = await zookeeper.getChildrenAsync(path, false); string name = MakePathName(sep, path); dict[name] = await GetDataByAbsolutePathAsync(path); foreach (var child in result.Children){var p = Combine(path, child); await action(p); }}catch (Exception ex){}}await action(CurrentPath); return dict; }/// /// 获取子节点/// /// 相对路径/// 是否按时间排序/// 子节点数组（节点路径不含父节点路径）public async Task GetChildrenAsync(string path, bool order = false){path = Combine(CurrentPath, path); return await GetChildrenByAbsolutePathAsync(path, order); }/// /// 获取指定路径绝对路径下的子节点/// /// 绝对路径/// 是否按时间排序/// 子节点数组（节点路径不含父节点路径）public async Task GetChildrenByAbsolutePathAsync(string absolutePath, bool order = false){var result = await zookeeper.getChildrenAsync(absolutePath, false); if (!order){return result.Children.ToArray(); }List<(string, long)> list = new List<(string, long)>(); foreach (var child in result.Children){var p = Combine(absolutePath, child); var stat = await zookeeper.existsAsync(p, false); if (stat != null){list.Add((child, stat.getCtime())); }}return list.OrderBy(l => l.Item2).Select(l => l.Item1).ToArray(); }/// /// 移除当前路径节点/// public void Delete(){DeleteAsync().Wait(); }/// /// 移除相对当前的指定路径节点及子节点/// /// 相对路径public void Delete(string path){DeleteAsync(path).Wait(); }/// /// 移除指定绝对路径节点及子节点/// /// 绝对路径public void DeleteByAbsolutePath(string absolutePath){DeleteByAbsolutePathAsync(absolutePath).Wait(); }/// /// 移除当前路径节点/// public async Task DeleteAsync(){await DeleteByAbsolutePathAsync(CurrentPath); }/// /// 移除相对当前的指定路径节点及子节点/// /// 相对路径public async Task DeleteAsync(string path){path = Combine(CurrentPath, path); await DeleteByAbsolutePathAsync(path); }/// /// 移除指定绝对路径节点及子节点/// /// 绝对路径public async Task DeleteByAbsolutePathAsync(string absolutePath){if (await ExistsByAbsolutePathAsync(absolutePath)){var children = await GetChildrenByAbsolutePathAsync(absolutePath); foreach (var child in children){var path = Combine(absolutePath, child); await DeleteByAbsolutePathAsync(path); }absolutePath = Combine(absolutePath); await zookeeper.deleteAsync(absolutePath); }}#endregion/// /// 释放资源/// public void Dispose(){OnDisposing?.Invoke(); Close(); timer?.Dispose(); nodeWatchers?.Clear(); are?.Dispose(); GC.Collect(); }/// /// 默认的监听器，用于初始化使用/// public class DefaultWatcher : Watcher{/// /// waithandle同步/// EventWaitHandle ewh; /// /// 辅助类/// ZookeeperHelper zookeeperHelper; public DefaultWatcher(ZookeeperHelper zookeeperHelper, EventWaitHandle ewh){this.ewh = ewh; this.zookeeperHelper = zookeeperHelper; }/// /// 回调/// /// 监听事件对象/// public override Task process(WatchedEvent @event){var state = @event.getState(); if (state == Event.KeeperState.ConnectedReadOnly || state == Event.KeeperState.SyncConnected)//连接时{ewh.Set(); }else if ((state == Event.KeeperState.Expired) && !zookeeperHelper.isClose)//回话过期重新建立连接{zookeeperHelper.timer.Enabled = true; }return Task.FromResult(1); }}}/// /// 认证类型/// public enum AuthScheme{/// /// 下面只有一个id，叫anyone，world:anyone代表任何人，zookeeper中对所有人有权限的结点就是属于world:anyone类型的。创建节点的默认权限。有唯一的id是anyone授权的时候的模式为 world:anyone:rwadc 表示所有人都对这个节点有rwadc的权限/// World = 0,/// ///不需要id,只要是通过authentication的user都有权限（zookeeper支持通过kerberos来进行authencation, 也支持username/password形式的authentication)/// Auth = 1,/// /// 它对应的id为username:BASE64(SHA1(password))，它需要先通过加密过的username:password形式的authentication。/// Digest = 2,/// ///它对应的id为客户机的IP地址，设置的时候可以设置一个ip段，比如ip:192.168.1.0/16。/// Ip = 3,/// /// 在这种scheme情况下，对应的id拥有超级权限，可以做任何事情(cdrwa）/// Super = 4}/// /// Zookeeper事件数据/// public class ZookeeperEvent{public ZookeeperEvent(WatchedEvent @event){switch (@event.getState()){case Watcher.Event.KeeperState.AuthFailed: State = EventState.AuthFailed; break; case Watcher.Event.KeeperState.ConnectedReadOnly: State = EventState.ConnectedReadOnly; break; case Watcher.Event.KeeperState.Disconnected: State = EventState.Disconnected; break; case Watcher.Event.KeeperState.Expired: State = EventState.Expired; break; case Watcher.Event.KeeperState.SyncConnected: State = EventState.SyncConnected; break; }switch (@event.get_Type()){case Watcher.Event.EventType.NodeChildrenChanged: Type = EventType.NodeChildrenChanged; break; case Watcher.Event.EventType.NodeCreated: Type = EventType.NodeCreated; break; case Watcher.Event.EventType.NodeDataChanged: Type = EventType.NodeDataChanged; break; case Watcher.Event.EventType.NodeDeleted: Type = EventType.NodeDeleted; break; case Watcher.Event.EventType.None: Type = EventType.None; break; }Path = @event.getPath(); }/// /// 当前连接状态/// public EventState State { get; private set; }/// /// 事件类型/// public EventType Type { get; private set; }/// /// 事件路径/// public string Path { get; private set; }/// /// 连接状态/// public enum EventState{/// /// 超时/// Expired = -112,/// /// 连接已断开/// Disconnected = 0,/// /// 已建立连接/// SyncConnected = 3,/// /// 认证失败/// AuthFailed = 4,/// /// 已建立连接，但是只支持只读模式/// ConnectedReadOnly = 5}/// /// 时间类型/// public enum EventType{/// /// 空类型，如：建立连接时/// None = -1,/// /// 节点创建时/// NodeCreated = 1,/// /// 节点删除时/// NodeDeleted = 2,/// /// 节点数据改变时/// NodeDataChanged = 3,/// /// 节点增加子节点时/// NodeChildrenChanged = 4}}/// /// 监控对象/// public class NodeWatcher{/// /// 节点创建时调用事件/// public event WatcherEvent NodeCreated; /// /// 节点删除时调用事件/// public event WatcherEvent NodeDeleted; /// /// 节点数据改变时调用事件/// public event WatcherEvent NodeDataChanged; /// /// 节点增加子节点时调用事件/// public event WatcherEvent NodeChildrenChanged; /// /// 不区分类型，所有的类型都会调用/// public event WatcherEvent AllTypeChanged; /// /// 触发，执行事件/// /// public void Process(ZookeeperEvent @event){try{switch (@event.Type){case ZookeeperEvent.EventType.NodeChildrenChanged:NodeChildrenChanged?.Invoke(@event); break; case ZookeeperEvent.EventType.NodeCreated:NodeCreated?.Invoke(@event); break; case ZookeeperEvent.EventType.NodeDeleted:NodeDeleted?.Invoke(@event); break; case ZookeeperEvent.EventType.NodeDataChanged:NodeDataChanged?.Invoke(@event); break; }AllTypeChanged?.Invoke(@event); }catch { }}}/// /// 监控事件委托/// /// public delegate void WatcherEvent(ZookeeperEvent @event); /// /// Zookeeper默认日志记录/// public class ZookeeperLogger : ILogConsumer{/// /// 是否记录日志到文件/// public bool LogToFile { get; set; } = false; /// /// 是否记录堆栈信息/// public bool LogToTrace { get; set; } = true; /// /// 日志级别/// public TraceLevel LogLevel { get; set; } = TraceLevel.Warning; /// /// 日志记录/// /// /// /// /// public virtual void Log(TraceLevel severity, string className, string message, Exception exception){Console.WriteLine(string.Format("Level:{0}className:{1}message:{2}", severity, className, message)); Console.WriteLine(exception.StackTrace); }}}

简单的使用例子：

using System; using System.Collections.Generic; using System.Text; using System.Threading; namespace AspNetCore.ZookeeperConsole{class Program{static void Main(string[] args){//Zookeeper连接字符串，采用host:port格式，多个地址之间使用逗号（,）隔开string[] address = new string[] { "192.168.209.133:2181", "192.168.209.133:2181", "192.168.209.133:2181" }; //会话超时时间,单位毫秒int sessionTimeOut = 10000; ZookeeperHelper zookeeperHelper = new ZookeeperHelper(address, "/"); zookeeperHelper.SessionTimeout = sessionTimeOut; zookeeperHelper.Connect(); //发起连接while (!zookeeperHelper.Connected){Thread.Sleep(1000); //停一秒，等待连接完成}//创建znode节点{zookeeperHelper.SetData("/mynode", "hello world", true, false); Console.WriteLine("完成创建节点"); }//节点是否存在{var exists = zookeeperHelper.Exists("/mynode"); Console.WriteLine("节点是否存在：" + exists); }//添加监听器{zookeeperHelper.WatchAsync("/mynode", (e) =>{Console.WriteLine($"recieve: Path-{e.Path}State-{e.State}Type-{e.Type}"); }).Wait(); }//获取节点数据{var value = https://www.it610.com/article/zookeeperHelper.GetData("/mynode"); Console.WriteLine("完成读取节点：" + value); }//设置节点数据{zookeeperHelper.SetData("/mynode", "hello world again"); Console.WriteLine("设置节点数据"); }//重新获取节点数据{var value = https://www.it610.com/article/zookeeperHelper.GetData("/mynode"); Console.WriteLine("重新获取节点数据：" + value); }//移除节点{zookeeperHelper.Delete("/mynode"); Console.WriteLine("移除节点"); }Console.WriteLine("完成"); Console.ReadKey(); }}}

执行结果：