為什麼要用分布式鎖?
先上一張截圖,這是在浏覽别人的部落格時看到的.
在了解為什麼要用分布式鎖之前,我們應該知道到底什麼是分布式鎖.
鎖按照不同的次元,有多種分類.比如
1.悲觀鎖,樂觀鎖;
2.公平鎖,非公平鎖;
3.獨享鎖,共享鎖;
4.線程鎖,程序鎖;
等等.
我們平時用的鎖,比如 lock,它是線程鎖,主要用來給方法,代碼塊加鎖.由于程序的記憶體單元是被其所有線程共享的,是以線程鎖控制的實際是多個線程對同一塊記憶體區域的通路.
有線程鎖,就必然有程序鎖.顧名思義,程序鎖的目的是控制多個程序對共享資源的通路.因為程序之間彼此獨立,各個程序是無法控制其他程序對資源的通路,是以隻能通過作業系統來控制.比如 Mutex.
但是程序鎖有一個前提,那就是需要多個程序在同一個系統中,如果多個程序不在同一個系統,那就隻能使用分布式鎖來控制了.
分布式鎖是控制分布式系統中不同系統之間通路共享資源的一種鎖實作.它和線程鎖,程序鎖的作用都是一樣,隻是範圍不一樣.
是以要實作分布式鎖,就必須依靠第三方存儲媒體來存儲鎖的資訊.因為各個程序之間彼此誰都不服誰,隻能找一個帶頭大哥咯;
以下示例需引用NUGET: CSRedisCore
示例一
CSRedisClient redisClient = new CSRedis.CSRedisClient("127.0.0.1:6379,defaultDatabase=0"); var lockKey = "lockKey"; var stock = 5;//商品庫存 var taskCount = 10;//線程數量 redisClient.Del(lockKey);//測試前,先把鎖删了. for (int i = 0; i < taskCount; i++) { Task.Run(() => { //擷取鎖 do { //setnx : key不存在才會成功,存在則失敗. var success = redisClient.SetNx(lockKey, 1); if (success == true) { break; } Thread.Sleep(TimeSpan.FromSeconds(1));//休息1秒再嘗試擷取鎖 } while (true); Console.WriteLine($"線程:{Task.CurrentId} 拿到了鎖,開始消費"); if (stock <= 0) { Console.WriteLine($"庫存不足,線程:{Task.CurrentId} 搶購失敗!"); redisClient.Del(lockKey); return; } stock--; //模拟處理業務 Thread.Sleep(TimeSpan.FromSeconds(new Random().Next(1, 3))); Console.WriteLine($"線程:{Task.CurrentId} 消費完畢!剩餘 {stock} 個"); //業務處理完後,釋放鎖. redisClient.Del(lockKey); }); } 運作結果:
看起來貌似沒毛病,實際上上述代碼有個緻命的問題:
當某個線程拿到鎖之後,如果系統崩潰了,那麼鎖永遠都不會被釋放.是以,我們應該給鎖加一個過期時間,當時間到了,還沒有被主動釋放,我們就讓redis釋放掉它,以保證其他消費者可以拿到鎖,進行消費.
這裡給鎖加過期時間也有講究,不能拿到鎖後再加,比如:
...... //setnx : key不存在才會成功,存在則失敗. var success = redisClient.SetNx(lockKey, 1); if (success == true) { redisClient.Set(lockKey, 1, expireSeconds: 5); break; } 這樣操作的話,擷取鎖和設定鎖的過期時間就不是原子操作,同樣會出現上面提到的問題.Redis 提供了一個合而為一的操作可以解決這個問題.
//set : key存在則失敗,不存在才會成功,并且過期時間5秒 var success = redisClient.Set(lockKey, 1, expireSeconds: 5, exists: RedisExistence.Nx); 這個問題雖然解決了,但随之産生了一個新的問題:
假設有3個線程A,B,C
當線程A拿到鎖後執行業務的時候逾時了,超過了鎖的過期時間還沒執行完,這時候鎖被Redis釋放了,
于是線程B拿到了鎖并開始執行業務邏輯.
當線程B的業務邏輯還沒執行完的時候,線程A的業務邏輯執行完了,于是乎就跑去釋放掉了鎖.
這時候線程C就可以拿到鎖開始執行它的業務邏輯.
這不就亂套了麼...
是以,線程在釋放鎖的時候應該判斷這個鎖還屬不屬于自己.
是以,在設定鎖的時候,redis的value值不能像上面代碼那樣,随便給個1,而應該給一個随機值,代表目前線程.
var id = Guid.NewGuid().ToString("N"); //擷取鎖 do { //set : key存在則失敗,不存在才會成功,并且過期時間5秒 var success = redisClient.Set(lockKey, id, expireSeconds: 5, exists: RedisExistence.Nx); if (success == true) { break; } Thread.Sleep(TimeSpan.FromSeconds(1));//休息1秒再嘗試擷取鎖 } while (true); Console.WriteLine($"線程:{Task.CurrentId} 拿到了鎖,開始消費"); .........//業務處理完後,釋放鎖. var value = redisClient.Get(lockKey); if (value == id) { redisClient.Del(lockKey); } 完美了嗎?
不完美.還是老生常談的問題,取value和删除key 分了兩步走,不是原子操作.
并且,這裡還不能用pipe,因為需要根據取到的value來決定下一個操作.上面設定過期時間倒是可以用pipe.
是以,這裡隻能用lua.
完整的代碼如下:
CSRedisClient redisClient = new CSRedis.CSRedisClient("127.0.0.1:6379,defaultDatabase=0"); var lockKey = "lockKey"; var stock = 5;//商品庫存 var taskCount = 10;//線程數量 var script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";//釋放鎖的redis腳本 redisClient.Del(lockKey);//測試前,先把鎖删了. for (int i = 0; i < taskCount; i++) { Task.Run(() => { var id = Guid.NewGuid().ToString("N"); //擷取鎖 do { //set : key存在則失敗,不存在才會成功,并且過期時間5秒 var success = redisClient.Set(lockKey, id, expireSeconds: 5, exists: RedisExistence.Nx); if (success == true) { break; } Thread.Sleep(TimeSpan.FromSeconds(1));//休息1秒再嘗試擷取鎖 } while (true); Console.WriteLine($"線程:{Task.CurrentId} 拿到了鎖,開始消費"); if (stock <= 0) { Console.WriteLine($"庫存不足,線程:{Task.CurrentId} 搶購失敗!"); redisClient.Eval(script,lockKey,id); return; } stock--; //模拟處理業務,這裡不考慮失敗的情況 Thread.Sleep(TimeSpan.FromSeconds(new Random().Next(1, 3))); Console.WriteLine($"線程:{Task.CurrentId} 消費完畢!剩餘 {stock} 個"); //業務處理完後,釋放鎖. redisClient.Eval(script, lockKey, id); }); } 這篇文章隻介紹了單節點Redis的分布式鎖,因為單節點,是以不是高可用.
多節點Redis則需要用官方介紹的RedLock,這玩意有點繞,我需要捋一捋.