Zookeeper基礎教程（五）：C#實現Zookeeper分散式鎖
分散式鎖
網際網路初期，我們系統一般都是單點部署，也就是在一臺伺服器完成系統的部署，後期隨著使用者量的增加，伺服器的壓力也越來越大，響應速度越來越慢，甚至出現伺服器崩潰的情況。
為解決伺服器壓力太大，響應慢的特點，分散式系統部署出現了。
簡單的說，就是我們將系統資源部署到多臺伺服器中，然後使用一臺伺服器做入口代理，根據一些決策將接收到的請求轉發到資源伺服器，這也就是我們常說的 反向代理（一般就是使用nginx）
雖然分散式解決了伺服器壓力的問題，但也帶來了新的問題。
比如，我們有一個下單統計的功能，當完成下單後，需要執行統計功能，而在高訪問的情況下，可能有兩個下單請求（A和B）同時完成，然後一起執行了統計功能，這樣可能導致的結果就是A請求未將B請求資料統計在內，而B請求可能也未將A請求資料統計在內，這樣就造成了資料的統計錯，這個問題的產生的根本原因就是統計功能的併發導致的，如果是單點部署的系統，我們簡單的使用一個鎖操作就能完成了，但是在分散式環境下，A和B請求可能同時執行在兩個伺服器中，普通的鎖就不能起到效果了，這個時候就要使用分散式鎖了。
Zookeeper分散式鎖原理
分散式鎖的實現發放有多種，簡單的，我們可以使用資料庫表去實現它，也可以使用redis去實現它，這裡要使用的Zookeeper去實現分散式鎖
Zookeeper分散式鎖的原理是巧妙的是使用了znode臨時節點的特點和監聽（watcher）機制，監聽機制很簡單，就是我們可以給znode新增一個監聽器，當znode節點狀態發生改變時（如：資料內容改變，節點被刪除），會通知到監聽器。
前面幾節介紹過znode有三種類型
PERSISTENT：持久節點，即使在建立該特定znode的客戶端斷開連線後，持久節點仍然存在。預設情況下，除非另有說明，否則所有znode都是持久的。 SEQUENTIAL：臨時節點，客戶端是連線狀態時，臨時節點就是有效的。當客戶端與ZooKeeper集合斷開連線時，臨時節點會自動刪除。臨時節點不允許有子節點。臨時節點在leader選舉中起著重要作用。 EPHEMERAL：順序節點，可以是持久的或臨時的。當一個新的znode被建立為一個順序節點時，ZooKeeper通過將10位的序列號附加到原始名稱來設定znode的路徑，順序節點在鎖定和同步中起重要作用。
其中，順序節點，可以是持久的或臨時的，而臨時節點有個特點，就是它屬於建立它的那個會話，當會話斷開，臨時節點就會自動刪除，如果在臨時節點上註冊了監聽器，那麼監聽器就會收到通知，如果臨時節點有了時間順序，那我們為實現分散式鎖就又有一個想法：
假如在Zookeeper中有一個znode節點/Locker
1、當client1連線Zookeeper時，先判斷/Locker節點是否存在子節點，如果沒有子節點，那麼會在/Locker節點下建立一個臨時順序的znode節點，假如是/client1，表示client1獲取了鎖狀態，client1可以繼續執行。
2、當client2連線Zookeeper時，先判斷/Locker節點是否存在子節點，發現已經存在子節點了，然後獲取/Locker下的所有子節點，同時按時間順序排序，在最後一個節點，也就是/client1節點上註冊一個監聽器（watcher1），同時在/Locker節點下建立一個臨時順序的znode節點，假如是/client2。同時client2將被阻塞，而阻塞狀態的釋放是在監聽器（watcher1）中的。
3、當client3連線Zookeeper時，先判斷/Locker節點是否存在子節點，發現已經存在子節點了，然後獲取/Locker下的所有子節點，同時按時間順序排序，在最後一個節點，也就是/client2節點上註冊一個監聽器（watcher2），同時在/Locker節點下建立一個臨時順序的znode節點，假如是/client3。同時client2將被阻塞，而阻塞狀態的釋放是在監聽器（watcher2）中的。
以此類推。
4、當client1執行完操作了，斷開Zookeeper的連線，因為/client1是臨時順序節點，於是將會自動刪除，而client2已經往/client1節點中註冊了一個監聽器（watcher1）,於是watcher1將會受到通知，而watcher1又會釋放client2的阻塞狀態。於是client2獲取鎖狀態，繼續執行。
5、當client2執行完操作了，斷開Zookeeper的連線，因為/client2是臨時順序節點，於是將會自動刪除，而client3已經往/client2節點中註冊了一個監聽器（watcher2）,於是watcher2將會受到通知，而watcher2又會釋放client3的阻塞狀態。於是client3獲取鎖狀態，繼續執行。
以此類推。
這樣，不管分散式環境中有幾臺伺服器，都可以保證程式的排隊似的執行了。
C#實現Zookeeper分散式鎖
上一節有封裝過一個ZookeeperHelper的輔助類（Zookeeper基礎教程（四）：C#連線使用Zookeeper），使用這個輔助類實現了一個ZookeeperLocker類：
using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; namespace AspNetCore.ZookeeperConsole { /// <summary> /// 基於Zookeeper的分散式鎖 /// </summary> public class ZookeeperLocker : IDisposable { /// <summary> /// 單點鎖 /// </summary> static object locker = new object(); /// <summary> /// Zookeeper叢集地址 /// </summary> string[] address; /// <summary> /// Zookeeper操作輔助類 /// </summary> ZookeeperHelper zookeeperHelper; /// <summary> /// 建構函式 /// </summary> /// <param name="lockerPath">分散式鎖的根路徑</param> /// <param name="address">叢集地址</param> public ZookeeperLocker(string lockerPath,params string[] address) : this(lockerPath,0,address) { } /// <summary> /// 建構函式 /// </summary> /// <param name="lockerPath">分散式鎖的根路徑</param> /// <param name="sessionTimeout">回話過期時間</param> /// <param name="address">叢集地址</param> 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); } } } /// <summary> /// 生成一個鎖 /// </summary> /// <returns>返回鎖名</returns> public string CreateLock() { var path = Guid.NewGuid().ToString().Replace("-",""); while (zookeeperHelper.Exists(path)) { path = Guid.NewGuid().ToString().Replace("-",""); } return CreateLock(path); } /// <summary> /// 使用指定的路徑名稱設定鎖 /// </summary> /// <param name="path">鎖名，不能包含路徑分隔符（/）</param> /// <returns>返回鎖名</returns> public string CreateLock(string path) { if (path.Contains("/")) { throw new ArgumentException("invalid path"); } return zookeeperHelper.SetData(path,false,true); } /// <summary> /// 獲取鎖 /// </summary> /// <param name="path">鎖名</param> /// <returns>如果獲得鎖返回true，否則一直等待</returns> public bool Lock(string path) { return LockAsync(path).GetAwaiter().GetResult(); } /// <summary> /// 獲取鎖 /// </summary> /// <param name="path">鎖名</param> /// <param name="millisecondsTimeout">超時時間，單位：毫秒</param> /// <returns>如果獲得鎖返回true，否則等待指定時間後返回false</returns> public bool Lock(string path,int millisecondsTimeout) { return LockAsync(path,millisecondsTimeout).GetAwaiter().GetResult(); } /// <summary> /// 非同步獲取鎖等等 /// </summary> /// <param name="path">鎖名</param> /// <returns>如果獲得鎖返回true，否則一直等待</returns> public async Task<bool> LockAsync(string path) { return await LockAsync(path,System.Threading.Timeout.Infinite); } /// <summary> /// 非同步獲取鎖等等 /// </summary> /// <param name="path">鎖名</param> /// <param name="millisecondsTimeout">超時時間，單位：毫秒</param> /// <returns>如果獲得鎖返回true，否則等待指定時間後返回false</returns> public async Task<bool> 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; } /// <summary> /// 釋放資源 /// </summary> public void Dispose() { zookeeperHelper.Dispose(); } } }View Code
現在寫個程式可以模擬一下
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(); } } }
執行結果如下：
可以發現，鎖功能是實現了的
如果程式執行中列印日誌：Client session timed out,have not heard from server in 8853ms for sessionid 0x1000000ec5500b2
或者直接丟擲異常：org.apache.zookeeper.KeeperException.ConnectionLossException:“Exception_WasThrown”
只需要適當調整sessionTimeOut時間即可