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 = Encoding.UTF8.GetBytes("hello world");
List<ACL> acl = ZooDefs.Ids.OPEN\_ACL\_UNSAFE;//创建节点时的acl权限,也可以使用下面的自定义权限
//List<ACL> acl = new List<ACL>() {
// 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 = Encoding.UTF8.GetString(dataResult.Data);
Console.WriteLine("完成读取节点:" + value);
}
//设置节点数据
{
var data = Encoding.UTF8.GetBytes("hello world again");
zooKeeper.setDataAsync("/mynode", data);
Console.WriteLine("设置节点数据");
}
//重新获取节点数据
{
var dataResult = zooKeeper.getDataAsync("/mynode", new MyWatcher("GetWatcher")).GetAwaiter().GetResult();
var value = 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> acl = new List<ACL>() {
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<Stat> existsAsync(string path, Watcher watcher);
public Task<Stat> existsAsync(string path, bool watch = false);
//获取znode节点数据
public Task<DataResult> getDataAsync(string path, bool watch = false);
public Task<DataResult> getDataAsync(string path, Watcher watcher);
//设置指定znode节点的数据
public Task<Stat> setDataAsync(string path, byte\[\] data, int version = -1);
//获取指定znode节点的子节点,注意,监听器是注册给当前节点的,而非子节点
public Task<ChildrenResult> getChildrenAsync(string path, Watcher watcher);
public Task<ChildrenResult> getChildrenAsync(string path, bool watch = false);
可以比较一下上一节介绍的zkCli对znode节点的操作就很容易理解了。
另外,需要注意的是,existsAsync方法、getDataAsync方法和getChildrenAsync方法可以在指定的znode注册一个监听器,setDataAsync方法却没有这个注册功能,这个是因为Zookeeper注册的监听器只会响应一次,当需要再次响应时,需要重新注册,这时就可以调用existsAsync方法或者getDataAsync方法或者getChildrenAsync方法进行重新注册了!
上一节说到ACL权限不仅可以在创建是给予,在创建后也可以修改,ZookeeperNetEx操作znode的ACL权限使用的方法如下:
//获取ACL权限
public Task<ACLResult> getACLAsync(string path);
//设置ACL权限
public Task<Stat> setACLAsync(string path, List<ACL> 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
{
/// <summary>
/// Zookeeper辅助类
/// </summary>
public class ZookeeperHelper : Watcher, IDisposable
{
/// <summary>
/// Zookeeper路径分隔符
/// </summary>
string sep = "/";
/// <summary>
/// Zookeeper访问对象
/// </summary>
ZooKeeper zookeeper;
/// <summary>
/// Zookeeper集群地址
/// </summary>
string\[\] address;
/// <summary>
/// 路径监控节点列表
/// </summary>
ConcurrentDictionary<string, NodeWatcher> nodeWatchers = new ConcurrentDictionary<string, NodeWatcher>();
/// <summary>
/// 节点的默认权限
/// </summary>
List<ACL> defaultACL = ZooDefs.Ids.OPEN\_ACL\_UNSAFE;
/// <summary>
/// 默认的监听器
/// </summary>
DefaultWatcher defaultWatcher;
/// <summary>
/// 监控定时器
/// </summary>
System.Timers.Timer timer;
/// <summary>
/// 同步锁
/// </summary>
AutoResetEvent are = new AutoResetEvent(false);
/// <summary>
/// 是否正常关闭
/// </summary>
bool isClose = false;
/// <summary>
/// 回话超时时间
/// </summary>
public int SessionTimeout { get; set; } = 10000;
/// <summary>
/// 当前路径
/// </summary>
public string CurrentPath { get; private set; }
/// <summary>
/// 是否已连接Zookeeper
/// </summary>
public bool Connected { get { return zookeeper != null && (zookeeper.getState() == ZooKeeper.States.CONNECTED || zookeeper.getState() == ZooKeeper.States.CONNECTEDREADONLY); } }
/// <summary>
/// Zookeeper是否有写的权限
/// </summary>
public bool CanWrite { get { return zookeeper != null && zookeeper.getState() == ZooKeeper.States.CONNECTED; } }
/// <summary>
/// 数据编码
/// </summary>
public Encoding Encoding { get; set; } = Encoding.Default;
/// <summary>
/// 释放时发生
/// </summary>
public event Action OnDisposing;
/// <summary>
/// 在重新连接时发生
/// </summary>
public event Action OnConnected;
/// <summary>
/// 构造函数
/// </summary>
/// <param name="address">集群地址(host:prot)</param>
public ZookeeperHelper(params string\[\] address) : this(address, "")
{
}
/// <summary>
/// 构造函数
/// </summary>
/// <param name="address">集群地址(host:prot)</param>
/// <param name="root">初始化根路经</param>
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;
}
/// <summary>
/// Zookeeper的日志设置
/// </summary>
/// <param name="log"></param>
public static void SetLogger(ZookeeperLogger log)
{
ZooKeeper.LogLevel = log.LogLevel;
ZooKeeper.LogToFile = log.LogToFile;
ZooKeeper.LogToTrace = log.LogToTrace;
ZooKeeper.CustomLogConsumer = log;
}
#region 私有方法
/// <summary>
/// 定时器
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
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);
}
}
/// <summary>
/// 检查连接是否正常
/// </summary>
private void CheckConnection()
{
if (!Connected)
{
throw new Exception("fail to connect to the server:" + string.Join(",", address));
}
}
/// <summary>
/// 检查是否具有写的权限
/// </summary>
private void CheckWriten()
{
if (!CanWrite)
{
throw new Exception("this connection is in readonly mode");
}
}
/// <summary>
/// 连接数据成Zookeeper的路径格式
/// </summary>
/// <param name="paths">路径</param>
/// <returns>连接后的路径</returns>
private string Combine(params string\[\] paths)
{
List<string> list = new List<string>();
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());
}
/// <summary>
/// 使用指定的分隔符连接路径
/// </summary>
/// <param name="sep">分隔符</param>
/// <param name="paths">路径</param>
/// <returns>连接后的路径</returns>
private string MakePathName(string sep, params string\[\] paths)
{
List<string> list = new List<string>();
foreach (var path in paths)
{
var ps = path.Split(new string\[\] { "/", "\\\\" }, StringSplitOptions.RemoveEmptyEntries);
list.AddRange(ps);
}
return string.Join(sep, list.ToArray());
}
/// <summary>
/// 获取绝对路径
/// </summary>
/// <param name="path">路径</param>
/// <param name="isAbsolute">路径是否是绝对路径</param>
/// <returns>绝对路径</returns>
private string GetAbsolutePath(string path, bool isAbsolute)
{
if (!isAbsolute)
{
path = Combine(CurrentPath, path);
}
else
{
path = Combine(path);
}
return path;
}
#endregion
/// <summary>
/// 连接Zookeeper
/// </summary>
/// <returns>成功连接返回true,否则返回false</returns>
public 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;
}
/// <summary>
/// 关闭连接
/// </summary>
public void Close()
{
isClose = true;
if (Connected)
{
zookeeper.closeAsync().Wait();
}
}
/// <summary>
/// 监控回调
/// </summary>
/// <param name="event">回调事件</param>
/// <returns>异步</returns>
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);//重新监控
}
}
}
}
/// <summary>
/// 修改当前目录地址
/// </summary>
/// <param name="path"></param>
public void ChangeDirectory(string path)
{
this.CurrentPath = Combine(path);
}
/// <summary>
/// 切换到相对目录下
/// </summary>
/// <param name="path"></param>
public void Goto(string path)
{
this.CurrentPath = Combine(this.CurrentPath, path);
}
/// <summary>
/// 使用认证
/// </summary>
/// <param name="scheme">认证类型</param>
/// <param name="auth">认证数据</param>
public void Authorize(AuthScheme scheme, string auth = "")
{
CheckConnection();
zookeeper.addAuthInfo(scheme.ToString().ToLower(), Encoding.GetBytes(auth));
}
#region 监听与取消
/// <summary>
/// 对当前路径添加监控
/// </summary>
/// <param name="delegate">监控</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool> WatchAsync(WatcherEvent @delegate)
{
return await WatchAsync(CurrentPath, @delegate, true);
}
/// <summary>
/// 对当前路径添加监控
/// </summary>
/// <param name="watcher">监控</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool> WatchAsync(NodeWatcher watcher)
{
return await WatchAsync(CurrentPath, watcher, true);
}
/// <summary>
/// 对指定路径添加监控
/// </summary>
/// <param name="path">节点路径</param>
/// <param name="delegate">监控</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool> WatchAsync(string path, WatcherEvent @delegate, bool isAbsolutePath = false)
{
var array = await WatchManyAsync(new string\[\] { path }, @delegate, isAbsolutePath);
return array.FirstOrDefault();
}
/// <summary>
/// 对指定路径添加监控
/// </summary>
/// <param name="path">节点路径</param>
/// <param name="watcher">监控</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool> WatchAsync(string path, NodeWatcher watcher, bool isAbsolutePath = false)
{
var array = await WatchManyAsync(new string\[\] { path }, watcher, isAbsolutePath);
return array.FirstOrDefault();
}
/// <summary>
/// 监控多个路径,但是不包括子路径
/// </summary>
/// <param name="paths">节点路径</param>
/// <param name="delegate">监控</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool\[\]> WatchManyAsync(string\[\] paths, WatcherEvent @delegate, bool isAbsolutePath = false)
{
var watcher = new NodeWatcher();
watcher.AllTypeChanged += @delegate;
return await WatchManyAsync(paths, watcher, isAbsolutePath);
}
/// <summary>
/// 监控多个路径,但是不包括子路径
/// </summary>
/// <param name="paths">节点路径</param>
/// <param name="watcher">监控</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool\[\]> WatchManyAsync(string\[\] paths, NodeWatcher watcher, bool isAbsolutePath = false)
{
CheckConnection();
List<bool> list = new List<bool>();
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();
}
/// <summary>
/// 监控当前路径,包括子路径
/// </summary>
/// <param name="delegate">监控</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool> WatchAllAsync(WatcherEvent @delegate)
{
return await WatchAllAsync(CurrentPath, @delegate, true);
}
/// <summary>
/// 监控当前路径,包括子路径
/// </summary>
/// <param name="watcher">监控</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool> WatchAllAsync(NodeWatcher watcher)
{
return await WatchAllAsync(CurrentPath, watcher, true);
}
/// <summary>
/// 监控指定路径,包括子路径
/// </summary>
/// <param name="path">节点路径</param>
/// <param name="delegate">监控</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool> WatchAllAsync(string path, WatcherEvent @delegate, bool isAbsolutePath = false)
{
var array = await WatchAllAsync(new string\[\] { path }, @delegate, isAbsolutePath);
return array.FirstOrDefault();
}
/// <summary>
/// 监控指定路径,包括子路径
/// </summary>
/// <param name="path">节点路径</param>
/// <param name="watcher">监控</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool> WatchAllAsync(string path, NodeWatcher watcher, bool isAbsolutePath = false)
{
var array = await WatchAllAsync(new string\[\] { path }, watcher, isAbsolutePath);
return array.FirstOrDefault();
}
/// <summary>
/// 监控所有路径,包括子路径
/// </summary>
/// <param name="paths">节点路径</param>
/// <param name="delegate">监控</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool\[\]> WatchAllAsync(string\[\] paths, WatcherEvent @delegate, bool isAbsolutePath = false)
{
var watcher = new NodeWatcher();
watcher.AllTypeChanged += @delegate;
return await WatchAllAsync(paths, watcher, isAbsolutePath);
}
/// <summary>
/// 监控所有路径,包括子路径
/// </summary>
/// <param name="paths">节点路径</param>
/// <param name="watcher">监控</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>异步,true表示成功,false表示失败</returns>
public async Task<bool\[\]> WatchAllAsync(string\[\] paths, NodeWatcher watcher, bool isAbsolutePath = false)
{
CheckConnection();
List<bool> list = new List<bool>();
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();
}
/// <summary>
/// 取消多个指定路径上的监控
/// </summary>
/// <param name="path">节点路径</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>异步</returns>
public async Task CancelAsync(string path, bool isAbsolutePath = true)
{
await CancelAsync(new string\[\] { path }, isAbsolutePath);
}
/// <summary>
/// 取消多个指定路径上的监控
/// </summary>
/// <param name="path">节点路径</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>异步</returns>
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;
}
}
/// <summary>
/// 获取指定路径上的监控
/// </summary>
/// <param name="path">节点路径</param>
/// <param name="isAbsolutePath">是否绝对路径</param>
/// <returns>存在则返回监控对象,否则返回null</returns>
public 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 基本数据操作
/// <summary>
/// 当前节点是否存在
/// </summary>
/// <returns>存在返回true,否则返回false</returns>
public bool Exists()
{
return ExistsAsync().GetAwaiter().GetResult();
}
/// <summary>
/// 指定节点是否存在(相对当前节点)
/// </summary>
/// <param name="path">节点路径</param>
/// <returns>存在返回true,否则返回false</returns>
public bool Exists(string path)
{
return ExistsAsync(path).GetAwaiter().GetResult();
}
/// <summary>
/// 指定节点是否存在
/// </summary>
/// <param name="absolutePath">绝对路径</param>
/// <returns>存在返回true,否则返回false</returns>
public bool ExistsByAbsolutePath(string absolutePath)
{
return ExistsByAbsolutePathAsync(absolutePath).GetAwaiter().GetResult();
}
/// <summary>
/// 当前节点是否存在
/// </summary>
/// <returns>异步,存在返回true,否则返回false</returns>
public async Task<bool> ExistsAsync()
{
return await ExistsByAbsolutePathAsync(CurrentPath);
}
/// <summary>
/// 指定节点是否存在(相对当前节点)
/// </summary>
/// <param name="path">节点路径</param>
/// <returns>异步,存在返回true,否则返回false</returns>
public async Task<bool> ExistsAsync(string path)
{
path = Combine(CurrentPath, path);
return await ExistsByAbsolutePathAsync(path);
}
/// <summary>
/// 指定节点是否存在
/// </summary>
/// <param name="absolutePath">绝对路径</param>
/// <returns>异步,存在返回true,否则返回false</returns>
public async Task<bool> ExistsByAbsolutePathAsync(string absolutePath)
{
absolutePath = Combine(absolutePath);
return await zookeeper.existsAsync(absolutePath, false) != null;
}
/// <summary>
/// 添加或者修改当前路径上的数据
/// </summary>
/// <param name="value">数据</param>
/// <param name="persistent">是否持久节点</param>
/// <param name="sequential">是否顺序节点</param>
/// <returns>znode节点名,不包含父节点路径</returns>
public string SetData(string value, bool persistent = false, bool sequential = false)
{
return SetDataAsync(value, persistent, sequential).GetAwaiter().GetResult();
}
/// <summary>
/// 添加或者修改指定相对路径上的数据
/// </summary>
/// <param name="path">相对路径</param>
/// <param name="value">数据</param>
/// <param name="persistent">是否持久节点</param>
/// <param name="sequential">是否顺序节点</param>
/// <returns>znode节点名,不包含父节点路径</returns>
public string SetData(string path, string value, bool persistent = false, bool sequential = false)
{
return SetDataAsync(path, value, persistent, sequential).GetAwaiter().GetResult();
}
/// <summary>
/// 添加或者修改指定绝对路径上的数据
/// </summary>
/// <param name="absolutePath">绝对路径</param>
/// <param name="value">数据</param>
/// <param name="persistent">是否持久节点</param>
/// <param name="sequential">是否顺序节点</param>
/// <returns>znode节点名,不包含父节点路径</returns>
public string SetDataByAbsolutePath(string absolutePath, string value, bool persistent = false, bool sequential = false)
{
return SetDataByAbsolutePathAsync(absolutePath, value, persistent, sequential).GetAwaiter().GetResult();
}
/// <summary>
/// 添加或者修改当前路径上的数据
/// </summary>
/// <param name="value">数据</param>
/// <param name="persistent">是否持久节点</param>
/// <param name="sequential">是否顺序节点</param>
/// <returns>znode节点名,不包含父节点路径</returns>
public async Task<string> SetDataAsync(string value, bool persistent = false, bool sequential = false)
{
return await SetDataByAbsolutePathAsync(CurrentPath, value, persistent, sequential);
}
/// <summary>
/// 添加或者修改指定相对路径上的数据
/// </summary>
/// <param name="path">相对路径</param>
/// <param name="value">数据</param>
/// <param name="persistent">是否持久节点</param>
/// <param name="sequential">是否顺序节点</param>
/// <returns>znode节点名,不包含父节点路径</returns>
public async Task<string> SetDataAsync(string path, string value, bool persistent = false, bool sequential = false)
{
path = Combine(CurrentPath, path);
return await SetDataByAbsolutePathAsync(path, value, persistent, sequential);
}
/// <summary>
/// 添加或者修改指定绝对路径上的数据
/// </summary>
/// <param name="absolutePath">绝对路径</param>
/// <param name="value">数据</param>
/// <param name="persistent">是否持久节点</param>
/// <param name="sequential">是否顺序节点</param>
/// <returns>znode节点名,不包含父节点路径</returns>
public async Task<string> 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();
}
/// <summary>
/// 获取指定相对路径上的数据
/// </summary>
/// <param name="path">相对路径</param>
/// <returns>相对路径上的数据</returns>
public string GetData(string path)
{
return GetDataAsync(path).GetAwaiter().GetResult();
}
/// <summary>
/// 获取指定绝对路径上的数据
/// </summary>
/// <param name="absolutePath">绝对路径</param>
/// <returns>相对路径上的数据</returns>
public string GetDataByAbsolutePath(string absolutePath)
{
return GetDataByAbsolutePathAsync(absolutePath).GetAwaiter().GetResult();
}
/// <summary>
/// 获取指定相对路径上的数据
/// </summary>
/// <param name="path">相对路径</param>
/// <returns>相对路径上的数据</returns>
public async Task<string> GetDataAsync(string path)
{
path = Combine(CurrentPath, path);
return await GetDataByAbsolutePathAsync(path);
}
/// <summary>
/// 获取指定绝对路径上的数据
/// </summary>
/// <param name="absolutePath">绝对路径</param>
/// <returns>绝对路径上的数据</returns>
public async Task<string> GetDataByAbsolutePathAsync(string absolutePath)
{
CheckConnection();
absolutePath = Combine(absolutePath);
if (await zookeeper.existsAsync(absolutePath, false) == null)
{
return "";
}
var data = await zookeeper.getDataAsync(absolutePath, false);
return Encoding.GetString(data.Data);
}
/// <summary>
/// 获取指定节点及其字节点的所有值,使用路径做键返回字典型
/// </summary>
/// <param name="sep"></param>
/// <returns></returns>
public async Task<IDictionary<string, string>> GetDictionaryAsync(string sep = ":")
{
CheckConnection();
Dictionary<string, string> dict = new Dictionary<string, string>();
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;
}
/// <summary>
/// 获取子节点
/// </summary>
/// <param name="path">相对路径</param>
/// <param name="order">是否按时间排序</param>
/// <returns>子节点数组(节点路径不含父节点路径)</returns>
public async Task<string\[\]> GetChildrenAsync(string path, bool order = false)
{
path = Combine(CurrentPath, path);
return await GetChildrenByAbsolutePathAsync(path, order);
}
/// <summary>
/// 获取指定路径绝对路径下的子节点
/// </summary>
/// <param name="absolutePath">绝对路径</param>
/// <param name="order">是否按时间排序</param>
/// <returns>子节点数组(节点路径不含父节点路径)</returns>
public async Task<string\[\]> 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();
}
/// <summary>
/// 移除当前路径节点
/// </summary>
public void Delete()
{
DeleteAsync().Wait();
}
/// <summary>
/// 移除相对当前的指定路径节点及子节点
/// </summary>
/// <param name="path">相对路径</param>
public void Delete(string path)
{
DeleteAsync(path).Wait();
}
/// <summary>
/// 移除指定绝对路径节点及子节点
/// </summary>
/// <param name="absolutePath">绝对路径</param>
public void DeleteByAbsolutePath(string absolutePath)
{
DeleteByAbsolutePathAsync(absolutePath).Wait();
}
/// <summary>
/// 移除当前路径节点
/// </summary>
public async Task DeleteAsync()
{
await DeleteByAbsolutePathAsync(CurrentPath);
}
/// <summary>
/// 移除相对当前的指定路径节点及子节点
/// </summary>
/// <param name="path">相对路径</param>
public async Task DeleteAsync(string path)
{
path = Combine(CurrentPath, path);
await DeleteByAbsolutePathAsync(path);
}
/// <summary>
/// 移除指定绝对路径节点及子节点
/// </summary>
/// <param name="absolutePath">绝对路径</param>
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
/// <summary>
/// 释放资源
/// </summary>
public void Dispose()
{
OnDisposing?.Invoke();
Close();
timer?.Dispose();
nodeWatchers?.Clear();
are?.Dispose();
GC.Collect();
}
/// <summary>
/// 默认的监听器,用于初始化使用
/// </summary>
public class DefaultWatcher : Watcher
{
/// <summary>
/// waithandle同步
/// </summary>
EventWaitHandle ewh;
/// <summary>
/// 辅助类
/// </summary>
ZookeeperHelper zookeeperHelper;
public DefaultWatcher(ZookeeperHelper zookeeperHelper, EventWaitHandle ewh)
{
this.ewh = ewh;
this.zookeeperHelper = zookeeperHelper;
}
/// <summary>
/// 回调
/// </summary>
/// <param name="event">监听事件对象</param>
/// <returns></returns>
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);
}
}
}
/// <summary>
/// 认证类型
/// </summary>
public enum AuthScheme
{
/// <summary>
/// 下面只有一个id,叫anyone,world:anyone代表任何人,zookeeper中对所有人有权限的结点就是属于world:anyone类型的。创建节点的默认权限。有唯一的id是anyone授权的时候的模式为 world:anyone:rwadc 表示所有人都对这个节点有rwadc的权限
/// </summary>
World = 0,
/// <summary>
///不需要id,只要是通过authentication的user都有权限(zookeeper支持通过kerberos来进行authencation, 也支持username/password形式的authentication)
/// </summary>
Auth = 1,
/// <summary>
/// 它对应的id为username:BASE64(SHA1(password)),它需要先通过加密过的username:password形式的authentication。
/// </summary>
Digest = 2,
/// <summary>
///它对应的id为客户机的IP地址,设置的时候可以设置一个ip段,比如ip:192.168.1.0/16。
/// </summary>
Ip = 3,
/// <summary>
/// 在这种scheme情况下,对应的id拥有超级权限,可以做任何事情(cdrwa)
/// </summary>
Super = 4
}
/// <summary>
/// Zookeeper事件数据
/// </summary>
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();
}
/// <summary>
/// 当前连接状态
/// </summary>
public EventState State { get; private set; }
/// <summary>
/// 事件类型
/// </summary>
public EventType Type { get; private set; }
/// <summary>
/// 事件路径
/// </summary>
public string Path { get; private set; }
/// <summary>
/// 连接状态
/// </summary>
public enum EventState
{/// <summary>
/// 超时
/// </summary>
Expired = -112,
/// <summary>
/// 连接已断开
/// </summary>
Disconnected = 0,
/// <summary>
/// 已建立连接
/// </summary>
SyncConnected = 3,
/// <summary>
/// 认证失败
/// </summary>
AuthFailed = 4,
/// <summary>
/// 已建立连接,但是只支持只读模式
/// </summary>
ConnectedReadOnly = 5
}
/// <summary>
/// 时间类型
/// </summary>
public enum EventType
{
/// <summary>
/// 空类型,如:建立连接时
/// </summary>
None = -1,
/// <summary>
/// 节点创建时
/// </summary>
NodeCreated = 1,
/// <summary>
/// 节点删除时
/// </summary>
NodeDeleted = 2,
/// <summary>
/// 节点数据改变时
/// </summary>
NodeDataChanged = 3,
/// <summary>
/// 节点增加子节点时
/// </summary>
NodeChildrenChanged = 4
}
}
/// <summary>
/// 监控对象
/// </summary>
public class NodeWatcher
{
/// <summary>
/// 节点创建时调用事件
/// </summary>
public event WatcherEvent NodeCreated;
/// <summary>
/// 节点删除时调用事件
/// </summary>
public event WatcherEvent NodeDeleted;
/// <summary>
/// 节点数据改变时调用事件
/// </summary>
public event WatcherEvent NodeDataChanged;
/// <summary>
/// 节点增加子节点时调用事件
/// </summary>
public event WatcherEvent NodeChildrenChanged;
/// <summary>
/// 不区分类型,所有的类型都会调用
/// </summary>
public event WatcherEvent AllTypeChanged;
/// <summary>
/// 触发,执行事件
/// </summary>
/// <param name="event"></param>
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 { }
}
}
/// <summary>
/// 监控事件委托
/// </summary>
/// <param name="event"></param>
public delegate void WatcherEvent(ZookeeperEvent @event);
/// <summary>
/// Zookeeper默认日志记录
/// </summary>
public class ZookeeperLogger : ILogConsumer
{
/// <summary>
/// 是否记录日志到文件
/// </summary>
public bool LogToFile { get; set; } = false;
/// <summary>
/// 是否记录堆栈信息
/// </summary>
public bool LogToTrace { get; set; } = true;
/// <summary>
/// 日志级别
/// </summary>
public TraceLevel LogLevel { get; set; } = TraceLevel.Warning;
/// <summary>
/// 日志记录
/// </summary>
/// <param name="severity"></param>
/// <param name="className"></param>
/// <param name="message"></param>
/// <param name="exception"></param>
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);
}
}
}
ZookeeperHelper
简单的使用例子:
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 = zookeeperHelper.GetData("/mynode");
Console.WriteLine("完成读取节点:" + value);
}
//设置节点数据
{
zookeeperHelper.SetData("/mynode", "hello world again");
Console.WriteLine("设置节点数据");
}
//重新获取节点数据
{
var value = zookeeperHelper.GetData("/mynode");
Console.WriteLine("重新获取节点数据:" + value);
}
//移除节点
{
zookeeperHelper.Delete("/mynode");
Console.WriteLine("移除节点");
}
Console.WriteLine("完成");
Console.ReadKey();
}
}
}
执行结果:
手机扫一扫
移动阅读更方便
你可能感兴趣的文章