最近專案中需要用到一個分散式的鎖，考慮到基於會話節點實現的zookeeper鎖效能不夠，於是想使用redis來實現一個分散式的鎖。看了網上的幾個實現方案後，發現都不夠嚴謹。比如這篇：用Redis實現分散式鎖裡面設計的鎖有個最大的問題是鎖的超時值TTL會一直被改寫，“儘管C3沒拿到鎖，但它改寫了C4設定的鎖的超時值，不過這一點非常微小的誤差帶來的影響可以忽略不計”，其實在高併發的時候會導致程序“餓死”(也有文章稱為死鎖)。還有這篇文章“兩種分散式鎖實現方案2”裡面的v2=getset(key,時間戮+超時+1)，其加1秒操作在大併發下也會觸發同樣的問題。網上這篇文章解決了這個“無休止的TTL”問題，我簡單翻譯了下。

鎖是程式設計中非常常見的概念。在維基百科上對鎖有個相當精確的定義：

在電腦科學中，鎖是一種在多執行緒環境中用於強行限制資源訪問的同步機制。鎖被設計用於執行一個互斥的併發控制策略。

In computer science, a lock is a synchronization mechanism for
enforcing limits on access to a resource in an environment where there
are many threads of execution. A lock is designed to enforce a mutual
exclusion concurrency control policy.

簡單的說，鎖是一個單一的參考點，多個執行緒基於它來檢查是否允許訪問資源。例如，一個想寫資料的執行緒，它必須先檢查是否存在一個寫鎖。如果寫鎖存在，需要等待直到鎖釋放後它才能獲取到屬於它的鎖並執行寫操作。這樣，通過鎖就可以避免多個執行緒的同時寫造成的資料衝突。

現代的作業系統提供了內建的函式來幫助程式設計師實現併發控制，例如 flock 函式。但是如果多執行緒的程式執行在多臺機器上呢？如何在分散式系統下控制對資源的訪問呢？

使用一箇中心化的鎖服務

首先，我們需要一個所有執行緒都可以訪問到的地方來儲存鎖。這個鎖只能存在於一個地方，從而保證只有一個權威的地方可以定義鎖的建立和釋放。

Redis是實現鎖的一個理想的候選方案。作為一個輕量級的記憶體資料庫，快速，事務性和一致性是選擇redis所為鎖服務的主要原因。

設計鎖

鎖本身是很簡單的，就是redis資料庫中一個簡單的key。建立和釋放鎖，並保證絕對的安全，是這個鎖的設計比較棘手的地方。有兩個潛在的陷阱：

1. 應用程式通過網路和redis互動，這意味著從應用程式發出命令到redis結果返回之間會有延遲。這段時間內，redis可能正在執行其他的命令，而redis內資料的狀態可能不是你的程式所期待的。如果保證程式中獲取鎖的執行緒和其他執行緒不發生衝突？

2. 如果程式在獲取鎖後突然crash，而無法釋放它？這個鎖會一直存在而導致程式進入“餓死”(原文成為“死鎖”，感覺不太準確)。

建立鎖

可能想到的最簡單的方法是“用GET方法檢查鎖，如果鎖不存在，就用SET方式設定一個值”。

這個方法雖然簡單，但是不能保證獨佔鎖。回顧前面所說的第1個陷阱：因為在GET和SET操作之間有延遲，我們沒法知道從“傳送命令”到“redis伺服器返回結果”之間的這段時間內是否有其他執行緒也去建立鎖。當然，這些都在幾毫秒之內，發生的可能性相當低。但是如果在一個繁忙的環境中執行著大量的併發執行緒和命令，重疊的可能性並不是微不足道的。

為了解決這個問題，應該用SETNX命令。SETNX消除了GET命令需要等待返回值的問題，SETNX只有在key不存在時才返回成功。這意味著只有一個執行緒可以成功執行SETNX命令，而其他執行緒會失敗，然後不斷重試，直到它們能建立鎖。

釋放鎖

一旦執行緒成功執行了SETNX命令，它就建立了鎖並且可以基於資源進行工作。工作完成後，執行緒需要通過刪除redis的key來釋放這個鎖，從而允許其他執行緒能儘快的獲取鎖。

儘管如此，也有需要小心的地方！回顧前面說的第2個陷阱：如果執行緒crash了，它永遠都不會刪除redis的key，所以這個鎖會一直存在，從而導致“餓死”現象。那麼如何避免這個問題呢？

鎖的存活時間

我們可以給鎖加一個存活時間(TTL)，這樣一旦TTL超時，這個鎖的key會被redis自動刪除。任何由於執行緒錯誤而遺留下來的鎖在一個合適的時間之後都會被釋放，從而避免了“餓死”。這純粹是一個安全特性，更有效的方式仍然是確保儘量線上程裡面釋放鎖。

可以通過PEXPIRE命令為Redis的key設定TTL，而且執行緒裡可以通過MULTI/EXEC事務的方式在SETNX命令後立即執行，例如：

MULTI
SETNX lock-key
PEXPIRE 10000 lock-key
EXEC

儘管如此，這會產生另外一個問題。PEXPIRE命令沒有判斷SETNX命令的返回結果，無論如何都會設定key的TTL。如果這個地方無法獲取到鎖或有異常，那麼多個執行緒每次想獲取鎖時，都會頻繁更新key的TTL，這樣會一直延長key的TTL，導致key永遠都不會過期。為了解決這個問題，我們需要Redis在一個命令裡面處理這個邏輯。我們可以通過Redis指令碼的方式來實現。

注意-如果不採用指令碼的方式來實現，可以使用Redis 2.6.12之後版本SET命令的PX和NX引數來實現。為了考慮相容2.6.0之前的版本，我們還是採用指令碼的方式來實現。

Redis指令碼

由於Redis支援指令碼，我們可以寫一個Lua指令碼在Redis服務端執行多個Redis命令。應用程式通過一條EVALSHA命令就可以呼叫被Redis服務端快取的指令碼。這裡強大的地方在於你的程式只需要執行一條命令(指令碼)就可以以事務的方式執行多個redis命令，還能避免併發衝突，因為一個redis指令碼同一時刻只能執行一次。

這是Redis裡面一個設定帶TTL的鎖的Lua指令碼：

--
-- Set a lock
--
-- KEYS[1]   - key
-- KEYS[2]   - ttl in ms
-- KEYS[3]   - lock content
local key     = KEYS[1]
local ttl     = KEYS[2]
local content = KEYS[3]

local lockSet = redis.call('setnx', key, content)

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

return lockSet

從這個指令碼可以很清楚的看到，我們通過在鎖上只執行PEXPIRE命令就解決了前面提到的“無休止的TTL”問題。

Warlock: 一個成熟的基於Redis的分散式鎖

上面我們介紹了基於Redis的鎖的理論，這裡有一個用Node.js寫的開源模組Warlock，通過npm可以獲取。它使用了redis指令碼來建立/釋放鎖，用於為快取，資料庫，任務佇列和其他需要併發的地方提供分散式的鎖服務，詳見Github。

原文中還缺少一個釋放鎖的指令碼，如果一直依賴TTL來釋放鎖，效率會很低。Redis的SET操作文件就提供了一個釋放鎖的指令碼：

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

應用程式中只要加鎖的時候指定一個隨機數或特定的value作為key的值，解鎖的時候用這個value去解鎖就可以了。當然，每次加鎖時的value必須要保證是唯一的。

下面是基於這篇文章，使用Java程式碼來實現分散式鎖，redis客戶端使用RedisTemplate

import java.util.Collections;

import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;

/**
 * redis分散式鎖.<br>
 * 思路：
 * <pre>
 * 用SETNX命令，SETNX只有在key不存在時才返回成功。這意味著只有一個執行緒可以成功執行SETNX命令，而其他執行緒會失敗，然後不斷重試，直到它們能建立鎖。
 * 然後使用指令碼來建立鎖，因為一個redis指令碼同一時刻只能執行一次。
 * 建立鎖程式碼：
 * <code>
-- KEYS[1] key,
-- ARGV[1] value,
-- ARGV[2] expireTimeMilliseconds

if redis.call('setnx', KEYS[1], ARGV[1]) == 1 then 
    redis.call('pexpire', KEYS[1], ARGV[2]) 
    return 1 
else 
    return 0 
end
 * </code>
 * 最後使用指令碼來解鎖。
 * 解鎖程式碼：
 * 
 * <code>
-- KEYS[1] key,
-- ARGV[1] value
if redis.call("get", KEYS[1]) == ARGV[1]
then
    return redis.call("del", KEYS[1])
else
    return 0
end
 * </code>
 * </pre>
 * 
 * @author tanghc
 */
public class RedisLockUtil {

    private static final Long SUCCESS = 1L;

    // 加鎖指令碼
    private static final String SCRIPT_LOCK = "if redis.call('setnx', KEYS[1], ARGV[1]) == 1 then redis.call('pexpire', KEYS[1], ARGV[2]) return 1 else return 0 end";
    // 解鎖指令碼
    private static final String SCRIPT_UNLOCK = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    // 加鎖指令碼sha1值
    private static final String SCRIPT_LOCK_SHA1 = Sha1Util.encrypt(SCRIPT_LOCK);
    // 解鎖指令碼sha1值
    private static final String SCRIPT_UNLOCK_SHA1 = Sha1Util.encrypt(SCRIPT_UNLOCK);

    /**
     * 嘗試獲取分散式鎖
     * 
     * @param redisTemplate
     *            Redis客戶端
     * @param lockKey
     *            鎖
     * @param requestId
     *            請求標識
     * @param expireTimeMilliseconds
     *            超期時間，多少毫秒後這把鎖自動釋放
     * @return 返回true表示拿到鎖
     */
    @SuppressWarnings("unchecked")
    public static boolean tryGetDistributedLock(@SuppressWarnings("rawtypes") final RedisTemplate redisTemplate,
            final String lockKey, final String requestId, final int expireTimeMilliseconds) {

        Object result = redisTemplate.execute(new RedisScript<Long>() {
            @Override
            public String getSha1() {
                return SCRIPT_LOCK_SHA1;
            }

            @Override
            public Class<Long> getResultType() {
                return Long.class;
            }

            @Override
            public String getScriptAsString() {
                return SCRIPT_LOCK;
            }

        }, Collections.singletonList(lockKey),// KEYS[1]
                requestId, // ARGV[1]
                String.valueOf(expireTimeMilliseconds) // ARGV[2]
                );

        return SUCCESS.equals(result);
    }

    /**
     * 釋放分散式鎖
     * 
     * @param redisTemplate
     *            Redis客戶端
     * @param lockKey
     *            鎖
     * @param requestId
     *            請求標識
     * @return 返回true表示釋放鎖成功
     */
    @SuppressWarnings("unchecked")
    public static boolean releaseDistributedLock(@SuppressWarnings("rawtypes") RedisTemplate redisTemplate,
            String lockKey, String requestId) {

        Object result = redisTemplate.execute(new RedisScript<Long>() {
            @Override
            public String getSha1() {
                return SCRIPT_UNLOCK_SHA1;
            }

            @Override
            public Class<Long> getResultType() {
                return Long.class;
            }

            @Override
            public String getScriptAsString() {
                return SCRIPT_UNLOCK;
            }

        }, Collections.singletonList(lockKey), requestId);

        return SUCCESS.equals(result);
    }

}

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;

public class Sha1Util {

    private static final String ZERO = "0";
    private static final String ALGORITHM = "SHA1";

    /**
     * sha1加密
     * @param str
     * @return 返回十六進位制的字串形式，全部小寫
     */
    public static String encrypt(String str) {
        if (str == null) {
            return null;
        }
        try {
            MessageDigest messageDigest = MessageDigest.getInstance(ALGORITHM);
            messageDigest.update(str.getBytes());
            return byte2hex(messageDigest.digest());
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 二進位制轉十六進位制字串
     * 
     * @param bytes
     * @return
     */
    public static String byte2hex(byte[] bytes) {
        StringBuilder sign = new StringBuilder();
        for (int i = 0; i < bytes.length; i++) {
            String hex = Integer.toHexString(bytes[i] & 0xFF);
            if (hex.length() == 1) {
                sign.append(ZERO);
            }
            sign.append(hex);
        }
        return sign.toString();
    }

}