C# 实现Zookeeper分布式锁的参考示例

 using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; namespace AspNetCore.ZookeeperConsole { /// 
 /// 基于Zookeeper的分布式锁 /// 
 public class ZookeeperLocker : IDisposable { /// 
 /// 单点锁 /// 
 static object locker = new object(); /// 
 /// Zookeeper集群地址 /// 
 string[] address; /// 
 /// Zookeeper操作辅助类 /// 
 ZookeeperHelper zookeeperHelper; /// 
 /// 构造函数 /// 
 /// 分布式锁的根路径 /// 集群地址 public ZookeeperLocker(string lockerPath, params string[] address) : this(lockerPath, 0, address) { } /// 
 /// 构造函数 /// 
 /// 分布式锁的根路径 /// 回话过期时间 /// 集群地址 public ZookeeperLocker(string lockerPath, int sessionTimeout, params string[] address) { this.address = address.ToArray(); zookeeperHelper = new ZookeeperHelper(address, lockerPath); if (sessionTimeout > 0) { zookeeperHelper.SessionTimeout = sessionTimeout; } if (!zookeeperHelper.Connect()) { throw new Exception("connect failed:" + string.Join(",", address)); } lock (locker) { if (!zookeeperHelper.Exists())//根节点不存在则创建 { zookeeperHelper.SetData("", "", true); } } } /// 
 /// 生成一个锁 /// 
 /// 返回锁名 public string CreateLock() { var path = Guid.NewGuid().ToString().Replace("-", ""); while (zookeeperHelper.Exists(path)) { path = Guid.NewGuid().ToString().Replace("-", ""); } return CreateLock(path); } /// 
 /// 使用指定的路径名称设置锁 /// 
 /// 锁名，不能包含路径分隔符（/） /// 返回锁名 public string CreateLock(string path) { if (path.Contains("/")) { throw new ArgumentException("invalid path"); } return zookeeperHelper.SetData(path, "", false, true); } /// 
 /// 获取锁 /// 
 /// 锁名 /// 如果获得锁返回true，否则一直等待 public bool Lock(string path) { return LockAsync(path).GetAwaiter().GetResult(); } /// 
 /// 获取锁 /// 
 /// 锁名 /// 超时时间，单位：毫秒 /// 如果获得锁返回true，否则等待指定时间后返回false public bool Lock(string path, int millisecondsTimeout) { return LockAsync(path, millisecondsTimeout).GetAwaiter().GetResult(); } /// 
 /// 异步获取锁等等 /// 
 /// 锁名 /// 如果获得锁返回true，否则一直等待 public async Task LockAsync(string path) { return await LockAsync(path, System.Threading.Timeout.Infinite); } /// 
 /// 异步获取锁等等 /// 
 /// 锁名 /// 超时时间，单位：毫秒 /// 如果获得锁返回true，否则等待指定时间后返回false public async Task LockAsync(string path, int millisecondsTimeout) { var array = await zookeeperHelper.GetChildrenAsync("", true); if (array != null && array.Length > 0) { var first = array.FirstOrDefault(); if (first == path)//正好是优先级最高的，则获得锁 { return true; } var index = array.ToList().IndexOf(path); if (index > 0) { //否则添加监听 var are = new AutoResetEvent(false); var watcher = new NodeWatcher(); watcher.NodeDeleted += (ze) => { are.Set(); }; if (await zookeeperHelper.WatchAsync(array[index - 1], watcher))//监听顺序节点中的前一个节点 { if (!are.WaitOne(millisecondsTimeout)) { return false; } } are.Dispose(); } else { throw new InvalidOperationException($"no locker found in path:{zookeeperHelper.CurrentPath}"); } } return true; } /// 
 /// 释放资源 /// 
 public void Dispose() { zookeeperHelper.Dispose(); } } }

 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; //锁节点根路径 string lockerPath = "/Locker"; for (var i = 0; i <10; i++) { string client = "client" + i; //多线程模拟并发 new Thread(() => { using (ZookeeperLocker zookeeperLocker = new ZookeeperLocker(lockerPath, sessionTimeOut, address)) { string path = zookeeperLocker.CreateLock(); if (zookeeperLocker.Lock(path)) { //模拟处理过程 Console.WriteLine($"【{client}】获得锁:{DateTime.Now}"); Thread.Sleep(3000); Console.WriteLine($"【{client}】处理完成:{DateTime.Now}"); } else { Console.WriteLine($"【{client}】获得锁失败:{DateTime.Now}"); } } }).Start(); } Console.ReadKey(); } } }