【Redis学习】Redis实现分布式锁
??目前幾乎很多大型網站及應用都是分布式部署的，分布式場景中的數據一致性問題一直是一個比較重要的話題。分布式的CAP理論告訴我們“任何一個分布式系統都無法同時滿足一致性（Consistency）、可用性（Availability）和分區容錯性（Partition tolerance），最多只能同時滿足兩項?！彼?#xff0c;很多系統在設計之初就要對這三者做出取舍。在互聯網領域的絕大多數的場景中，都需要犧牲強一致性來換取系統的高可用性，系統往往只需要保證“最終一致性”，只要這個最終時間是在用戶可以接受的范圍內即可。
??在很多場景中，我們為了保證數據的最終一致性，需要很多的技術方案來支持，比如分布式事務、分布式鎖等。
??分布式鎖在很多應用場景下是非常有效的手段，比如當運行在多個機器上的不同進程需要訪問同一個競爭資源的時候，那么就會涉及到進程對資源的加鎖和釋放，這樣才能保證數據的安全訪問。分布式鎖實現的方案有很多，比如基于ZooKeeper實現、或者基于Mysql實現等等，今天我們來一起看看如何基于Redis實現分布式鎖服務。
一、分布式鎖的要點：
1、任何一個時間點必須只能夠有一個客戶端擁有鎖。
2、不能夠有死鎖，也就是最終客戶端都能夠獲得鎖，盡管可能會經歷失敗。
3、錯誤容忍性要好，只要有大部分的redis實例存活客戶端就應該能夠獲得鎖。
二、使用分布式鎖要滿足的幾個條件：
1、系統是一個分布式系統（關鍵是分布式，單機的可以使用ReentrantLock或者synchronized代碼塊來實現）；
2、共享資源（各個系統訪問同一個資源，資源的載體可能是傳統關系型數據庫或者NoSQL）；
3、同步訪問（即有很多個進程同時訪問同一個共享資源。沒有同步訪問，誰管你資源競爭不競爭）。
三、應用場景實例：
??多臺服務器要訪問redis全局緩存的資源，如果不使用分布式鎖就會出現問題。
long N=0L; //N從redis獲取值 if(N<5){ N++； //N寫回redis }上面的代碼主要實現以下功能：
??從redis獲取值N，對數值N進行邊界檢查，如果小于5就自增1，然后寫回到redis中。這種應用場景很常見，像秒殺，全局遞增ID，IP訪問限制等。以IP訪問限制來說，惡意攻擊者可能發起無限次訪問，并發量比較大，分布式環境下對N的邊界檢查就不可靠，因為從redis讀的N可能已經是臟數據了，傳統的加鎖做法（如Java的synchronized和Lock）也沒用，因為這是分布式環境，這時我們就要考慮使用分布式鎖了。
四、使用redis的setNX實現分布式鎖：
分布式鎖的基本原理：用一個狀態值表示鎖，對鎖的占用和釋放通過狀態值來標識。
redis實現分布式鎖的原理：redis為單進程單線程模式，采用隊列模式將并發訪問變成串行訪問，且多客戶端對redis的連接并不存在競爭關系。redis的SETNX命令可以方便的實現分布式鎖。
1、基本命令：
1）、setNX（SET IF NOT eXists）
語法：SETNX key value
將key的值設為value，當且僅當key不存在；若給定的key已經存在，則SETNX不做任何動作。
返回值：設置成功返回1；設置失敗返回0.
示例：
redis> EXISTS job # job 不存在 (integer) 0 redis> SETNX job "programmer" # job 設置成功 (integer) 1 redis> SETNX job "code-farmer" # 嘗試覆蓋 job ，失敗 (integer) 0 redis> GET job # 沒有被覆蓋 "programmer"所以我們使用執行下面的命令：
SETNX lock.foo <current Unix time + lock timeout + 1>（current Unix time + lock timeout為該鎖的到期時間）
返回1，則該客戶端獲得鎖，把lock.foo的鍵值設置為時間值表示該鍵已經被鎖定，該客戶端最后可以通過DEL lock.foo來釋放鎖。
返回0，表明該鎖已經被其它客戶端取得，這時我們可以先返回或進行重試等對方完成或等待鎖超時。
2）、getSET
語法：GETSET key value
將給定key的值設為value，并返回key的舊值（old value）。當key存在但不是字符串類型時，返回一個錯誤。
返回值：返回給定key的舊值。當key沒有舊值時，也即是key不存在時，返回nil。
3）、get
語法：GET key
返回值：當key不存在時，返回nil，否則返回key的值；如果key不是字符串類型，返回一個錯誤。
4）、expire
語法：expire key timeout
為key設置一個超時時間，單位為second，超過這個時間鎖會自動釋放，避免死鎖。
5）、delete
語法：delete key
刪除key
2、解決死鎖：
??上面的鎖定邏輯有一個問題：如果一個持有鎖的客戶端失敗或者崩潰了不能釋放鎖，該怎么解決？
??我們可以通過鎖對應的時間戳來判斷這種情況是否發生了，如果當前的時間已經大于lock.foo的值（鎖的到期時間），說明該鎖已經失效，可以被重新使用。
??發生這種情況時，可不能簡單的通過DEL刪除鎖，然后再SETNX一次（刪除鎖的操作應該是鎖擁有者來執行的，這里只需要等它超時即可），當多個客戶端檢測到鎖超時后都會嘗試去釋放它，這里就可能出現一個競態條件，讓我們來模擬一下這個場景：
C0操作超時了，但它還持有著鎖，C1和C2讀取lock.foo檢查時間戳，先后發現超時了。
C1發送DEL lock.foo
C1發送SETNX lock.foo并且成功了；
C2發送DEL lock.foo
C2發送SETNX lock.foo并且成功了；
這樣一來，C1，C2都拿到鎖，就出問題了。
幸好這種問題是可以避免的，讓我們來看看C3這個客戶端是怎樣做的：
C3發送SETNX lock.foo想要獲得鎖，由于C0還持有鎖，所以redis返回給C3一個0；
C3發送GET lock.foo以檢查鎖是否超時了，如果沒有超時，則等待或者重試；
反之，如果已經超時，C3通過下面的操作來嘗試獲得鎖：
GETSET lock.foo
??SET NX命令只會在key不存在的時候給key賦值，PX命令通知redis保存這個key為30000MS。
??key的值會被設置為my_random_value。這個值在多個客戶端和鎖中必須是唯一的，我們使用random value是為了方便安全地釋放鎖，通過下面的腳本為申請成功的鎖解鎖:
如果key對應的value一致，則刪除這個key。通過這個方式釋放鎖是為了避免client釋放其它client申請的鎖。
例如：
1、Client A 獲得了一個鎖
2、當嘗試釋放鎖的請求發送給Redis時被阻塞，沒有及時到達Redis
3、鎖定時間超時，Redis認為鎖的租約到期，釋放了這個鎖
4、client B 重新申請到了這個鎖
5、client A的解鎖請求到達,將Client B鎖定的key解鎖
6、Client C 也獲得了鎖
7、Client B client C 同時持有鎖
通過執行上面腳本的方式釋放鎖，Client的解鎖操作只會解鎖自己曾經加鎖的資源。
官方推薦通從 /dev/urandom/中取20個byte作為隨機數或者采用更加簡單的方式， 例如使用RC4加密算法在/dev/urandom中得到一個種子(Seed)，然后生成一個偽隨機流。
也可以用更簡單的使用時間戳+客戶端編號的方式生成隨機數，這種方式的安全性較差一些，但是對于絕大多數的場景來說也已經足夠安全了。
PX 操作后面的參數代表的是這key的存活時間，稱作鎖過期時間。當資源被鎖定超過這個時間，鎖將自動釋放。
獲得鎖的客戶端如果沒有在這個時間窗口內完成操作,就可能會有其他客戶端獲得鎖，引起爭用問題。
通過上面的兩個操作，我們可以完成獲得鎖和釋放鎖操作。如果這個系統不宕機，那么單點的鎖服務已經足夠安全,接下來我們開始把場景擴展到分布式系統。
5、基于Redlock算法在redis集群上實現分布式鎖
??在分布式環境下，假設我們有N個master，這些節點都是獨立的，因此我們沒有配置復制策略。上面我們已經學會了如何在單機環境下獲取鎖和釋放鎖，我們假設的更具體一些，N=5，為了能獲取鎖，客戶端的步驟為：
1、得到本地時間
2、Client使用相同的key和隨機數,按照順序在每個Master實例中嘗試獲得鎖。在獲得鎖的過程中，為每一個鎖操作設置一個快速失敗時間(如果想要獲得一個10秒的鎖， 那么每一個鎖操作的失敗時間設為5-50ms)。這樣可以避免客戶端與一個已經故障的Master通信占用太長時間，通過快速失敗的方式盡快的與集群中的其他節點完成鎖操作。
3、客戶端計算出與master獲得鎖操作過程中消耗的時間，當且僅當Client獲得鎖消耗的時間小于鎖的存活時間，并且在一半以上的master節點中獲得鎖。才認為client成功的獲得了鎖。
4、如果已經獲得了鎖，Client執行任務的時間窗口是鎖的存活時間減去獲得鎖消耗的時間。
5、如果Client獲得鎖的數量不足一半以上，或獲得鎖的時間超時，那么認為獲得鎖失敗?？蛻舳诵枰獓L試在所有的master節點中釋放鎖， 即使在第二步中沒有成功獲得該Master節點中的鎖，仍要進行釋放操作
RedLock能保證鎖同步嗎？
這個算法成立的一個條件是：即使集群中沒有同步時鐘，各個進程的時間流逝速度也要大體一致，并且誤差與鎖存活時間相比是比較小的。實際應用中的計算機也能滿足這個條件：各個計算機中間有幾毫秒的時鐘漂移(clock drift)。
失敗重試機制
如果一個Client無法獲得鎖，它將在一個隨機延時后開始重試。使用隨機延時的目的是為了與其他申請同一個鎖的Client錯開申請時間，減少腦裂(split brain)發生的可能性。
三個Client同時嘗試獲得鎖，分別獲得了2,2,1個實例中的鎖，三個鎖請求全部失敗。
一個client在全部Redis實例中完成的申請時間越短，發生腦裂的時間窗口越小。所以比較理想的做法是同時向N個Redis實例發出異步的SET請求。當Client沒有在大多數Master中獲得鎖時，立即釋放已經取得的鎖時非常必要的。(PS.當極端情況發生時，比如獲得了部分鎖以后，client發生網絡故障，無法再釋放鎖資源。那么其他client重新獲得鎖的時間將是鎖的過期時間)。無論Client認為在指定的Master中有沒有獲得鎖，都需要執行釋放鎖操作。
6、RedLock算法安全性分析
我們將從不同的場景分析RedLock算法是否足夠安全。首先我們假設一個client在大多數的Redis實例中取得了鎖,那么:
1、每個實例中的鎖的剩余存活時間相等為TTL。
2、每個鎖請求到達各個Redis實例中的時間有差異。
3、第一個鎖成功請求最先在T1后返回，最后返回的請求在T2后返回。(T1,T2都小于最大失敗時間)
4、并且每個實例之間存在時鐘漂移CLOCK_DRIFT(Time Drift)。
于是，最先被SET的鎖將在TTL-(T2-T1)-CLOCK_DIRFT后自動過期，其他的鎖將在之后陸續過期。
所以可以得到結論：所有的key這段時間內是同時被鎖住的。
在這段時間內，一半以上的Redis實例中這個key都處在被鎖定狀態，其他的客戶端無法獲得這個鎖。
7、鎖的可用性分析(Liveness)
分布式鎖系統的可用性主要依靠以下三種機制
1、鎖的自動釋放(key expire)，最終鎖將被釋放并且被再次申請。
2、客戶端在未申請到鎖以及申請到鎖并完成任務后都將進行釋放鎖的操作，所以大部分情況下都不需要等待到鎖的自動釋放期限，其他client即可重新申請到鎖。
3、假設一個Client在大多數Redis實例中申請鎖請求所成功花費的時間為Tac。那么如果某個Client第一次沒有申請到鎖，需要重試之前，必須等待一段時間T。T需要遠大于Tac。 因為多個Client同時請求鎖資源，他們有可能都無法獲得一半以上的鎖，導致腦裂雙方均失敗。設置較久的重試時間是為了減少腦裂產生的概率。
如果一直持續的發生網絡故障，那么沒有客戶端可以申請到鎖。分布式鎖系統也將無法提供服務直到網絡故障恢復為止。
8、性能，故障恢復與文件同步
用戶使用redis作為鎖服務的主要優勢是性能。其性能的指標有兩個
1、加鎖和解鎖的延遲
2、每秒可以進行多少加鎖和解鎖操作
所以，在客戶端與N個Redis節點通信時，必須使用多路發送的方式(multiplex)，減少通信延時。
為了實現故障恢復還需要考慮數據持久化的問題。
我們還是從某個特定的場景分析：
Redis實例的配置不進行任何持久化,集群中5個實例 M1,M2,M3,M4,M5
client A獲得了M1,M2,M3實例的鎖。
此時M1宕機并重啟。
由于沒有進行持久化，M1重啟后不存在任何KEY
client B獲得M4,M5和重啟后的M1中的鎖。
此時client A 和Client B 同時獲得鎖
如果使用AOF的方式進行持久化，情況會稍好一些。例如我們可以向某個實例發送shutdown和restart命令。即使節點被關閉，EX設置的時間仍在計算，鎖的排他性仍能保證。
但當Redis發生電源瞬斷的情況又會遇到有新的問題出現。如果Redis配置中的進行磁盤持久化的時間是每分鐘進行，那么會有部分key在重新啟動后丟失。
如果為了避免key的丟失，將持久化的設置改為Always,那么性能將大幅度下降。
另一種解決方案是在這臺實例重新啟動后，令其在一定時間內不參與任何加鎖。在間隔了一整個鎖生命周期后，重新參與到鎖服務中。這樣可以保證所有在這臺實例宕機期間內的key都已經過期或被釋放。
延時重啟機制能夠保證Redis即使不使用任何持久化策略，仍能保證鎖的可靠性。但是這種策略可能會犧牲掉一部分可用性。
例如集群中超過半數的實例都宕機了，那么整個分布式鎖系統需要等待一整個鎖有效期的時間才能重新提供鎖服務。
9、 使鎖算法更加可靠：鎖續約
如果Client進行的工作耗時較短，那么可以默認使用一個較小的鎖有效期，然后實現一個鎖續約機制。
當一個Client在工作計算到一半時發現鎖的剩余有效期不足?？梢韵騌edis實例發送續約鎖的Lua腳本。如果Client在一定的期限內(耗間與申請鎖的耗時接近)成功的續約了半數以上的實例，那么續約鎖成功。
為了提高系統的可用性，每個Client申請鎖續約的次數需要有一個最大限制，避免其不斷續約造成該key長時間不可用。
10、Java代碼實現：
實現思想（使用jedis來連接redis）：
1、獲取鎖的時候，使用setnx加鎖，并使用expire命令為鎖添加一個超時時間，超過該事件則自動釋放鎖。鎖的value值為一個隨機生成的UUID，通過該value值在釋放鎖的時候進行判斷；
2、獲取鎖的時候還設置一個獲取的超時時間，若超過這個時間則放棄獲取鎖；
3、釋放鎖的時候，通過UUID來判斷是不是該鎖，若是該鎖，則執行delete操作進行鎖釋放。
分布式鎖的核心代碼如下：
import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; import redis.clients.jedis.Transaction; import redis.clients.jedis.exceptions.JedisException; import java.util.List; import java.util.UUID;public class DistributedLock {private final JedisPool jedisPool;public DistributedLock(JedisPool jedisPool) {this.jedisPool = jedisPool;}/*** 加鎖* @param locaName 鎖的key* @param acquireTimeout 獲取超時時間* @param timeout 鎖的超時時間* @return 鎖標識*/public String lockWithTimeout(String locaName,long acquireTimeout, long timeout) {Jedis conn = null;String retIdentifier = null;try {// 獲取連接conn = jedisPool.getResource();// 隨機生成一個valueString identifier = UUID.randomUUID().toString();// 鎖名，即key值String lockKey = "lock:" + locaName;// 超時時間，上鎖后超過此時間則自動釋放鎖int lockExpire = (int)(timeout / 1000);// 獲取鎖的超時時間，超過這個時間則放棄獲取鎖long end = System.currentTimeMillis() + acquireTimeout;while (System.currentTimeMillis() < end) {if (conn.setnx(lockKey, identifier) == 1) {conn.expire(lockKey, lockExpire);// 返回value值，用于釋放鎖時間確認retIdentifier = identifier;return retIdentifier;}// 返回-1代表key沒有設置超時時間，為key設置一個超時時間if (conn.ttl(lockKey) == -1) {conn.expire(lockKey, lockExpire);}try {Thread.sleep(10);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}} catch (JedisException e) {e.printStackTrace();} finally {if (conn != null) {conn.close();}}return retIdentifier;}/*** 釋放鎖* @param lockName 鎖的key* @param identifier 釋放鎖的標識* @return*/public boolean releaseLock(String lockName, String identifier) {Jedis conn = null;String lockKey = "lock:" + lockName;boolean retFlag = false;try {conn = jedisPool.getResource();while (true) {// 監視lock，準備開始事務conn.watch(lockKey);// 通過前面返回的value值判斷是不是該鎖，若是該鎖，則刪除，釋放鎖if (identifier.equals(conn.get(lockKey))) {Transaction transaction = conn.multi();transaction.del(lockKey);List<Object> results = transaction.exec();if (results == null) {continue;}retFlag = true;}conn.unwatch();break;}} catch (JedisException e) {e.printStackTrace();} finally {if (conn != null) {conn.close();}}return retFlag;} }測試：
模擬秒殺服務，在其中配置了jedis線程池，在初始化的時候傳給分布式鎖，供其使用。使用50個線程模擬秒殺一個商品，使用–運算符來實現商品的減少，從結果有序性可以看出是否為加鎖狀態。
import redis.clients.jedis.JedisPool; import redis.clients.jedis.JedisPoolConfig;public class Service {private static JedisPool pool = null;static {JedisPoolConfig config = new JedisPoolConfig();// 設置最大連接數config.setMaxTotal(200);// 設置最大空閑數config.setMaxIdle(8);// 設置最大等待時間config.setMaxWaitMillis(1000 * 100);// 在borrow一個jedis實例時，是否需要驗證，若為true，則所有jedis實例均是可用的config.setTestOnBorrow(true);pool = new JedisPool(config, "127.0.0.1", 6379, 3000);}DistributedLock lock = new DistributedLock(pool);int n = 500;public void seckill() {// 返回鎖的value值，供釋放鎖時候進行判斷String indentifier = lock.lockWithTimeout("resource", 5000, 1000);System.out.println(Thread.currentThread().getName() + "獲得了鎖");System.out.println(--n);lock.releaseLock("resource", indentifier);} } //模擬線程進行秒殺 public class ThreadA extends Thread {private Service service;public ThreadA(Service service) {this.service = service;}@Overridepublic void run() {service.seckill();} } public class Test {public static void main(String[] args) {Service service = new Service();for (int i = 0; i < 50; i++) {ThreadA threadA = new ThreadA(service);threadA.start();}} } 超強干貨來襲 云風專訪：近40年碼齡，通宵達旦的技術人生總結
以上是生活随笔為你收集整理的【Redis学习】Redis实现分布式锁的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 深入理解WEB请求过程
- 下一篇： 乐观锁与悲观锁深入学习