不用找了，基於 Redis 的分佈式鎖實戰來了！

Java技術棧

www.javastack.cn

優秀的Java技術公衆號

做者：菜蚜

my.oschina.net/wnjustdoit/blog/1606215

前言：在分佈式環境中，咱們常用鎖來進行併發控制，鎖可分爲樂觀鎖和悲觀鎖，

基於數據庫版本戳的實現是樂觀鎖，基於redis或zookeeper的實現可認爲是悲觀鎖了。樂觀鎖和悲觀鎖最根本的區別在於線程之間是否相互阻塞。

那麼，本文主要來討論基於redis的分佈式鎖算法問題。

從2.6.12版本開始，redis爲SET命令增長了一系列選項(set [key] NX/XX EX/PX [expiration]):

• EX seconds – 設置鍵key的過時時間，單位時秒
• PX milliseconds – 設置鍵key的過時時間，單位時毫秒
• NX – 只有鍵key不存在的時候纔會設置key的值
• XX – 只有鍵key存在的時候纔會設置key的值

原文地址：https://redis.io/commands/set
中文地址：http://redis.cn/commands/set.html

注意: 因爲SET命令加上選項已經能夠徹底取代SETNX, SETEX, PSETEX的功能，因此在未來的版本中，redis可能會不推薦使用而且最終拋棄這幾個命令。

這裏簡單提一下，在舊版本的redis中（指2.6.12版本以前），使用redis實現分佈式鎖通常須要setNX、expire、getSet、del等命令。並且會發現這種實現有不少邏輯判斷的原子操做以及本地時間等並無控制好。

而在舊版本的redis中，redis的超時時間很難控制，用戶迫切須要把setNX和expiration結合爲一體的命令，把他們做爲一個原子操做，這樣新版本的多選項set命令誕生了。然而這並無徹底解決複雜的超時控制帶來的問題。

接下來，咱們的一切討論都基於新版redis。

在這裏，我先提出幾個在實現redis分佈式鎖中須要考慮的關鍵問題

一、死鎖問題；

1.一、爲了防止死鎖，redis至少須要設置一個超時時間；

1.二、由1.1引伸出來，當鎖自動釋放了，可是程序並無執行完畢，這時候其餘線程又獲取到鎖執行一樣的程序，可能會形成併發問題，這個問題咱們須要考慮一下是否歸屬於分佈式鎖帶來問題的範疇。

二、鎖釋放問題，這裏會有兩個問題；

2.一、每一個獲取redis鎖的線程應該釋放本身獲取到的鎖，而不是其餘線程的，因此咱們須要在每一個線程獲取鎖的時候給鎖作上不一樣的標記以示區分；

2.二、由2.1帶來的問題是線程在釋放鎖的時候須要判斷當前鎖是否屬於本身，若是屬於本身才釋放，這裏涉及到邏輯判斷語句，至少是兩個操做在進行，那麼咱們須要考慮這兩個操做要在一個原子內執行，否者在兩個行爲之間可能會有其餘線程插入執行，致使程序紊亂。

三、更可靠的鎖；

單實例的redis（這裏指只有一個master節點）每每是不可靠的，雖然實現起來相對簡單一些，可是會面臨着宕機等不可用的場景，即便在主從複製的時候也顯得並不可靠（由於redis的主從複製每每是異步的）。

關於Martin Kleppmann的Redlock的分析

原文地址：https://redis.io/topics/distlock
中文地址：http://redis.cn/topics/distlock.html

文章分析得出，這種算法只需具有3個特性就能夠實現一個最低保障的分佈式鎖。

• 安全屬性（Safety property）: 獨享（相互排斥）。在任意一個時刻，只有一個客戶端持有鎖。
• 活性A(Liveness property A): 無死鎖。即使持有鎖的客戶端崩潰（crashed)或者網絡被分裂（gets partitioned)，鎖仍然能夠被獲取。
• 活性B(Liveness property B): 容錯。只要大部分Redis節點都活着，客戶端就能夠獲取和釋放鎖.

咱們來分析一下：

第一點安全屬性意味着悲觀鎖（互斥鎖）是咱們作redis分佈式鎖的前提，否者將可能形成併發；

第二點代表爲了不死鎖，咱們須要設置鎖超時時間，保證在必定的時間事後，鎖能夠從新被利用；

第三點是說對於客戶端來講，獲取鎖和手動釋放鎖能夠有更高的可靠性。

更進一步分析，結合上文提到的關鍵問題，這裏能夠引伸出另外的兩個問題：

• 怎麼才能合理判斷程序真正處理的有效時間範圍？（這裏有個時間偏移的問題）
• redis Master節點宕機後恢復（可能尚未持久化到磁盤）、主從節點切換，(N/2)+1這裏的N應該怎麼動態計算更合理？

接下來再看，redis之父antirez對Redlock的評價

原文地址：http://antirez.com/news/101

文中主要提到了網絡延遲和本地時鐘的修改（不論是時間服務器或人爲修改）對這種算法可能形成的影響。

最後，來點實踐吧

I、傳統的單實例redis分佈式鎖實現（關鍵步驟）

獲取鎖（含自動釋放鎖）：

SET resource_name my_random_value NX PX 30000
 手動刪除鎖（Lua腳本）：

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

II、分佈式環境的redis（多master節點）的分佈式鎖實現

爲了保證在儘量短的時間內獲取到(N/2)+1個節點的鎖，能夠並行去獲取各個節點的鎖（固然，並行可能須要消耗更多的資源，由於串行只須要count到足夠數量的鎖就能夠中止獲取了）；

另外，怎麼動態實時統一獲取redis master nodes須要更進一步去思考了。

QA，補充一下說明（如下爲我與朋友溝通的狀況，以說明文中你們可能不夠明白的地方）：

一、在關鍵問題2.1中，刪除就刪除了，會形成什麼問題？

線程A超時，準備刪除鎖；但此時的鎖屬於線程B；線程B還沒執行完，線程A把鎖刪除了，這時線程C獲取到鎖，同時執行程序；因此不能亂刪。

二、在關鍵問題2.2中，只要在key生成時，跟線程相關就不用考慮這個問題了嗎？

不一樣的線程執行程序，線程之間肯雖然有差別呀，而後在redis鎖的value設置有線程信息，好比線程id或線程名稱，是分佈式環境的話加個機器id前綴咯（相似於twitter的snowflake算法！），可是在del命令只會涉及到key，不會再次檢查value，因此仍是須要lua腳本控制if(condition){xxx}的原子性。

三、那要不要考慮鎖的重入性？

不須要重入；try…finally 沒得重入的場景；對於單個線程來講，執行是串行的，獲取鎖以後一定會釋放，由於finally的代碼一定會執行啊（只要進入了try塊，finally一定會執行）。

四、爲何兩個線程都會去刪除鎖？（貌似重複的問題。無論怎樣，仍是耐心解答吧）

每一個線程只能管理本身的鎖，不能管理別人線程的鎖啊。這裏能夠聯想一下ThreadLocal。

五、若是加鎖的線程掛了怎麼辦？只能等待自動超時？

看你怎麼寫程序的了，一種是問題3的回答；另外，那就自動超時咯。這種狀況也適用於網絡over了。

六、時間太長，程序異常就會蛋疼，時間過短，就會出現程序尚未處理完就超時了，這豈不是很尷尬？

是呀，因此須要更好的衡量這個超時時間的設置。

實踐部分主要代碼：

RedisLock工具類：

package com.caiya.cms.web.component;

import com.caiya.cache.CacheException;
import com.caiya.cache.redis.JedisCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * redis實現分佈式鎖
 * 可實現特性:
 * 一、使多線程無序排隊獲取和釋放鎖;
 * 二、丟棄未成功得到鎖的線程處理;
 * 三、只釋放線程自己加持的鎖;
 * 四、避免死鎖
 *
 * @author wangnan
 * @since 1.0
 */
public final class RedisLock {

    private static final Logger logger = LoggerFactory.getLogger(RedisLock.class);

    /**
     * 嘗試加鎖(僅一次)
     *
     * @param lockKey       鎖key
     * @param lockValue     鎖value
     * @param expireSeconds 鎖超時時間(秒)
     * @return 是否加鎖成功
     * @throws CacheException
     */
    public static boolean tryLock(String lockKey, String lockValue, long expireSeconds) throws CacheException {
        JedisCache jedisCache = JedisCacheFactory.getInstance().getJedisCache();
        try {
            String response = jedisCache.set(lockKey, lockValue, "nx", "ex", expireSeconds);
            return Objects.equals(response, "OK");
        } finally {
            jedisCache.close();
        }
    }

    /**
     * 加鎖(指定最大嘗試次數範圍內)
     *
     * @param lockKey       鎖key
     * @param lockValue     鎖value
     * @param expireSeconds 鎖超時時間(秒)
     * @param tryTimes      最大嘗試次數
     * @param sleepMillis   每兩次嘗試之間休眠時間(毫秒)
     * @return 是否加鎖成功
     * @throws CacheException
     */
    public static boolean lock(String lockKey, String lockValue, long expireSeconds, int tryTimes, long sleepMillis) throws CacheException {
        boolean result;
        int count = 0;
        do {
            count++;
            result = tryLock(lockKey, lockValue, expireSeconds);
            try {
                TimeUnit.MILLISECONDS.sleep(sleepMillis);
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        } while (!result && count <= tryTimes);
        return result;
    }

    /**
     * 釋放鎖
     *
     * @param lockKey   鎖key
     * @param lockValue 鎖value
     */
    public static void unlock(String lockKey, String lockValue) {
        JedisCache jedisCache = JedisCacheFactory.getInstance().getJedisCache();
        try {
            String luaScript = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
            Object result = jedisCache.eval(luaScript, 1, lockKey, lockValue);
//            Objects.equals(result, 1L);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            jedisCache.close();
        }
//        return false;
    }


    private RedisLock() {
    }

}

使用工具類的代碼片斷1：

        ...
        String lockKey = Constant.DEFAULT_CACHE_NAME + ":addItemApply:" + applyPriceDTO.getItemId() + "_" + applyPriceDTO.getSupplierId();// 跟業務相關的惟一拼接鍵
        String lockValue = Constant.DEFAULT_CACHE_NAME + ":" + System.getProperty("JvmId") + ":" + Thread.currentThread().getName() + ":" + System.currentTimeMillis();// 生成集羣環境中的惟一值
        boolean locked = RedisLock.tryLock(lockKey, lockValue, 100);// 只嘗試一次，在本次處理過程當中直接拒絕其餘線程的請求
        if (!locked) {
            throw new IllegalAccessException("您的操做太頻繁了,休息一下再來吧~");
        }
        try {
            // 開始處理核心業務邏輯
            Item item = itemService.queryItemByItemId(applyPriceDTO.getItemId());
            ...
            ...
        } finally {
            RedisLock.unlock(lockKey, lockValue);// 在finally塊中釋放鎖
        }

使用工具類的代碼片斷2：

        ...
        String lockKey = Constant.DEFAULT_CACHE_NAME + ":addItemApply:" + applyPriceDTO.getItemId() + "_" + applyPriceDTO.getSupplierId();
        String lockValue = Constant.DEFAULT_CACHE_NAME + ":機器編號:" + Thread.currentThread().getName() + ":" + System.currentTimeMillis();
        boolean locked = RedisLock.lock(lockKey, lockValue, 100, 20, 100);// 非公平鎖，無序競爭（這裏須要合理根據業務處理狀況設置最大嘗試次數和每次休眠時間）
        if (!locked) {
            throw new IllegalAccessException("系統太忙,本次操做失敗");// 通常來講，不會走到這一步；若是真的有這種狀況，而且在合理設置鎖嘗試次數和等待響應時間以後仍然處理不過來，可能須要考慮優化程序響應時間或者用消息隊列排隊執行了
        }

        try {
            // 開始處理核心業務邏輯
            Item item = itemService.queryItemByItemId(applyPriceDTO.getItemId());
            ...
            ...
        } finally {
            RedisLock.unlock(lockKey, lockValue);
        }
        ...

附加：

基於redis的分佈式鎖實現客戶端Redisson：

https://github.com/redisson/redisson/wiki/8.-Distributed-locks-and-synchronizers

基於zookeeper的分佈式鎖實現：

http://curator.apache.org/curator-recipes/shared-reentrant-lock.html

- END -

關注Java技術棧微信公衆號，在後臺回覆關鍵字：Java，能夠獲取一份棧長整理的 Java 最新技術乾貨。

最近乾貨分享

點擊「閱讀原文」加入棧長的戰隊～