分布式應用進行邏輯處理時經常會遇到并發問題。?
比如一個操作要修改用戶的狀態，修改狀態需要先讀出用戶的狀態，在內存里進行修改，改完了再存回去。如果這樣的操作同時進行了，就會出現并發問題，因為讀取和保存狀態這兩個操作不是原子的。（Wiki 解釋：所謂原子操作原子操作是指不會被線程調度機制打斷的操作；這種操作一旦開始，就一直運行到結束，中間不會有任何 context switch 線程切換。）

這個時候就要使用到分布式鎖來限制程序的并發執行。Redis 分布式鎖使用非常廣泛，它是面試的重要考點之一，很多同學都知道這個知識，也大致知道分布式鎖的原理，但是具體到細節的使用上往往并不完全正確。?

分布式鎖?

分布式鎖本質上要實現的目標就是在 Redis 里面占一個“茅坑”，當別的進程也要來占時，發現已經有人蹲在那里了，就只好放棄或者稍后再試。?
占坑一般是使用 setnx(set if not exists) 指令，只允許被一個客戶端占坑。先來先占， 用完了，再調用 del 指令釋放茅坑。?

// 這里的冒號:就是一個普通的字符，沒特別含義，它可以是任意其它字符，不要誤解? > setnx lock:codehole true? OK? ... do something critical ...? > del lock:codehole? (integer) 1?

但是有個問題，如果邏輯執行到中間出現異常了，可能會導致 del 指令沒有被調用， 這樣就會陷入死鎖，鎖永遠得不到釋放。?
于是我們在拿到鎖之后，再給鎖加上一個過期時間，比如 5s，這樣即使中間出現異常也可以保證 5 秒之后鎖會自動釋放。?
?

> setnx lock:codehole true? OK? > expire lock:codehole 5? ... do something critical ...? > del lock:codehole? (integer) 1?

但是以上邏輯還有問題。如果在 setnx 和 expire 之間服務器進程突然掛掉了，可能是因為機器掉電或者是被人為殺掉的，就會導致 expire 得不到執行，也會造成死鎖。?
這種問題的根源就在于 setnx 和 expire 是兩條指令而不是原子指令。如果這兩條指令可以一起執行就不會出現問題。 也許你會想到用 Redis 事務來解決。 但是這里不行， 因為 expire?是依賴于 setnx 的執行結果的， 如果 setnx 沒搶到鎖， expire 是不應該執行的。 事務里沒有 if-else 分支邏輯，事務的特點是一口氣執行，要么全部執行要么一個都不執行。?
為了解決這個疑難，Redis 開源社區涌現了一堆分布式鎖的 library，專門用來解決這個問題。 實現方法極為復雜， 小白用戶一般要費很大的精力才可以搞懂。 如果你需要使用分布式鎖，意味著你不能僅僅使用 Jedis 或者 redis-py 就行了，還得引入分布式鎖的 library。?

為了治理這個亂象，Redis 2.8 版本中作者加入了 set 指令的擴展參數，使得 setnx 和?expire 指令可以一起執行，徹底解決了分布式鎖的亂象。從此以后所有的第三方分布式鎖?library 可以休息了。

> set lock:codehole true ex 5 nx OK ... do something critical ... > del?lock:codehole

上面這個指令就是 setnx 和 expire 組合在一起的原子指令，它就是分布式鎖的奧義所在。?

超時問題?

Redis 的分布式鎖不能解決超時問題，如果在加鎖和釋放鎖之間的邏輯執行的太長，以至于超出了鎖的超時限制， 就會出現問題。 因為這時候鎖過期了， 第二個線程重新持有了這把鎖，但是緊接著第一個線程執行完了業務邏輯，就把鎖給釋放了，第三個線程就會在第二個線程邏輯執行完之間拿到了鎖。?
為了避免這個問題，Redis 分布式鎖不要用于較長時間的任務。如果真的偶爾出現了，數據出現的小波錯亂可能需要人工介入解決。?

tag = random.nextint() ?# 隨機數? if redis.set(key, tag, nx=True, ex=5):?do_something()?redis.delifequals(key, tag) ?# 假象的 delifequals 指令

有一個更加安全的方案是為 set 指令的 value 參數設置為一個隨機數，釋放鎖時先匹配隨機數是否一致， 然后再刪除 key。 但是匹配 value 和刪除 key 不是一個原子操作， Redis 也沒有提供類似于 delifequals 這樣的指令，這就需要使用 Lua 腳本來處理了，因為 Lua 腳本可以保證連續多個指令的原子性執行。??

# delifequals? if redis.call("get",KEYS[1]) == ARGV[1] then?return redis.call("del",KEYS[1])? else?return 0? end?

可重入性?

可重入性是指線程在持有鎖的情況下再次請求加鎖， 如果一個鎖支持同一個線程的多次加鎖， 那么這個鎖就是可重入的。 比如 Java 語言里有個 ReentrantLock 就是可重入鎖。 Redis 分布式鎖如果要支持可重入，需要對客戶端的 set 方法進行包裝，使用線程的 Threadlocal 變量
存儲當前持有鎖的計數。?

# -*- coding: utf-8 ? import redis ? import threading ?locks = threading.local() ? locks.redis = {} ?def key_for(user_id): ? return "account_{}".format(user_id) ?def _lock(client, key): ? return bool(client.set(key, True, nx=True, ex=5)) ?def _unlock(client, key): ? client.delete(key) ?def lock(client, user_id): ? key = key_for(user_id) ? if key in locks.redis:? locks.redis[key] += 1 ? return True ? ok = _lock(client, key) ? if not ok: ? return False ? locks.redis[key] = 1 ? return True ?def unlock(client, user_id): ? key = key_for(user_id) ? if key in locks.redis: ? locks.redis[key] -= 1 if locks.redis[key] <= 0: ? del locks.redis[key] ? return True ? return False ?client = redis.StrictRedis() ? print "lock", lock(client, "codehole") ? print "lock", lock(client, "codehole") ? print "unlock", unlock(client, "codehole") ? print "unlock", unlock(client, "codehole") ?

以上還不是可重入鎖的全部，精確一點還需要考慮內存鎖計數的過期時間，代碼復雜度將會繼續升高。老錢不推薦使用可重入鎖，它加重了客戶端的復雜性，在編寫業務方法時注意在邏輯結構上進行調整完全可以不使用可重入鎖。下面是 Java 版本的可重入鎖。?

public class RedisWithReentrantLock {private ThreadLocal<Map> lockers = new ThreadLocal<>();private Jedis jedis;public RedisWithReentrantLock(Jedis jedis) {this.jedis = jedis;}private boolean _lock(String key) {return jedis.set(key, "", "nx", "ex", 5L) != null;}private void _unlock(String key) {jedis.del(key);}private Map<String, Integer> currentLockers() {Map<String, Integer> refs = lockers.get();if (refs != null) {return refs;}lockers.set(new HashMap<>());return lockers.get();}public boolean lock(String key) {Map refs = currentLockers();Integer refCnt = refs.get(key);if (refCnt != null) {refs.put(key, refCnt + 1);?return true;}boolean ok = this._lock(key);if (!ok) {return false;}refs.put(key, 1);return true;}public boolean unlock(String key) {Map refs = currentLockers();?Integer refCnt = refs.get(key);if (refCnt == null) {return false;}refCnt -= 1;if (refCnt > 0) {refs.put(key, refCnt);} else {refs.remove(key);?this._unlock(key);}return true;}public static void main(String[] args) {Jedis jedis = new Jedis();RedisWithReentrantLock redis = new RedisWithReentrantLock(jedis);System.out.println(redis.lock("codehole"));System.out.println(redis.lock("codehole"));System.out.println(redis.unlock("codehole"));System.out.println(redis.unlock("codehole"));} }?

跟 Python 版本區別不大，也是基于 ThreadLocal 和引用計數。?
以上還不是分布式鎖的全部，在小冊的拓展篇《拾遺漏補 —— 再談分布式鎖》，我們還會繼續對分布式鎖做進一步的深入理解。?

思考思考題?

1、Review 下你自己的項目代碼中的分布式鎖，它的使用方式是否標準正確？?
2、如果你還沒用過分布式鎖，想想自己的項目中是否可以用上？?

總結

以上是生活随笔為你收集整理的【Redis核心原理和应用实践】应用 1：千帆竞发 —— 分布式锁的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。