C#如何連接使用Zookeeper
更新時(shí)間：2021年06月24日 10:30:45 作者：沒有星星的夏季
Zookeeper作為分布式的服務(wù)框架，雖然是java寫的，但是強(qiáng)大的C#也可以連接使用。而現(xiàn)在主要有兩個(gè)插件可供使用，分別是ZooKeeperNetEx和Zookeeper.Net，個(gè)人推薦使用ZooKeeperNetEx做開發(fā)，本文也已介紹ZooKeeperNetEx為主
Zookeeper作為分布式的服務(wù)框架，雖然是java寫的，但是強(qiáng)大的C#也可以連接使用。
C#要連接使用Zookeeper，需要借助第三方插件，而現(xiàn)在主要有兩個(gè)插件可供使用，分別是ZooKeeperNetEx和Zookeeper.Net
Zookeeper.Net好像是是Apache官方提供的，但是5年沒更新了，也就是說他依賴于.net framework，因此無法在.net core項(xiàng)目中使用
ZooKeeperNetEx是從java改過來的，因此里面的一些習(xí)慣是java風(fēng)格的，但是好像有人在提供更新維護(hù)，支持最新的Zookeeper特性，而且擺脫了對(duì).net framework的依賴，所以個(gè)人推薦使用ZooKeeperNetEx做開發(fā)，本文也已介紹ZooKeeperNetEx為主
新建一個(gè)控制臺(tái)項(xiàng)目，在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格式，多個(gè)地址之間使用逗號(hào)（,）隔開
string connectionString = "192.168.209.133:2181,192.168.209.133:2181,192.168.209.133:2181";
//會(huì)話超時(shí)時(shí)間,單位毫秒
int sessionTimeOut = 10000;
//異步監(jiān)聽
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;
}
//創(chuàng)建znode節(jié)點(diǎn)
{
var data = Encoding.UTF8.GetBytes("hello world");
List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;//創(chuàng)建節(jié)點(diǎn)時(shí)的acl權(quán)限，也可以使用下面的自定義權(quán)限
//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("完成創(chuàng)建節(jié)點(diǎn)");
}
//節(jié)點(diǎn)是否存在
{
var exists = zooKeeper.existsAsync("/mynode", new MyWatcher("ExistsWatcher")).GetAwaiter().GetResult();
Console.WriteLine("節(jié)點(diǎn)是否存在：" + exists);
}
//獲取節(jié)點(diǎn)數(shù)據(jù)
{
var dataResult = zooKeeper.getDataAsync("/mynode", new MyWatcher("GetWatcher")).GetAwaiter().GetResult();
var value = Encoding.UTF8.GetString(dataResult.Data);
Console.WriteLine("完成讀取節(jié)點(diǎn)：" + value);
}
//設(shè)置節(jié)點(diǎn)數(shù)據(jù)
{
var data = Encoding.UTF8.GetBytes("hello world again");
zooKeeper.setDataAsync("/mynode", data);
Console.WriteLine("設(shè)置節(jié)點(diǎn)數(shù)據(jù)");
}
//重新獲取節(jié)點(diǎn)數(shù)據(jù)
{
var dataResult = zooKeeper.getDataAsync("/mynode", new MyWatcher("GetWatcher")).GetAwaiter().GetResult();
var value = Encoding.UTF8.GetString(dataResult.Data);
Console.WriteLine("重新獲取節(jié)點(diǎn)數(shù)據(jù)：" + value);
}
//移除節(jié)點(diǎn)
{
zooKeeper.deleteAsync("/mynode").Wait();
Console.WriteLine("移除節(jié)點(diǎn)");
}
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);
}
}
}
運(yùn)行后顯示結(jié)果：
這個(gè)簡(jiǎn)單的例子是使用ZooKeeperNetEx操作的簡(jiǎn)單例子，下面具體介紹
ZooKeeperNetEx連接Zookeeper只需要實(shí)例化ZooKeeper對(duì)象即可
//Zookeeper連接字符串，采用host:port格式，多個(gè)地址之間使用逗號(hào)（,）隔開
string connectionString = "192.168.209.133:2181,192.168.209.133:2181,192.168.209.133:2181";
//會(huì)話超時(shí)時(shí)間,單位毫秒
int sessionTimeOut = 10000;
//異步監(jiān)聽
var watcher = new MyWatcher("ConnectWatcher");
//連接
ZooKeeper zooKeeper = new ZooKeeper(connectionString, sessionTimeOut, watcher);
實(shí)例化過程中至少需要三個(gè)參數(shù)
連接字符串(connectstring)：host:port形式，多個(gè)地址之間使用英文逗號(hào)隔開
會(huì)話超時(shí)時(shí)間(sessionTimeout)：當(dāng)會(huì)話中，Zookeeper超過此時(shí)間未響應(yīng)，則表示會(huì)話超時(shí)
監(jiān)聽器(watcher)：這個(gè)連接過程中可以注冊(cè)一個(gè)監(jiān)聽器，當(dāng)連接過程中出現(xiàn)狀態(tài)改變時(shí)，會(huì)通知到監(jiān)聽器
ZooKeeper對(duì)象實(shí)例化過程中會(huì)異步的去連接Zookeeper，所以例子中才有一個(gè)while循環(huán)來判斷狀態(tài)
Thread.Sleep(1000);//停一秒，等待連接完成
while (zooKeeper.getState() == ZooKeeper.States.CONNECTING)
{
Console.WriteLine("等待連接完成...");
Thread.Sleep(1000);
}
而Zookeeper的連接狀態(tài)有6種：
//ZooKeeper.States的枚舉 CONNECTING = 0, //連接中 CONNECTED = 1, //已連接 CONNECTEDREADONLY = 2, //已連接，但是只能只讀訪問 CLOSED = 3, //已關(guān)閉連接 AUTH_FAILED = 4, //認(rèn)證失敗 NOT_CONNECTED = 5 //未連接
當(dāng)應(yīng)用連接到Zookeeper時(shí)，一般都是讀取數(shù)據(jù)，所以主需要只讀連接就可以滿足的，不過具體還是要看需求。
當(dāng)在指定的會(huì)話時(shí)間內(nèi)未成功連接時(shí)，則會(huì)導(dǎo)致連接超時(shí)，因?yàn)檫@個(gè)過程是異步的，所以需要一個(gè)監(jiān)聽器來接收。
監(jiān)聽器其實(shí)是org.apache.zookeeper.Watcher的一個(gè)子類，這個(gè)需要開發(fā)者去繼承實(shí)現(xiàn)它的process方法，比如上面的例子中我們就簡(jiǎn)單的實(shí)現(xiàn)
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);
}
}
這里僅僅只是簡(jiǎn)單的輸出節(jié)點(diǎn)路徑、監(jiān)聽事件響應(yīng)狀態(tài)和監(jiān)聽事件類型
//監(jiān)聽事件響應(yīng)狀態(tài)，Watcher.Event.KeeperState的枚舉 Expired = -112, //連接超時(shí) Disconnected = 0, //連接斷開 SyncConnected = 3, //已同步連接 AuthFailed = 4, //認(rèn)證失敗 ConnectedReadOnly = 5 //只讀連接 //監(jiān)聽事件類型，Watcher.Event.EventType的枚舉 None = -1, //非節(jié)點(diǎn)操作事件 NodeCreated = 1, //創(chuàng)建節(jié)點(diǎn)事件 NodeDeleted = 2, //刪除節(jié)點(diǎn)事件 NodeDataChanged = 3, //節(jié)點(diǎn)數(shù)據(jù)改變 NodeChildrenChanged = 4 //子節(jié)點(diǎn)發(fā)生改變
為什么要有監(jiān)聽器？監(jiān)聽器就類似一個(gè)回調(diào)，當(dāng)發(fā)生某個(gè)事件時(shí)，我們的應(yīng)用可能需要進(jìn)行相應(yīng)的處理，如當(dāng)連接斷開時(shí)，由于監(jiān)聽器的存在，我們可以讓我們的應(yīng)用程序重新與Zookeeper建立連接。
ZooKeeperNetEx創(chuàng)建znode節(jié)點(diǎn)使用的是createAsync異步方法，傳入4個(gè)參數(shù)，分別是
節(jié)點(diǎn)路徑(path):：創(chuàng)建的節(jié)點(diǎn)路徑
節(jié)點(diǎn)數(shù)據(jù)(data)：節(jié)點(diǎn)數(shù)據(jù)，它是一個(gè)字節(jié)數(shù)組，可以通過編碼將字符串轉(zhuǎn)化為字符數(shù)組
ACL權(quán)限(acl)：ACL權(quán)限，可以使用已定義好的，也可以使用自定義，如：
//已經(jīng)定義好的，ZooDefs.Ids的枚舉 OPEN_ACL_UNSAFE：完全開放 CREATOR_ALL_ACL：創(chuàng)建該znode的連接擁有所有權(quán)限 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"))
};
節(jié)點(diǎn)類型(createMode)：節(jié)點(diǎn)類型有4種，分別是CreateMode類的4個(gè)靜態(tài)字段
PERSISTENT：持久化節(jié)點(diǎn)
PERSISTENT_SEQUENTIAL：持久化有序節(jié)點(diǎn)
EPHEMERAL：臨時(shí)節(jié)點(diǎn)（連接斷開自動(dòng)刪除）
EPHEMERAL_SEQUENTIAL：臨時(shí)有序節(jié)點(diǎn)（連接斷開自動(dòng)刪除）
createAsync異步方法會(huì)返回實(shí)際創(chuàng)建的znode路徑，貌似沒什么用（在創(chuàng)建順序節(jié)點(diǎn)時(shí)會(huì)用到，當(dāng)一個(gè)新的znode被創(chuàng)建為一個(gè)順序節(jié)點(diǎn)時(shí)，ZooKeeper通過將10位的序列號(hào)附加到原始名稱來設(shè)置znode的路徑）
上面這個(gè)是ZooKeeperNetEx創(chuàng)建znode節(jié)點(diǎn)的方法，而對(duì)znode的其他操作的參數(shù)就很簡(jiǎn)單了，這里就不在重述，需要具體操作才能理解，一個(gè)簡(jiǎn)單的介紹如下：
//刪除znode節(jié)點(diǎn)
public Task deleteAsync(string path, int version = -1);
//指定的znode節(jié)點(diǎn)是否存在
public Task<Stat> existsAsync(string path, Watcher watcher);
public Task<Stat> existsAsync(string path, bool watch = false);
//獲取znode節(jié)點(diǎn)數(shù)據(jù)
public Task<DataResult> getDataAsync(string path, bool watch = false);
public Task<DataResult> getDataAsync(string path, Watcher watcher);
//設(shè)置指定znode節(jié)點(diǎn)的數(shù)據(jù)
public Task<Stat> setDataAsync(string path, byte[] data, int version = -1);
//獲取指定znode節(jié)點(diǎn)的子節(jié)點(diǎn)，注意，監(jiān)聽器是注冊(cè)給當(dāng)前節(jié)點(diǎn)的，而非子節(jié)點(diǎn)
public Task<ChildrenResult> getChildrenAsync(string path, Watcher watcher);
public Task<ChildrenResult> getChildrenAsync(string path, bool watch = false);
可以比較一下上一節(jié)介紹的zkCli對(duì)znode節(jié)點(diǎn)的操作就很容易理解了。
另外，需要注意的是，existsAsync方法、getDataAsync方法和getChildrenAsync方法可以在指定的znode注冊(cè)一個(gè)監(jiān)聽器，setDataAsync方法卻沒有這個(gè)注冊(cè)功能，這個(gè)是因?yàn)閆ookeeper注冊(cè)的監(jiān)聽器只會(huì)響應(yīng)一次，當(dāng)需要再次響應(yīng)時(shí)，需要重新注冊(cè)，這時(shí)就可以調(diào)用existsAsync方法或者getDataAsync方法或者getChildrenAsync方法進(jìn)行重新注冊(cè)了！
上一節(jié)說到ACL權(quán)限不僅可以在創(chuàng)建是給予，在創(chuàng)建后也可以修改，ZookeeperNetEx操作znode的ACL權(quán)限使用的方法如下：
//獲取ACL權(quán)限
public Task<ACLResult> getACLAsync(string path);
//設(shè)置ACL權(quán)限
public Task<Stat> setACLAsync(string path, List<ACL> acl, int aclVersion = -1);
說到ACL，自然就會(huì)認(rèn)證存在，ZookeeperNetEx添加認(rèn)證使用的是addAuthInfo方法
public void addAuthInfo(string scheme, byte[] auth);
其中scheme就是我們上一節(jié)介紹的那幾種：
world：默認(rèn)模式，所有客戶端都擁有指定的權(quán)限。world下只有一個(gè)id選項(xiàng)，就是anyone，通常組合寫法為world:anyone:[permissons]； auth：只有經(jīng)過認(rèn)證的用戶才擁有指定的權(quán)限。通常組合寫法為auth:user:password:[permissons]，使用這種模式時(shí)，你需要先進(jìn)行登錄，之后采用auth模式設(shè)置權(quán)限時(shí)，user和password都將使用登錄的用戶名和密碼；比如： digest：只有經(jīng)過認(rèn)證的用戶才擁有指定的權(quán)限。通常組合寫法為digest:user:BASE64(SHA1(password)):[permissons]，這種形式下的密碼必須通過SHA1和BASE64進(jìn)行雙重加密； ip：限制只有特定IP的客戶端才擁有指定的權(quán)限。通常組成寫法為ip:182.168.0.168:[permissions]； super：代表超級(jí)管理員，擁有所有的權(quán)限，需要修改Zookeeper啟動(dòng)腳本進(jìn)行配置。
auth是認(rèn)證數(shù)據(jù)，如果沒有則可以是空的字節(jié)數(shù)組，如：
//world模式認(rèn)證
zk.addAuthInfo("world",new byte[0]);
//auth模式認(rèn)證
byte[] auth=Encoding.UTF8.GetBytes("id:pass")
zk.addAuthInfo("auth",new byte[0]);
//digest模式認(rèn)證
byte[] auth=Encoding.UTF8.GetBytes("加密后的字符串")
zk.addAuthInfo("digest",new byte[0]);
ZookeeperNetEx關(guān)閉會(huì)話使用的是closeAsync方法，調(diào)用這個(gè)方法之后，當(dāng)前連接對(duì)象ZooKeeper就不能再訪問了
public Task closeAsync();
其他常用方法就不介紹了，一般時(shí)候基本上也用不上。
簡(jiǎn)單封裝
真實(shí)項(xiàng)目中，我們連接Zookeeper多數(shù)只是為了創(chuàng)建znode節(jié)點(diǎn)，讀取數(shù)據(jù)等等操作，一般不會(huì)去設(shè)置ACL等權(quán)限，甚至連認(rèn)證都可能不會(huì)用到，為了更好使用ZookeeperNetEx，我做了一層簡(jiǎn)單的封裝，用以滿足常見的CRUD操作，同時(shí)也讓它更符合我們.net開發(fā)的一些習(xí)慣。
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訪問對(duì)象
/// </summary>
ZooKeeper zookeeper;
/// <summary>
/// Zookeeper集群地址
/// </summary>
string[] address;
/// <summary>
/// 路徑監(jiān)控節(jié)點(diǎn)列表
/// </summary>
ConcurrentDictionary<string, NodeWatcher> nodeWatchers = new ConcurrentDictionary<string, NodeWatcher>();
/// <summary>
/// 節(jié)點(diǎn)的默認(rèn)權(quán)限
/// </summary>
List<ACL> defaultACL = ZooDefs.Ids.OPEN_ACL_UNSAFE;
/// <summary>
/// 默認(rèn)的監(jiān)聽器
/// </summary>
DefaultWatcher defaultWatcher;
/// <summary>
/// 監(jiān)控定時(shí)器
/// </summary>
System.Timers.Timer timer;
/// <summary>
/// 同步鎖
/// </summary>
AutoResetEvent are = new AutoResetEvent(false);
/// <summary>
/// 是否正常關(guān)閉
/// </summary>
bool isClose = false;
/// <summary>
/// 回話超時(shí)時(shí)間
/// </summary>
public int SessionTimeout { get; set; } = 10000;
/// <summary>
/// 當(dāng)前路徑
/// </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是否有寫的權(quán)限
/// </summary>
public bool CanWrite { get { return zookeeper != null && zookeeper.getState() == ZooKeeper.States.CONNECTED; } }
/// <summary>
/// 數(shù)據(jù)編碼
/// </summary>
public Encoding Encoding { get; set; } = Encoding.Default;
/// <summary>
/// 釋放時(shí)發(fā)生
/// </summary>
public event Action OnDisposing;
/// <summary>
/// 在重新連接時(shí)發(fā)生
/// </summary>
public event Action OnConnected;
/// <summary>
/// 構(gòu)造函數(shù)
/// </summary>
/// <param name="address">集群地址(host:prot)</param>
public ZookeeperHelper(params string[] address) : this(address, "")
{
}
/// <summary>
/// 構(gòu)造函數(shù)
/// </summary>
/// <param name="address">集群地址(host:prot)</param>
/// <param name="root">初始化根路經(jīng)</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的日志設(shè)置
/// </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>
/// 定時(shí)器
/// </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))//每次只能一個(gè)線程進(jìn)去
{
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)//會(huì)話未超時(shí)，表示成功連接
{
//掛載監(jiān)聽器
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>
/// 檢查是否具有寫的權(quán)限
/// </summary>
private void CheckWriten()
{
if (!CanWrite)
{
throw new Exception("this connection is in readonly mode");
}
}
/// <summary>
/// 連接數(shù)據(jù)成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 == ".")//當(dāng)前路徑
{
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>
/// 獲取絕對(duì)路徑
/// </summary>
/// <param name="path">路徑</param>
/// <param name="isAbsolute">路徑是否是絕對(duì)路徑</param>
/// <returns>絕對(duì)路徑</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>
/// 關(guān)閉連接
/// </summary>
public void Close()
{
isClose = true;
if (Connected)
{
zookeeper.closeAsync().Wait();
}
}
/// <summary>
/// 監(jiān)控回調(diào)
/// </summary>
/// <param name="event">回調(diào)事件</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);//重新監(jiān)控
}
}
}
}
/// <summary>
/// 修改當(dāng)前目錄地址
/// </summary>
/// <param name="path"></param>
public void ChangeDirectory(string path)
{
this.CurrentPath = Combine(path);
}
/// <summary>
/// 切換到相對(duì)目錄下
/// </summary>
/// <param name="path"></param>
public void Goto(string path)
{
this.CurrentPath = Combine(this.CurrentPath, path);
}
/// <summary>
/// 使用認(rèn)證
/// </summary>
/// <param name="scheme">認(rèn)證類型</param>
/// <param name="auth">認(rèn)證數(shù)據(jù)</param>
public void Authorize(AuthScheme scheme, string auth = "")
{
CheckConnection();
zookeeper.addAuthInfo(scheme.ToString().ToLower(), Encoding.GetBytes(auth));
}
#region 監(jiān)聽與取消
/// <summary>
/// 對(duì)當(dāng)前路徑添加監(jiān)控
/// </summary>
/// <param name="delegate">監(jiān)控</param>
/// <returns>異步，true表示成功，false表示失敗</returns>
public async Task<bool> WatchAsync(WatcherEvent @delegate)
{
return await WatchAsync(CurrentPath, @delegate, true);
}
/// <summary>
/// 對(duì)當(dāng)前路徑添加監(jiān)控
/// </summary>
/// <param name="watcher">監(jiān)控</param>
/// <returns>異步，true表示成功，false表示失敗</returns>
public async Task<bool> WatchAsync(NodeWatcher watcher)
{
return await WatchAsync(CurrentPath, watcher, true);
}
/// <summary>
/// 對(duì)指定路徑添加監(jiān)控
/// </summary>
/// <param name="path">節(jié)點(diǎn)路徑</param>
/// <param name="delegate">監(jiān)控</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</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>
/// 對(duì)指定路徑添加監(jiān)控
/// </summary>
/// <param name="path">節(jié)點(diǎn)路徑</param>
/// <param name="watcher">監(jiān)控</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</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>
/// 監(jiān)控多個(gè)路徑，但是不包括子路徑
/// </summary>
/// <param name="paths">節(jié)點(diǎn)路徑</param>
/// <param name="delegate">監(jiān)控</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</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>
/// 監(jiān)控多個(gè)路徑，但是不包括子路徑
/// </summary>
/// <param name="paths">節(jié)點(diǎn)路徑</param>
/// <param name="watcher">監(jiān)控</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</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>
/// 監(jiān)控當(dāng)前路徑，包括子路徑
/// </summary>
/// <param name="delegate">監(jiān)控</param>
/// <returns>異步，true表示成功，false表示失敗</returns>
public async Task<bool> WatchAllAsync(WatcherEvent @delegate)
{
return await WatchAllAsync(CurrentPath, @delegate, true);
}
/// <summary>
/// 監(jiān)控當(dāng)前路徑，包括子路徑
/// </summary>
/// <param name="watcher">監(jiān)控</param>
/// <returns>異步，true表示成功，false表示失敗</returns>
public async Task<bool> WatchAllAsync(NodeWatcher watcher)
{
return await WatchAllAsync(CurrentPath, watcher, true);
}
/// <summary>
/// 監(jiān)控指定路徑，包括子路徑
/// </summary>
/// <param name="path">節(jié)點(diǎn)路徑</param>
/// <param name="delegate">監(jiān)控</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</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>
/// 監(jiān)控指定路徑，包括子路徑
/// </summary>
/// <param name="path">節(jié)點(diǎn)路徑</param>
/// <param name="watcher">監(jiān)控</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</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>
/// 監(jiān)控所有路徑，包括子路徑
/// </summary>
/// <param name="paths">節(jié)點(diǎn)路徑</param>
/// <param name="delegate">監(jiān)控</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</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>
/// 監(jiān)控所有路徑，包括子路徑
/// </summary>
/// <param name="paths">節(jié)點(diǎn)路徑</param>
/// <param name="watcher">監(jiān)控</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</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>
/// 取消多個(gè)指定路徑上的監(jiān)控
/// </summary>
/// <param name="path">節(jié)點(diǎn)路徑</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</param>
/// <returns>異步</returns>
public async Task CancelAsync(string path, bool isAbsolutePath = true)
{
await CancelAsync(new string[] { path }, isAbsolutePath);
}
/// <summary>
/// 取消多個(gè)指定路徑上的監(jiān)控
/// </summary>
/// <param name="path">節(jié)點(diǎn)路徑</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</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>
/// 獲取指定路徑上的監(jiān)控
/// </summary>
/// <param name="path">節(jié)點(diǎn)路徑</param>
/// <param name="isAbsolutePath">是否絕對(duì)路徑</param>
/// <returns>存在則返回監(jiān)控對(duì)象，否則返回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 基本數(shù)據(jù)操作
/// <summary>
/// 當(dāng)前節(jié)點(diǎn)是否存在
/// </summary>
/// <returns>存在返回true，否則返回false</returns>
public bool Exists()
{
return ExistsAsync().GetAwaiter().GetResult();
}
/// <summary>
/// 指定節(jié)點(diǎn)是否存在（相對(duì)當(dāng)前節(jié)點(diǎn)）
/// </summary>
/// <param name="path">節(jié)點(diǎn)路徑</param>
/// <returns>存在返回true，否則返回false</returns>
public bool Exists(string path)
{
return ExistsAsync(path).GetAwaiter().GetResult();
}
/// <summary>
/// 指定節(jié)點(diǎn)是否存在
/// </summary>
/// <param name="absolutePath">絕對(duì)路徑</param>
/// <returns>存在返回true，否則返回false</returns>
public bool ExistsByAbsolutePath(string absolutePath)
{
return ExistsByAbsolutePathAsync(absolutePath).GetAwaiter().GetResult();
}
/// <summary>
/// 當(dāng)前節(jié)點(diǎn)是否存在
/// </summary>
/// <returns>異步，存在返回true，否則返回false</returns>
public async Task<bool> ExistsAsync()
{
return await ExistsByAbsolutePathAsync(CurrentPath);
}
/// <summary>
/// 指定節(jié)點(diǎn)是否存在（相對(duì)當(dāng)前節(jié)點(diǎn)）
/// </summary>
/// <param name="path">節(jié)點(diǎn)路徑</param>
/// <returns>異步，存在返回true，否則返回false</returns>
public async Task<bool> ExistsAsync(string path)
{
path = Combine(CurrentPath, path);
return await ExistsByAbsolutePathAsync(path);
}
/// <summary>
/// 指定節(jié)點(diǎn)是否存在
/// </summary>
/// <param name="absolutePath">絕對(duì)路徑</param>
/// <returns>異步，存在返回true，否則返回false</returns>
public async Task<bool> ExistsByAbsolutePathAsync(string absolutePath)
{
absolutePath = Combine(absolutePath);
return await zookeeper.existsAsync(absolutePath, false) != null;
}
/// <summary>
/// 添加或者修改當(dāng)前路徑上的數(shù)據(jù)
/// </summary>
/// <param name="value">數(shù)據(jù)</param>
/// <param name="persistent">是否持久節(jié)點(diǎn)</param>
/// <param name="sequential">是否順序節(jié)點(diǎn)</param>
/// <returns>znode節(jié)點(diǎn)名，不包含父節(jié)點(diǎn)路徑</returns>
public string SetData(string value, bool persistent = false, bool sequential = false)
{
return SetDataAsync(value, persistent, sequential).GetAwaiter().GetResult();
}
/// <summary>
/// 添加或者修改指定相對(duì)路徑上的數(shù)據(jù)
/// </summary>
/// <param name="path">相對(duì)路徑</param>
/// <param name="value">數(shù)據(jù)</param>
/// <param name="persistent">是否持久節(jié)點(diǎn)</param>
/// <param name="sequential">是否順序節(jié)點(diǎn)</param>
/// <returns>znode節(jié)點(diǎn)名，不包含父節(jié)點(diǎn)路徑</returns>
public string SetData(string path, string value, bool persistent = false, bool sequential = false)
{
return SetDataAsync(path, value, persistent, sequential).GetAwaiter().GetResult();
}
/// <summary>
/// 添加或者修改指定絕對(duì)路徑上的數(shù)據(jù)
/// </summary>
/// <param name="absolutePath">絕對(duì)路徑</param>
/// <param name="value">數(shù)據(jù)</param>
/// <param name="persistent">是否持久節(jié)點(diǎn)</param>
/// <param name="sequential">是否順序節(jié)點(diǎn)</param>
/// <returns>znode節(jié)點(diǎn)名，不包含父節(jié)點(diǎn)路徑</returns>
public string SetDataByAbsolutePath(string absolutePath, string value, bool persistent = false, bool sequential = false)
{
return SetDataByAbsolutePathAsync(absolutePath, value, persistent, sequential).GetAwaiter().GetResult();
}
/// <summary>
/// 添加或者修改當(dāng)前路徑上的數(shù)據(jù)
/// </summary>
/// <param name="value">數(shù)據(jù)</param>
/// <param name="persistent">是否持久節(jié)點(diǎn)</param>
/// <param name="sequential">是否順序節(jié)點(diǎn)</param>
/// <returns>znode節(jié)點(diǎn)名，不包含父節(jié)點(diǎn)路徑</returns>
public async Task<string> SetDataAsync(string value, bool persistent = false, bool sequential = false)
{
return await SetDataByAbsolutePathAsync(CurrentPath, value, persistent, sequential);
}
/// <summary>
/// 添加或者修改指定相對(duì)路徑上的數(shù)據(jù)
/// </summary>
/// <param name="path">相對(duì)路徑</param>
/// <param name="value">數(shù)據(jù)</param>
/// <param name="persistent">是否持久節(jié)點(diǎn)</param>
/// <param name="sequential">是否順序節(jié)點(diǎn)</param>
/// <returns>znode節(jié)點(diǎn)名，不包含父節(jié)點(diǎn)路徑</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>
/// 添加或者修改指定絕對(duì)路徑上的數(shù)據(jù)
/// </summary>
/// <param name="absolutePath">絕對(duì)路徑</param>
/// <param name="value">數(shù)據(jù)</param>
/// <param name="persistent">是否持久節(jié)點(diǎn)</param>
/// <param name="sequential">是否順序節(jié)點(diǎn)</param>
/// <returns>znode節(jié)點(diǎn)名，不包含父節(jié)點(diǎn)路徑</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>
/// 獲取指定相對(duì)路徑上的數(shù)據(jù)
/// </summary>
/// <param name="path">相對(duì)路徑</param>
/// <returns>相對(duì)路徑上的數(shù)據(jù)</returns>
public string GetData(string path)
{
return GetDataAsync(path).GetAwaiter().GetResult();
}
/// <summary>
/// 獲取指定絕對(duì)路徑上的數(shù)據(jù)
/// </summary>
/// <param name="absolutePath">絕對(duì)路徑</param>
/// <returns>相對(duì)路徑上的數(shù)據(jù)</returns>
public string GetDataByAbsolutePath(string absolutePath)
{
return GetDataByAbsolutePathAsync(absolutePath).GetAwaiter().GetResult();
}
/// <summary>
/// 獲取指定相對(duì)路徑上的數(shù)據(jù)
/// </summary>
/// <param name="path">相對(duì)路徑</param>
/// <returns>相對(duì)路徑上的數(shù)據(jù)</returns>
public async Task<string> GetDataAsync(string path)
{
path = Combine(CurrentPath, path);
return await GetDataByAbsolutePathAsync(path);
}
/// <summary>
/// 獲取指定絕對(duì)路徑上的數(shù)據(jù)
/// </summary>
/// <param name="absolutePath">絕對(duì)路徑</param>
/// <returns>絕對(duì)路徑上的數(shù)據(jù)</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>
/// 獲取指定節(jié)點(diǎn)及其字節(jié)點(diǎn)的所有值，使用路徑做鍵返回字典型
/// </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>
/// 獲取子節(jié)點(diǎn)
/// </summary>
/// <param name="path">相對(duì)路徑</param>
/// <param name="order">是否按時(shí)間排序</param>
/// <returns>子節(jié)點(diǎn)數(shù)組（節(jié)點(diǎn)路徑不含父節(jié)點(diǎn)路徑）</returns>
public async Task<string[]> GetChildrenAsync(string path, bool order = false)
{
path = Combine(CurrentPath, path);
return await GetChildrenByAbsolutePathAsync(path, order);
}
/// <summary>
/// 獲取指定路徑絕對(duì)路徑下的子節(jié)點(diǎn)
/// </summary>
/// <param name="absolutePath">絕對(duì)路徑</param>
/// <param name="order">是否按時(shí)間排序</param>
/// <returns>子節(jié)點(diǎn)數(shù)組（節(jié)點(diǎn)路徑不含父節(jié)點(diǎn)路徑）</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>
/// 移除當(dāng)前路徑節(jié)點(diǎn)
/// </summary>
public void Delete()
{
DeleteAsync().Wait();
}
/// <summary>
/// 移除相對(duì)當(dāng)前的指定路徑節(jié)點(diǎn)及子節(jié)點(diǎn)
/// </summary>
/// <param name="path">相對(duì)路徑</param>
public void Delete(string path)
{
DeleteAsync(path).Wait();
}
/// <summary>
/// 移除指定絕對(duì)路徑節(jié)點(diǎn)及子節(jié)點(diǎn)
/// </summary>
/// <param name="absolutePath">絕對(duì)路徑</param>
public void DeleteByAbsolutePath(string absolutePath)
{
DeleteByAbsolutePathAsync(absolutePath).Wait();
}
/// <summary>
/// 移除當(dāng)前路徑節(jié)點(diǎn)
/// </summary>
public async Task DeleteAsync()
{
await DeleteByAbsolutePathAsync(CurrentPath);
}
/// <summary>
/// 移除相對(duì)當(dāng)前的指定路徑節(jié)點(diǎn)及子節(jié)點(diǎn)
/// </summary>
/// <param name="path">相對(duì)路徑</param>
public async Task DeleteAsync(string path)
{
path = Combine(CurrentPath, path);
await DeleteByAbsolutePathAsync(path);
}
/// <summary>
/// 移除指定絕對(duì)路徑節(jié)點(diǎn)及子節(jié)點(diǎn)
/// </summary>
/// <param name="absolutePath">絕對(duì)路徑</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>
/// 默認(rèn)的監(jiān)聽器，用于初始化使用
/// </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>
/// 回調(diào)
/// </summary>
/// <param name="event">監(jiān)聽事件對(duì)象</param>
/// <returns></returns>
public override Task process(WatchedEvent @event)
{
var state = @event.getState();
if (state == Event.KeeperState.ConnectedReadOnly || state == Event.KeeperState.SyncConnected)//連接時(shí)
{
ewh.Set();
}
else if ((state == Event.KeeperState.Expired) && !zookeeperHelper.isClose)//回話過期重新建立連接
{
zookeeperHelper.timer.Enabled = true;
}
return Task.FromResult(1);
}
}
}
/// <summary>
/// 認(rèn)證類型
/// </summary>
public enum AuthScheme
{
/// <summary>
/// 下面只有一個(gè)id，叫anyone，world:anyone代表任何人，zookeeper中對(duì)所有人有權(quán)限的結(jié)點(diǎn)就是屬于world:anyone類型的。創(chuàng)建節(jié)點(diǎn)的默認(rèn)權(quán)限。有唯一的id是anyone授權(quán)的時(shí)候的模式為 world:anyone:rwadc 表示所有人都對(duì)這個(gè)節(jié)點(diǎn)有rwadc的權(quán)限
/// </summary>
World = 0,
/// <summary>
///不需要id,只要是通過authentication的user都有權(quán)限（zookeeper支持通過kerberos來進(jìn)行authencation, 也支持username/password形式的authentication)
/// </summary>
Auth = 1,
/// <summary>
/// 它對(duì)應(yīng)的id為username:BASE64(SHA1(password))，它需要先通過加密過的username:password形式的authentication。
/// </summary>
Digest = 2,
/// <summary>
///它對(duì)應(yīng)的id為客戶機(jī)的IP地址，設(shè)置的時(shí)候可以設(shè)置一個(gè)ip段，比如ip:192.168.1.0/16。
/// </summary>
Ip = 3,
/// <summary>
/// 在這種scheme情況下，對(duì)應(yīng)的id擁有超級(jí)權(quán)限，可以做任何事情(cdrwa）
/// </summary>
Super = 4
}
/// <summary>
/// Zookeeper事件數(shù)據(jù)
/// </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>
/// 當(dāng)前連接狀態(tài)
/// </summary>
public EventState State { get; private set; }
/// <summary>
/// 事件類型
/// </summary>
public EventType Type { get; private set; }
/// <summary>
/// 事件路徑
/// </summary>
public string Path { get; private set; }
/// <summary>
/// 連接狀態(tài)
/// </summary>
public enum EventState
{/// <summary>
/// 超時(shí)
/// </summary>
Expired = -112,
/// <summary>
/// 連接已斷開
/// </summary>
Disconnected = 0,
/// <summary>
/// 已建立連接
/// </summary>
SyncConnected = 3,
/// <summary>
/// 認(rèn)證失敗
/// </summary>
AuthFailed = 4,
/// <summary>
/// 已建立連接，但是只支持只讀模式
/// </summary>
ConnectedReadOnly = 5
}
/// <summary>
/// 時(shí)間類型
/// </summary>
public enum EventType
{
/// <summary>
/// 空類型，如：建立連接時(shí)
/// </summary>
None = -1,
/// <summary>
/// 節(jié)點(diǎn)創(chuàng)建時(shí)
/// </summary>
NodeCreated = 1,
/// <summary>
/// 節(jié)點(diǎn)刪除時(shí)
/// </summary>
NodeDeleted = 2,
/// <summary>
/// 節(jié)點(diǎn)數(shù)據(jù)改變時(shí)
/// </summary>
NodeDataChanged = 3,
/// <summary>
/// 節(jié)點(diǎn)增加子節(jié)點(diǎn)時(shí)
/// </summary>
NodeChildrenChanged = 4
}
}
/// <summary>
/// 監(jiān)控對(duì)象
/// </summary>
public class NodeWatcher
{
/// <summary>
/// 節(jié)點(diǎn)創(chuàng)建時(shí)調(diào)用事件
/// </summary>
public event WatcherEvent NodeCreated;
/// <summary>
/// 節(jié)點(diǎn)刪除時(shí)調(diào)用事件
/// </summary>
public event WatcherEvent NodeDeleted;
/// <summary>
/// 節(jié)點(diǎn)數(shù)據(jù)改變時(shí)調(diào)用事件
/// </summary>
public event WatcherEvent NodeDataChanged;
/// <summary>
/// 節(jié)點(diǎn)增加子節(jié)點(diǎn)時(shí)調(diào)用事件
/// </summary>
public event WatcherEvent NodeChildrenChanged;
/// <summary>
/// 不區(qū)分類型，所有的類型都會(huì)調(diào)用
/// </summary>
public event WatcherEvent AllTypeChanged;
/// <summary>
/// 觸發(fā)，執(zhí)行事件
/// </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>
/// 監(jiān)控事件委托
/// </summary>
/// <param name="event"></param>
public delegate void WatcherEvent(ZookeeperEvent @event);
/// <summary>
/// Zookeeper默認(rèn)日志記錄
/// </summary>
public class ZookeeperLogger : ILogConsumer
{
/// <summary>
/// 是否記錄日志到文件
/// </summary>
public bool LogToFile { get; set; } = false;
/// <summary>
/// 是否記錄堆棧信息
/// </summary>
public bool LogToTrace { get; set; } = true;
/// <summary>
/// 日志級(jí)別
/// </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);
}
}
}
簡(jiǎn)單的使用例子：
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格式，多個(gè)地址之間使用逗號(hào)（,）隔開
string[] address = new string[] { "192.168.209.133:2181", "192.168.209.133:2181", "192.168.209.133:2181" };
//會(huì)話超時(shí)時(shí)間,單位毫秒
int sessionTimeOut = 10000;
ZookeeperHelper zookeeperHelper = new ZookeeperHelper(address, "/");
zookeeperHelper.SessionTimeout = sessionTimeOut;
zookeeperHelper.Connect();//發(fā)起連接
while (!zookeeperHelper.Connected)
{
Thread.Sleep(1000); //停一秒，等待連接完成
}
//創(chuàng)建znode節(jié)點(diǎn)
{
zookeeperHelper.SetData("/mynode", "hello world", true, false);
Console.WriteLine("完成創(chuàng)建節(jié)點(diǎn)");
}
//節(jié)點(diǎn)是否存在
{
var exists = zookeeperHelper.Exists("/mynode");
Console.WriteLine("節(jié)點(diǎn)是否存在：" + exists);
}
//添加監(jiān)聽器
{
zookeeperHelper.WatchAsync("/mynode", (e) =>
{
Console.WriteLine($"recieve: Path-{e.Path} State-{e.State} Type-{e.Type}");
}).Wait();
}
//獲取節(jié)點(diǎn)數(shù)據(jù)
{
var value = zookeeperHelper.GetData("/mynode");
Console.WriteLine("完成讀取節(jié)點(diǎn)：" + value);
}
//設(shè)置節(jié)點(diǎn)數(shù)據(jù)
{
zookeeperHelper.SetData("/mynode", "hello world again");
Console.WriteLine("設(shè)置節(jié)點(diǎn)數(shù)據(jù)");
}
//重新獲取節(jié)點(diǎn)數(shù)據(jù)
{
var value = zookeeperHelper.GetData("/mynode");
Console.WriteLine("重新獲取節(jié)點(diǎn)數(shù)據(jù)：" + value);
}
//移除節(jié)點(diǎn)
{
zookeeperHelper.Delete("/mynode");
Console.WriteLine("移除節(jié)點(diǎn)");
}
Console.WriteLine("完成");
Console.ReadKey();
}
}
}
執(zhí)行結(jié)果：
以上就是C#如何連接使用Zookeeper的詳細(xì)內(nèi)容，更多關(guān)于C# 連接使用Zookeeper的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
您可能感興趣的文章:
相關(guān)文章
這篇文章主要為大家詳細(xì)介紹了C#如何給PDF文件添加水印的相關(guān)資料，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2016-09-09
C#實(shí)現(xiàn)漢字轉(zhuǎn)拼音(多音字)功能詳解
這篇文章主要為大家詳細(xì)介紹了如何利用C#實(shí)現(xiàn)漢字轉(zhuǎn)拼音(支持多音字)的功能，文中的示例代碼講解詳細(xì)，對(duì)我們學(xué)習(xí)C#有一定的幫助，感興趣的小伙伴可以跟隨小編一起了解一下2023-02-02
這篇文章介紹了C#泛型的使用案例，文中通過示例代碼介紹的非常詳細(xì)。對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2022-10-10
Unity3D實(shí)現(xiàn)攝像機(jī)鏡頭移動(dòng)并限制角度
這篇文章主要為大家詳細(xì)介紹了Unity3D實(shí)現(xiàn)攝像機(jī)鏡頭移動(dòng)并限制角度，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2020-05-05
這篇文章介紹了C#中new操作符的工作機(jī)制，文中通過示例代碼介紹的非常詳細(xì)。對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2022-08-08
C#對(duì)XML文件的各種操作實(shí)現(xiàn)方法
C#對(duì)XML文件的各種操作實(shí)現(xiàn)方法，需要的朋友可以參考一下2013-04-04
C# 判斷字符串為空的幾種辦法，需要的朋友可以參考一下2013-04-04
這篇文章主要介紹了如何利用WPF實(shí)現(xiàn)抽屜菜單效果，文中的示例代碼講解詳細(xì)，對(duì)我們學(xué)習(xí)或工作有一定幫助，需要的可以參考一下2022-08-08
本文主要介紹了C# Winform 分頁功能的實(shí)現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-06-06
這篇文章主要介紹了c#后臺(tái)線程訪問前臺(tái)控件并顯示信息示例,需要的朋友可以參考下2014-03-03