[ZooKeeper.net] 3 ZooKeeper的分布式锁

  1 public class ZooKeeprDistributedLock : IWatcher
  2     {
  3         /// <summary>
  4         /// zk链接字符串
  5         /// </summary>
  6         private String connectString = "127.0.0.1:2181";
  7         private ZooKeeper zk;
  8         private string root = "/locks"; //根        
  9         private string lockName; //竞争资源的标志        
 10         private string waitNode; //等待前一个锁        
 11         private string myZnode; //当前锁               
 12         private AutoResetEvent autoevent;
 13         private TimeSpan sessionTimeout = TimeSpan.FromMilliseconds(50000);
 14         private IList<Exception> exception = new List<Exception>();
 15 
 16         /// <summary>
 17         /// 创建分布式锁
 18         /// </summary>
 19         /// <param name="lockName">竞争资源标志,lockName中不能包含单词lock</param>
 20         public ZooKeeprDistributedLock(string lockName)
 21         {
 22             this.lockName = lockName;
 23             // 创建一个与服务器的连接            
 24             try
 25             {
 26                 zk = new ZooKeeper(connectString, sessionTimeout, this);
 27                 Stopwatch sw = new Stopwatch();
 28                 sw.Start();
 29                 while (true)
 30                 {
 31                     if (zk.State == States.CONNECTING) { break; }
 32                     if (zk.State == States.CONNECTED) { break; }
 33                 }
 34                 sw.Stop();
 35                 TimeSpan ts2 = sw.Elapsed;
 36                 Console.WriteLine("zoo连接总共花费{0}ms.", ts2.TotalMilliseconds);
 37 
 38                 var stat = zk.Exists(root, false);
 39                 if (stat == null)
 40                 {
 41                     // 创建根节点                    
 42                     zk.Create(root, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.Persistent);
 43                 }
 44             }
 45             catch (KeeperException e)
 46             {
 47                 throw e;
 48             }
 49         }
 50 
 51         /// <summary>        
 52         /// zookeeper节点的监视器        
 53         /// </summary>        
 54         public virtual void Process(WatchedEvent @event)
 55 
 56         {
 57             if (this.autoevent != null)
 58             {
 59                 //将事件状态设置为终止状态，允许一个或多个等待线程继续；如果该操作成功，则返回true；否则，返回false
 60                 this.autoevent.Set();
 61             }
 62         }
 63 
 64         public virtual bool tryLock()
 65         {
 66             try
 67             {
 68                 string splitStr = "_lock_";
 69                 if (lockName.Contains(splitStr))
 70                 {
 71                     //throw new LockException("lockName can not contains \\u000B");                
 72                 }
 73                 //创建临时子节点                
 74                 myZnode = zk.Create(root + "/" + lockName + splitStr, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EphemeralSequential);
 75                 Console.WriteLine(myZnode + "    创建完成！ ");
 76                 //取出所有子节点                
 77                 IList<string> subNodes = zk.GetChildren(root, false).ToList<string>();
 78                 //取出所有lockName的锁                
 79                 IList<string> lockObjNodes = new List<string>();
 80                 foreach (string node in subNodes)
 81                 {
 82                     if (node.StartsWith(lockName))
 83                     {
 84                         lockObjNodes.Add(node);
 85                     }
 86                 }
 87                 Array alockObjNodes = lockObjNodes.ToArray();
 88                 Array.Sort(alockObjNodes);
 89                 Console.WriteLine(myZnode + "==" + lockObjNodes[0]);
 90                 if (myZnode.Equals(root + "/" + lockObjNodes[0]))
 91                 {
 92                     //如果是最小的节点,则表示取得锁   
 93                     Console.WriteLine(myZnode + "    获取锁成功！ ");
 94                     return true;
 95                 }
 96                 //如果不是最小的节点，找到比自己小1的节点               
 97                 string subMyZnode = myZnode.Substring(myZnode.LastIndexOf("/", StringComparison.Ordinal) + 1);
 98                 waitNode = lockObjNodes[Array.BinarySearch(alockObjNodes, subMyZnode) - 1];
 99             }
100             catch (KeeperException e)
101             {
102                 throw e;
103             }
104             return false;
105         }
106 
107 
108         public virtual bool tryLock(TimeSpan time)
109         {
110             try
111             {
112                 if (this.tryLock())
113                 {
114                     return true;
115                 }
116                 return waitForLock(waitNode, time);
117             }
118             catch (KeeperException e)
119             {
120                 throw e;
121             }
122         }
123 
124         /// <summary>
125         /// 等待锁
126         /// </summary>
127         /// <param name="lower">需等待的锁节点</param>
128         /// <param name="waitTime">等待时间</param>
129         /// <returns></returns>
130         private bool waitForLock(string lower, TimeSpan waitTime)
131         {
132             var stat = zk.Exists(root + "/" + lower, true);
133             //判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听            
134             if (stat != null)
135             {
136                 Console.WriteLine("Thread " + System.Threading.Thread.CurrentThread.Name + " waiting for " + root + "/" + lower);
137                 autoevent = new AutoResetEvent(false);
138                 //阻止当前线程，直到当前实例收到信号，使用 TimeSpan 度量时间间隔并指定是否在等待之前退出同步域
139                 bool r = autoevent.WaitOne(waitTime);
140                 autoevent.Dispose();
141                 autoevent = null;
142                 return r;
143             }
144             else return true;
145         }
146 
147         /// <summary>
148         /// 解除锁
149         /// </summary>
150         public virtual void unlock()
151         {
152             try
153             {
154                 Console.WriteLine("unlock " + myZnode);
155                 zk.Delete(myZnode, -1);
156                 myZnode = null;
157                 zk.Dispose();
158             }
159             catch (KeeperException e)
160             {
161                 throw e;
162             }
163         }
164     }