使用 Redis 實現分佈式鎖（轉載）

時間 2019-12-14

原文原文鏈接

背景

在通常的分佈式應用中，要安全有效地同步多服務器多進程之間的共享資源訪問，就要涉及到分佈式鎖。目前項目是基於 Tornado 實現的分佈式部署，同時也使用了 Redis 做爲緩存。參考了一些資料並結合項目自身的要求後，決定直接使用Redis實現全局的分佈式鎖。python

使用 Redis 實現分佈式鎖

使用 Redis 實現分佈式鎖最簡單方式是建立一對 key-value 值，key 被建立爲有必定的生存期，所以它最終會被釋放。而當客戶端想要釋放時，則直接刪除 key 。基於不一樣的 Redis 命令，有兩種實現方式：git

Redis 官方早期給的一個實現，使用 SETNX，將 value 設置爲超時時間，由代碼實現鎖超時的檢測[有缺陷，有限制，併發不高時可用]；
有同窗本身的實現：使用 INCR + EXPIRE，利用 Redis 的超時機制控制鎖的生存期[不建議使用]；
Redis 官方給的一個改進實現：使用 SET resource-name anystring NX EX max-lock-time（Redis 2.6.12 後支持）實現，利用 Redis 的超時機制控制鎖的生存期[Redis 2.6.12 之後建議使用]。

使用 SETNX 實現

Redis 官方最先在 SETNX 命令頁給了一個基於該命令的分佈式鎖實現。github

1	Acquire lock: SETNX lock.foo <current Unix time + lock timeout + 1>

1	Release lock: DEL lock.foo

若是 SETNX 返回 1，則代表客戶端獲取鎖成功， lock.foo 被設置爲有效 Unix time。客戶端操做完成後調用 DEL 命令釋放鎖。redis
若是 SETNX 返回 0，則代表鎖已經被其餘客戶端持有。這時咱們能夠先返回或進行重試等對方完成或等待鎖超時。算法

處理死鎖問題：
上述算法中，若是持有鎖的客戶端發生故障、意外崩潰、或者其餘因素因素致使沒有釋放鎖，該怎麼解決？。咱們能夠經過鎖的鍵對應的時間戳來判斷這種狀況是否發生了，若是當前的時間已經大於lock.foo的值，說明該鎖已失效，能夠被從新使用。
發生這種狀況時，可不能簡單的經過DEL來刪除鎖，而後再SETNX一次，當多個客戶端檢測到鎖超時後都會嘗試去釋放它，這裏就可能出現一個競態條件：緩存

C1 和 C2 讀取 lock.foo 檢查時間戳，前後發現超時了。
C1 發送DEL lock.foo。
C1 發送SETNX lock.foo 而且成功了。
C2 發送DEL lock.foo
C2 發送SETNX lock.foo 而且成功了。
ERROR: 因爲競態的問題，C1 和 C2 都獲取了鎖，這下子問題大了。

幸運的是，使用下面的算法能夠避免這個問題。咱們看看客戶端 C4 是怎麼作的：安全

C4 發送 SETNX lock.foo 想要獲取鎖。
可是因爲發生故障的客戶端 C3 仍然持有鎖，因此返回 0 給 C4。
C4 發送 GET lock.foo 來檢查鎖是否過時，若是沒超時，則等待或重試。

反之，若是已經超時， C4 則嘗試執行下面的命令來獲取鎖:服務器

1	Acquire lock when time expired: GETSET lock.foo <current Unix timestamp + lock timeout + 1>

經過 GETSET ，C4 拿到的時間戳若是仍然是超時的，那就代表 C4 如願以償拿到鎖了。併發
若是在 C4 以前，有個叫 C5 的客戶端比 C4 快一步執行了上面的操做，那麼 C4 拿到的時間戳是個未超時的值，這時，C4 沒有如期得到鎖，須要再次等待或重試。留意一下，儘管 C4 沒拿到鎖，但它改寫了 C5 設置的鎖的超時值，可是這點微小的偏差（通常狀況下鎖的持有的時間很是短，因此在該競態下出現的偏差是能夠容忍的）是能夠容忍的。（Note that even if C4 set the key a bit a few seconds in the future this is not a problem）。

爲了這個鎖的算法更健壯一些，持有鎖的客戶端在解鎖以前應該再檢查一次本身的鎖是沒有超時，再去作 DEL 操做，由於客戶端失敗的緣由很複雜，不只僅是崩潰也多是由於某個耗時的操做而掛起，操做完的時候鎖由於超時已經鎖已經被別人得到，這時就沒必要解鎖了。app

仔細的分析這個方案，咱們就會發現這裏有一個漏洞：Release lock 使用的 DEL 命令不支持 CAS 刪除（check-and-set，delete if current value equals old value），在高併發狀況下就會有一些問題：確認持有的鎖沒有超時後執行 DEL 釋放鎖，因爲競態的存在 Redis 服務器執行命令時鎖可能已過時（「真的」恰好過時或者被其餘客戶端競爭鎖時設置了一個較小的過時時間而致使過時）且被其餘客戶端持有。這種狀況下將會（非法）釋放其餘客戶端持有的鎖。

解決方案： 先肯定鎖沒有超時，再經過 EVAL 命令（在 Redis 2.6 及以上版本提供）在執行 Lua 腳本：先執行 GET 指令獲取鎖的時間戳，確認和本身的時間戳一致後再執行 DEL 釋放鎖。

設計缺陷：

上述解決方案並不完美，只解決了過時鎖的釋放問題，可是因爲這個方案自己的缺陷，客戶端獲取鎖時發生競爭（C4 改寫 C5 時間戳的例子），那麼 lock.foo 的「時間戳」將與本地的不一致，這個時候不會執行 DEL 命令，而是等待鎖失效，這在高併發的環境下是低效的。
考慮多服務器環境下，須要服務器進行時間同步校準。

在咱們的項目中使用了 tornadoredis 庫，這個庫實現的分佈式鎖便採用了上述算法。可是在釋放鎖時有些限制，不過併發量不高的狀況下不會有太大的問題，詳細的分析參考下述代碼註釋。實現代碼以下所示：

class Lock(object):
    """
 A shared, distributed Lock that uses a Redis server to hold its state.
 This Lock can be shared across processes and/or machines. It works
 asynchronously and plays nice with the Tornado IOLoop.
 """

    LOCK_FOREVER = float(2 ** 31 + 1)  # 1 past max unix time

    def __init__(self, redis_client, lock_name, lock_ttl=None, polling_interval=0.1):
        """
 Create a new Lock object using the Redis key ``lock_name`` for
 state, that behaves like a threading.Lock.
 This method is synchronous, and returns immediately. It doesn't acquire the
 Lock or in fact trigger any sort of communications with the Redis server.
 This must be done using the Lock object itself.
 If specified, ``lock_ttl`` indicates the maximum life time for the lock.
 If none is specified, it will remain locked until release() is called.
 ``polling_interval`` indicates the time between acquire attempts (polling)
 when the lock is in blocking mode and another client is currently
 holding the lock.
 Note: If using ``lock_ttl``, you should make sure all the hosts
 that are running clients have their time synchronized with a network
 time service like ntp.
 """
        self.redis_client = redis_client
        self.lock_name = lock_name
        self.acquired_until = None
        self.lock_ttl = lock_ttl
        self.polling_interval = polling_interval
        if self.lock_ttl and self.polling_interval > self.lock_ttl:
            raise LockError("'polling_interval' must be less than 'lock_ttl'")

    @gen.engine
    def acquire(self, blocking=True, callback=None):
        """
 Acquire the lock.
 Returns True once the lock is acquired.
 If ``blocking`` is False, always return immediately. If the lock
 was acquired, return True, otherwise return False.
 Otherwise, block until the lock is acquired (or an error occurs).
 If ``callback`` is supplied, it is called with the result.
 """

        # Loop until we have a conclusive result
        while 1:

            # Get the current time
            unixtime = int(mod_time.time())

            # If the lock has a limited lifetime, create a timeout value
            if self.lock_ttl:
                timeout_at = unixtime + self.lock_ttl
            # Otherwise, set the timeout value at forever (dangerous)
            else:
                timeout_at = Lock.LOCK_FOREVER
            timeout_at = float(timeout_at)

            # Try and get the lock, setting the timeout value in the appropriate key,
            # but only if a previous value does not exist in Redis
            result = yield gen.Task(self.redis_client.setnx, self.lock_name, timeout_at)

            # If we managed to get the lock
            if result:

                # We successfully acquired the lock!
                self.acquired_until = timeout_at
                if callback:
                    callback(True)
                return

            # We didn't get the lock, another value is already there
            # Check to see if the current lock timeout value has already expired
            result = yield gen.Task(self.redis_client.get, self.lock_name)
            existing = float(result or 1)

            # Has it expired?
            if existing < unixtime:

                # The previous lock is expired. We attempt to overwrite it, getting the current value
                # in the server, just in case someone tried to get the lock at the same time
                result = yield gen.Task(self.redis_client.getset,
                                        self.lock_name,
                                        timeout_at)
                existing = float(result or 1)

                # If the value we read is older than our own current timestamp, we managed to get the
                # lock with no issues - the timeout has indeed expired
                if existing < unixtime:

                    # We successfully acquired the lock!
                    self.acquired_until = timeout_at
                    if callback:
                        callback(True)
                    return

                # However, if we got here, then the value read from the Redis server is newer than
                # our own current timestamp - meaning someone already got the lock before us.
                # We failed getting the lock.

            # If we are not signalled to block
            if not blocking:

                # We failed acquiring the lock...
                if callback:
                    callback(False)
                return

            # Otherwise, we "sleep" for an amount of time equal to the polling interval, after which
            # we will try getting the lock again.
            yield gen.Task(self.redis_client._io_loop.add_timeout,
                           self.redis_client._io_loop.time() + self.polling_interval)

    @gen.engine
    def release(self, callback=None):
        """
 Releases the already acquired lock.
 If ``callback`` is supplied, it is called with True when finished.
 """

        if self.acquired_until is None:
            raise ValueError("Cannot release an unlocked lock")

        # Get the current lock value
        result = yield gen.Task(self.redis_client.get, self.lock_name)
        existing = float(result or 1)

        # 從上下文代碼中能夠看出，在這個實現中，有一個限制：獲取鎖的時候設置的 lock_ttl 必須可以保證釋放鎖時，鎖未過時。
        # 不然，當前鎖過時後，將會非法釋放其餘客戶端持有的鎖。若是沒法估計持有鎖後代碼的執行時間，則能夠增長當前鎖的過時檢測，
        # 當 self.acquired_until <= int(mod_time.time()) 時不執行 DEL 命令。不過，這個限制在通常的應用中卻是能夠知足，
        # 因此這個實現不會有太大的問題。
        # 因爲 GET、DEL 之間的時間差，以及 DEL 命令發出到 執行 之間的時間差，高併發狀況下，鎖過時釋放的問題依然存在，這個是
        # 算法缺陷。併發不大的狀況下，問題不大。
        #
        # 注：這個條件判斷 existing >= self.acquired_until 是有這樣一個潛在的前提，使用鎖的客戶端代碼正常運行的狀況下，
        # 考慮到併發代碼使用相同的 lock_ttl 獲取鎖，競爭失敗的客戶端將會把鎖的過時時間設置的更長一些，這裏的判斷是有意義的。
        # If the lock time is in the future, delete the lock
        if existing >= self.acquired_until:
            yield gen.Task(self.redis_client.delete, self.lock_name)
        self.acquired_until = None

        # That is it.
        if callback:
            callback(True)

使用 INCR + EXPIRE 實現

該方案的實現來源這篇 blog 《Redis實現分佈式全局鎖》

客戶端A經過 INCR locker.foo 獲取名爲 locker.foo 的鎖，若獲取的值爲1，則表示獲取成功，轉入下一步，不然獲取失敗；
執行 EXPIRE locker.foo seconds 設置鎖的過時時間，設置成功轉入下一步；
執行共享資源訪問；
執行 DEL locker.foo 釋放鎖。

僞代碼以下所示：

if(INCR('locker.foo') == 1)
{
     // 設置鎖的超時時間爲1分鐘，這個能夠設置爲一個較大的值來避免鎖提早過時釋放。
     EXPIRE(60)

     // 執行共享資源訪問
	 DO_SOMETHING()

     // 釋放鎖
     DEL('locker.foo')}
}

該實現有一個嚴重的「死鎖」問題：若是 INCR 命令獲取鎖成功後，EXPIRE 失敗，會致使鎖沒法正常釋放。可用的解決方案是：藉助 EVAL 命令，將 INCR 、 EXPIRE 操做封裝在一個 Lua 腳本中執行，先執行 INCR 命令，成功獲取鎖後再執行 EXPIRE。如下是示例 Lua 代碼：

-- Set a lock
-- KEYS[1] - key
-- KEYS[2] - ttl in ms

local key     = KEYS[1]
local ttl     = KEYS[2]

local lockSet = redis.call('incr', key)

if lockSet == 1 then
  redis.call('pexpire', key, ttl)
end

return lockSet

注：因爲 EVAL 命令僅在 Redis 2.6 版本後提供，對於以前的版本只能經過 MULTI/EXEC 將 INCR 、 EXPIRE 封裝在一個事務中來處理。可是因爲 MULTI/EXEC 的限制，沒有辦法和使用 Lua 腳本同樣根據 INCR 執行結果來執行 EXPIRE ，因此若是獲取鎖失敗，會致使 TTL 不斷被延長，在高併發的環境裏若是拿到鎖的進程意外掛掉而沒有正常釋放鎖，鎖便只能等到過時才能被其餘客戶端持有，而這個過時時間的長短取決於獲取鎖時的競爭激烈狀況。該解決方案有嚴重缺陷，不適合高併發環境。

++實際上，因爲不能經過一個原語完成獲取鎖和設置鎖過時時間的操做，即便經過上述 Lua 腳原本獲取鎖，仍然是有問題的。因爲 Redis 事務的特色，只保證 INCR 、 EXPIRE 兩條命令在 Redis 上是連續執行的，但當 EXPIRE 命令失敗後並不會回滾 INCR 命令，因此「死鎖」問題依然沒有解決（取決於 Redis 的穩定性）。同時，也存在鎖過時後非法釋放其餘客戶端持有的鎖的問題，且因爲依賴 redis 的自動過時機制，便沒法檢測到此問題。++

使用 SET resource-name anystring NX EX max-lock-time 實現

該方案在 Redis 官方 SET 命令頁有詳細介紹。
在介紹該分佈式鎖設計以前，咱們先來看一下在從 Redis 2.6.12 開始 SET 提供的新特性，命令 SET key value [EX seconds] [PX milliseconds] [NX|XX]，其中：

EX seconds — 以秒爲單位設置 key 的過時時間；
PX milliseconds — 以毫秒爲單位設置 key 的過時時間；
NX — 將key 的值設爲value ，當且僅當key 不存在，等效於 SETNX。
XX — 將key 的值設爲value ，當且僅當key 存在，等效於 SETEX。

注：因爲 SET 已經可以取代 SETNX, SETEX, PSETEX 命令，因此在將來的版本中，官方將逐漸放棄這3個命令，並最終移除。

使用 SET 的新特性，改進舊版的分佈式鎖設計，主要有兩個優化：

客戶端經過 SET 命令能夠同時完成獲取鎖和設置鎖的過時時間：SET lock.foo token NX EX max-lock-time（原子操做，沒有INCR 、 EXPIRE兩個操做的事務問題），鎖將在超時後自動過時，不擔憂以前設計的「死鎖」問題，也沒有多服務器時間同步校準的問題。
使用 Lua 腳本實現 CAS 刪除，使鎖更健壯。獲取鎖時爲鎖設置一個 token （一個沒法猜想的隨機字符串），釋放鎖時先比較 token 的值以保證只釋放持有的有效鎖。釋放鎖的 Lua 代碼示例：
1

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