C#使用zookeeper

https://blog.csdn.net/XuWei_XuWei/article/details/80611659

1.簡述

zookeeper適用於分布式鎖，配置管理，服務器管理，服務發現場景
c#使用zookeeper基於開源組件ZooKeeperNetEx，詳情GitHub搜一下

2.安裝開發包

使用nuget安裝ZooKeeperNetEx和ZooKeeperNetEx.Recipes，版本是3.4.9.3

3.配置

4.封裝（依賴log4net）

--------AuthEnum.cs------
public enum AuthEnum
{
world = 0,
auth = 1,
digest = 2,
ip = 3,
super = 4
}

-------DefaultWatcher.cs-----
/// <summary>
/// 默認監聽
/// </summary>
public class DefaultWatcher : Watcher
{
private readonly ILog _log;
internal static readonly Task CompletedTask = TaskEx.FromResult(1);

public DefaultWatcher(ILog log)
{
_log = log;
}

/// <summary>
/// 接收通知
/// </summary>
/// <param name="event"></param>
/// <returns></returns>
public override Task process(WatchedEvent @event)
{
_log.ErrorFormat("接收到ZooKeeper服務端的通知，State是：{0}，EventType是：{1}，Path是：{2}", @event.getState(), @event.get_Type(), @event.getPath() ?? string.Empty);
return CompletedTask;
}
}

---------NodeWatcher.cs--------
/// <summary>
/// 節點監聽
/// </summary>
public class NodeWatcher : Watcher
{
private readonly ILog _log;
private string _state;
private Event.EventType _type;
private string _path;
internal static readonly Task CompletedTask = TaskEx.FromResult(1);

public NodeWatcher(ILog log)
{
_log = log;
}

public override Task process(WatchedEvent @event)
{
_state = @event.getState().ToString();
_type = @event.get_Type();
_path = @event.getPath();
switch (_type)
{
case Event.EventType.NodeCreated:
HandleCreate();
break;
case Event.EventType.NodeDeleted:
HandleDelete();
break;
case Event.EventType.NodeDataChanged:
HandleDataChange();
break;
case Event.EventType.NodeChildrenChanged:
HandleChildrenChange();
break;
default:
HandleNone();
break;
}
return CompletedTask;
}

/// <summary>
/// 創建節點事件
/// </summary>
private void HandleCreate()
{
_log.ErrorFormat("NodeCreated");
}

private void HandleDelete()
{
_log.ErrorFormat("NodeDeleted");
}

private void HandleDataChange()
{
_log.ErrorFormat("NodeDataChanged");
}

private void HandleChildrenChange()
{
_log.ErrorFormat("NodeChildrenChanged");
}

private void HandleNone()
{
_log.ErrorFormat(_state);
}
}

--------ZooKeeperHelper.cs--------
public class ZooKeeperHelper
{
private readonly ILog _log;
private bool _isonline = Convert.ToBoolean(ConfigurationManager.AppSettings["isonline"].ConventToInt32());
private List<string> _address;
private int _sessiontTimeout = 10*1000;//10秒
private ZooKeeper _zooKeeper;
private int _connectTimeout = 3*30*1000;//每個zookeeper實例嘗試連接最長30秒
private string _success = "success";
private string _fail = "fail";

public ZooKeeperHelper(ILog log,int sessionTimeOut = 10*1000)
{
_log = log;
_sessiontTimeout = sessionTimeOut;
if (_isonline)
{
//正式環境
_address = new List<string> { "192.168.204.92:2181", "192.168.204.92:2182", "192.168.204.92:2183" };
}
else
{
//本地調試環境
_address = new List<string> { "10.168.100.102:2181", "10.168.100.102:2182", "10.168.100.102:2183" };
}
}

/// <summary>
/// 返回null表示連接不成功
/// </summary>
/// <param name="authEnum"></param>
/// <param name="authInfo"></param>
/// <returns></returns>
public ZooKeeper Connect(AuthEnum authEnum ,string authInfo) {
try
{
foreach (string address in _address)
{
_zooKeeper = new ZooKeeper(address, _sessiontTimeout, new DefaultWatcher(_log));
if (authEnum != AuthEnum.world)
{
_zooKeeper.addAuthInfo(authEnum.ToString(), System.Text.Encoding.UTF8.GetBytes(authInfo));
}
Stopwatch stopwatch = new Stopwatch();
stopwatch.Start();
while (stopwatch.ElapsedMilliseconds < _connectTimeout/_address.Count)
{
ZooKeeper.States states = _zooKeeper.getState();
if (states == ZooKeeper.States.CONNECTED || states == ZooKeeper.States.CONNECTEDREADONLY)
{
break;
}
}
stopwatch.Stop();
if (_zooKeeper.getState().ToString().ToUpper().Contains("CONNECTED"))
{
break;
}
}

return _zooKeeper;
}
catch (Exception ex)
{
_log.ErrorFormat("連接zookeeper發生異常：{0}", ex.Message + ex.StackTrace);
}
return null;
}

/// <summary>
/// 創建節點,不能在臨時節點下創建子節點
/// </summary>
/// <param name="path">不要使用path等關鍵字作為路徑</param>
/// <param name="data"></param>
/// <param name="persistent"></param>
/// <returns></returns>
public string CreateNode(string path,string data,bool persistent = false) {
try {
Task<string> task = _zooKeeper.createAsync(path, System.Text.Encoding.UTF8.GetBytes(data), ZooDefs.Ids.OPEN_ACL_UNSAFE, persistent ? CreateMode.PERSISTENT : CreateMode.EPHEMERAL);
task.Wait();
if (!string.IsNullOrEmpty(task.Result) && task.Status.ToString().ToLower() == "RanToCompletion".ToLower())
{
return task.Result;
}
}
catch (Exception ex) {
_log.ErrorFormat("創建節點發生異常：{0}({1}),{2}",path ,data,ex.Message + ex.StackTrace);
}
return _fail;
}

/// <summary>
/// 刪除節點,刪除節點的子節點個數必須為0，否則請先刪除子節點
/// </summary>
/// <param name="path">不要使用path等關鍵字作為路徑</param>
/// <returns></returns>
public string DeleteNode(string path) {
try
{
Task task = _zooKeeper.deleteAsync(path);
task.Wait();
if (task.Status.ToString().ToLower() == "RanToCompletion".ToLower())
{
return _success;
}
}
catch (Exception ex)
{
_log.ErrorFormat("刪除節點發生異常：{0},{1}", path,ex.Message + ex.StackTrace);
}
return _fail;
}

/// <summary>
/// 給節點設置數據
/// </summary>
/// <param name="path">不要使用path等關鍵字作為路徑</param>
/// <param name="data"></param>
/// <returns></returns>
public string SetData(string path,string data) {
try
{
Task<org.apache.zookeeper.data.Stat> stat = _zooKeeper.setDataAsync(path, System.Text.Encoding.UTF8.GetBytes(data));
stat.Wait();
if (stat.Result != null && stat.Status.ToString().ToLower() == "RanToCompletion".ToLower())
{
return _success;
}
}
catch (Exception ex)
{
_log.ErrorFormat("設置節點數據發生異常：{0}({1}),{2}",path,data, ex.Message + ex.StackTrace);
}
return _fail;
}

/// <summary>
/// 判斷節點是否存在
/// </summary>
/// <param name="path">不要使用path等關鍵字作為路徑</param>
/// <param name="watcher"></param>
/// <returns></returns>
public string ExistsNode(string path, Watcher watcher = null) {
try
{
Task<org.apache.zookeeper.data.Stat> stat = _zooKeeper.existsAsync(path, watcher);
stat.Wait();
if (stat.Result != null && stat.Status.ToString().ToLower() == "RanToCompletion".ToLower())
{
return _success;
}
}
catch (Exception ex)
{
_log.ErrorFormat("判定節點存在與否發生異常：{0},{1}", path, ex.Message + ex.StackTrace);
}
return _fail;
}

/// <summary>
/// 得到節點相關信息
/// </summary>
/// <param name="path">不要使用path等關鍵字作為路徑</param>
/// <param name="watcher"></param>
/// <returns></returns>
public Stat GetNode(string path, Watcher watcher = null)
{
try
{
Task<org.apache.zookeeper.data.Stat> stat = _zooKeeper.existsAsync(path, watcher);
stat.Wait();
if (stat.Result != null && stat.Status.ToString().ToLower() == "RanToCompletion".ToLower())
{
return stat.Result;
}
}
catch (Exception ex)
{
_log.ErrorFormat("得到節點信息發生異常：{0},{1}", path, ex.Message + ex.StackTrace);
}
return null;
}

/// <summary>
/// 得到節點數據
/// </summary>
/// <param name="path">不要使用path等關鍵字作為路徑</param>
/// <param name="watcher"></param>
/// <returns></returns>
public string GetData(string path, Watcher watcher = null) {
try
{
Task<DataResult> dataResult = _zooKeeper.getDataAsync(path, watcher);
dataResult.Wait();
if (dataResult.Result != null && dataResult.Status.ToString().ToLower() == "RanToCompletion".ToLower())
{
return Encoding.UTF8.GetString(dataResult.Result.Data);
}
}
catch (Exception ex)
{
_log.ErrorFormat("得到節點數據發生異常：{0},{1}", path, ex.Message + ex.StackTrace);
}
return _fail;
}

/// <summary>
/// 得到后代節點路徑
/// </summary>
/// <param name="path">不要使用path等關鍵字作為路徑</param>
/// <param name="watcher"></param>
/// <returns></returns>
public List<string> GetChildren(string path,Watcher watcher = null) {
try
{
Task<ChildrenResult> childrenResult = _zooKeeper.getChildrenAsync(path, watcher);
childrenResult.Wait();
if (childrenResult.Result != null && childrenResult.Status.ToString().ToLower() == "RanToCompletion".ToLower())
{
return childrenResult.Result.Children;
}
}
catch (Exception ex)
{
_log.ErrorFormat("得到后代節點發生異常：{0},{1}", path, ex.Message + ex.StackTrace);
}
return null;
}

/// <summary>
/// 關閉連接
/// </summary>
/// <returns></returns>
public string Close() {
try
{
Task task = _zooKeeper.closeAsync();
task.Wait();
if (task.Status.ToString().ToLower() == "RanToCompletion".ToLower())
{
return _success;
}
}
catch (Exception ex)
{
LogHelper.GetLoger().ErrorFormat("關閉zookeeper發生異常：{0}", ex.Message + ex.StackTrace);
}
return _fail;
}

/// <summary>
/// 得到連接狀態
/// </summary>
/// <returns></returns>
public string GetState() {
try
{
if (_zooKeeper != null)
{
ZooKeeper.States states = _zooKeeper.getState();
return states.ToString();
}
}
catch (Exception ex)
{
_log.ErrorFormat("獲取zookeeper連接狀態發生異常：{0}", ex.Message + ex.StackTrace);
}
return _fail;
}

/// <summary>
/// 是否已經連接
/// </summary>
/// <returns></returns>
public bool Connected()
{
try
{
if (_zooKeeper != null)
{
ZooKeeper.States states = _zooKeeper.getState();
if (states == ZooKeeper.States.CONNECTED || states == ZooKeeper.States.CONNECTEDREADONLY)
{
return true;
}
}
}
catch (Exception ex)
{
_log.ErrorFormat("獲取zookeeper連接狀態發生異常：{0}", ex.Message + ex.StackTrace);
}
return false;
}
}

--------TaskHelper.cs--------
public class TaskHelper
{
private readonly ILog _log;
private int forcerun = Convert.ToInt32(ConfigurationManager.AppSettings["forcerun"]);
private string servicename = ConfigurationManager.AppSettings["servicename"];
private string pathname = ConfigurationManager.AppSettings["pathname"];
private ZooKeeperHelper _zooKeeperHelper;
private ZooKeeper _zooKeeper;
//單例輸出，否則通知過多可能導致內存溢出
private static TaskHelper _taskHelper;
private static object _obj = new object();

private TaskHelper(ILog log,int sessionTimeOut = 10 * 1000)
{
_log = log;
_zooKeeperHelper = new ZooKeeperHelper(_log,sessionTimeOut);
}

public static TaskHelper GetInstance(ILog log, int sessionTimeOut = 10 * 1000)
{
if (_taskHelper == null)
{
lock (_obj)
{
if (_taskHelper == null)
{
_taskHelper = new TaskHelper(log,sessionTimeOut);
}
}
}
return _taskHelper;
}

public bool Return()
{
if (forcerun != 1)
{
try
{
if (!_zooKeeperHelper.Connected())
{
_zooKeeper = _zooKeeperHelper.Connect(AuthEnum.world, "");
if (_zooKeeper == null)
{
_log.ErrorFormat("連接zooKeeper失敗，時間是：{0}", DateTime.Now);
return true;
}
}
string path = ("/" + pathname);
string data = servicename;
string str = _zooKeeperHelper.ExistsNode(path, new NodeWatcher(_log));
if (str != "success")
{
str = _zooKeeperHelper.CreateNode(path, data);
if (str != path)
{
_log.ErrorFormat("創建路徑失敗，時間是：{0}", DateTime.Now);
return true;
}
}

string lockname = _zooKeeperHelper.GetData(path, new NodeWatcher(_log));

#region 測試通知
//string cg = _zooKeeperHelper.SetData(path, "hahhahahah");
//cg = _zooKeeperHelper.GetData(path, new NodeWatcher(_log));
//cg = _zooKeeperHelper.SetData(path, "1111111111");
//cg = _zooKeeperHelper.GetData(path, new NodeWatcher(_log));
//cg = _zooKeeperHelper.DeleteNode(path);
#endregion

//執行標識
if (lockname != servicename)
{
_log.ErrorFormat("非工作時間，當前執行的服務是：{0}，時間是：{1}", lockname, DateTime.Now);
return true;
}
}
catch (Exception exception)
{
_log.ErrorFormat("zooKeeperHelper出現異常：{0}，時間是：{1}", exception.Message + exception.StackTrace, DateTime.Now);
}
}
return false;
}
}

5.分布式鎖應用場景

//協調分布式服務
TaskHelper.GetInstance(log).Return()
---------------------
作者：DO_DAJIANGJUN
來源：CSDN
原文：https://blog.csdn.net/XuWei_XuWei/article/details/80611659
版權聲明：本文為博主原創文章，轉載請附上博文鏈接！

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 Zookeeper基礎教程（四）：C#連接使用Zookeeper zookeeper的使用demo（c#/java） zookeeper 的介紹以及c api的使用 Zookeeper基礎教程（五）：C#實現Zookeeper分布式鎖 Zookeeper 使用 C# GridView 的使用 C# 使用 redis C# Dapper的使用 SQLite在C#的使用（二）Protobuf的C#使用