【分散式緩存系列】Redis實現分散式鎖的正確姿勢

一、前言在我們日常工作中，除了Spring和Mybatis外，用到最多無外乎分散式緩存框架——Redis。但是很多工作很多年的朋友對Redis還處於一個最基礎的使用和認識。所以我就像把自己對分散式緩存的一些理解和應用整理一個系列，希望可以幫助到大家加深對Redis的理解。本系列的文章思路先從Red ...

一、前言

　　在我們日常工作中，除了Spring和Mybatis外，用到最多無外乎分散式緩存框架——Redis。但是很多工作很多年的朋友對Redis還處於一個最基礎的使用和認識。所以我就像把自己對分散式緩存的一些理解和應用整理一個系列，希望可以幫助到大家加深對Redis的理解。本系列的文章思路先從Redis的應用開始。再解析Redis的內部實現原理。最後以經常會問到Redist相關的面試題為結尾。

二、分散式鎖的實現要點

為了實現分散式鎖，需要確保鎖同時滿足以下四個條件：

互斥性。在任意時刻，只有一個客戶端能持有鎖
不會發送死鎖。即使一個客戶端持有鎖的期間崩潰而沒有主動釋放鎖，也需要保證後續其他客戶端能夠加鎖成功
加鎖和解鎖必須是同一個客戶端，客戶端自己不能把別人加的鎖給釋放了。
容錯性。只要大部分的Redis節點正常運行，客戶端就可以進行加鎖和解鎖操作。

三、Redis實現分散式鎖的錯誤姿勢

3.1 加鎖錯誤姿勢

　　在講解使用Redis實現分散式鎖的正確姿勢之前，我們有必要來看下錯誤實現方式。

　　首先，為了保證互斥性和不會發送死鎖2個條件，所以我們在加鎖操作的時候，需要使用SETNX指令來保證互斥性——只有一個客戶端能夠持有鎖。為了保證不會發送死鎖，需要給鎖加一個過期時間，這樣就可以保證即使持有鎖的客戶端期間崩潰了也不會一直不釋放鎖。

　　為了保證這2個條件，有些人錯誤的實現會用如下代碼來實現加鎖操作：

/**
     * 實現加鎖的錯誤姿勢
     * @param jedis
     * @param lockKey
     * @param requestId
     * @param expireTime
     */
    public static void wrongGetLock1(Jedis jedis, String lockKey, String requestId, int expireTime) {
        Long result = jedis.setnx(lockKey, requestId);
        if (result == 1) {
            // 若在這裡程式突然崩潰，則無法設置過期時間，將發生死鎖
            jedis.expire(lockKey, expireTime);
        }
    }

　　可能一些初學者還沒看出以上實現加鎖操作的錯誤原因。這樣我們解釋下。setnx 和expire是兩條Redis指令，不具備原子性，如果程式在執行完setnx之後突然崩潰，導致沒有設置鎖的過期時間，從而就導致死鎖了。因為這個客戶端持有的所有不會被其他客戶端釋放，持有鎖的客戶端又崩潰了，也不會主動釋放。從而該鎖永遠不會釋放，導致其他客戶端也獲得不能鎖。從而其他客戶端一直阻塞。所以針對該代碼正確姿勢應該保證setnx和expire原子性。

　　實現加鎖操作的錯誤姿勢2。具體實現如下代碼所示

/**
     * 實現加鎖的錯誤姿勢2
     * @param jedis
     * @param lockKey
     * @param expireTime
     * @return
     */
    public static boolean wrongGetLock2(Jedis jedis, String lockKey, int expireTime) {
        long expires = System.currentTimeMillis() + expireTime;
        String expiresStr = String.valueOf(expires);
        // 如果當前鎖不存在，返回加鎖成功
        if (jedis.setnx(lockKey, expiresStr) == 1) {
            return true;
        }

        // 如果鎖存在，獲取鎖的過期時間
        String currentValueStr = jedis.get(lockKey);
        if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
            // 鎖已過期，獲取上一個鎖的過期時間，並設置現在鎖的過期時間
            String oldValueStr = jedis.getSet(lockKey, expiresStr);
            if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
                // 考慮多線程併發的情況，只有一個線程的設置值和當前值相同，它才有權利加鎖
                return true;
            }
        }
        // 其他情況，一律返回加鎖失敗
        return false;
    }

　　這個加鎖操作咋一看沒有毛病對吧。那以上這段代碼的問題毛病出在哪裡呢？

　　1. 由於客戶端自己生成過期時間，所以需要強制要求分散式環境下所有客戶端的時間必須同步。

　　2. 當鎖過期的時候，如果多個客戶端同時執行jedis.getSet()方法，雖然最終只有一個客戶端加鎖，但是這個客戶端的鎖的過期時間可能被其他客戶端覆蓋。不具備加鎖和解鎖必須是同一個客戶端的特性。解決上面這段代碼的方式就是為每個客戶端加鎖添加一個唯一標示，已確保加鎖和解鎖操作是來自同一個客戶端。

3.2 解鎖錯誤姿勢

　　分散式鎖的實現無法就2個方法，一個加鎖，一個就是解鎖。下麵我們來看下解鎖的錯誤姿勢。

　　錯誤姿勢1.

/**
     * 解鎖錯誤姿勢1
     * @param jedis
     * @param lockKey
     */
    public static void wrongReleaseLock1(Jedis jedis, String lockKey) {
        jedis.del(lockKey);
    }

　　上面實現是最簡單直接的解鎖方式，這種不先判斷擁有者而直接解鎖的方式，會導致任何客戶端都可以隨時解鎖。即使這把鎖不是它上鎖的。

　　錯誤姿勢2：

/**
     * 解鎖錯誤姿勢2
     * @param jedis
     * @param lockKey
     * @param requestId
     */
    public static void wrongReleaseLock2(Jedis jedis, String lockKey, String requestId) {

        // 判斷加鎖與解鎖是不是同一個客戶端
        if (requestId.equals(jedis.get(lockKey))) {
            // 若在此時，這把鎖突然不是這個客戶端的，則會誤解鎖
            jedis.del(lockKey);
        }

　　既然錯誤姿勢1中沒有判斷鎖的擁有者，那姿勢2中判斷了擁有者，那錯誤原因又在哪裡呢？答案又是原子性上面。因為判斷和刪除不是一個原子性操作。在併發的時候很可能發生解除了別的客戶端加的鎖。具體場景有：客戶端A加鎖，一段時間之後客戶端A進行解鎖操作時，在執行jedis.del()之前，鎖突然過期了，此時客戶端B嘗試加鎖成功，然後客戶端A再執行del方法，則客戶端A將客戶端B的鎖給解除了。從而不也不滿足加鎖和解鎖必須是同一個客戶端特性。解決思路就是需要保證GET和DEL操作在一個事務中進行，保證其原子性。

四、Redis實現分散式鎖的正確姿勢

　　剛剛介紹完了錯誤的姿勢後，從上面錯誤姿勢中，我們可以知道，要使用Redis實現分散式鎖。加鎖操作的正確姿勢為：

使用setnx命令保證互斥性
需要設置鎖的過期時間，避免死鎖
setnx和設置過期時間需要保持原子性，避免在設置setnx成功之後在設置過期時間客戶端崩潰導致死鎖
加鎖的Value 值為一個唯一標示。可以採用UUID作為唯一標示。加鎖成功後需要把唯一標示返回給客戶端來用來客戶端進行解鎖操作

　　解鎖的正確姿勢為：

　　1. 需要拿加鎖成功的唯一標示要進行解鎖，從而保證加鎖和解鎖的是同一個客戶端

　　2. 解鎖操作需要比較唯一標示是否相等，相等再執行刪除操作。這2個操作可以採用Lua腳本方式使2個命令的原子性。

　　Redis分散式鎖實現的正確姿勢的實現代碼：

public interface DistributedLock {
    /**
     * 獲取鎖
     * @author zhi.li
     * @return 鎖標識
     */
    String acquire();

    /**
     * 釋放鎖
     * @author zhi.li
     * @param indentifier
     * @return
     */
    boolean release(String indentifier);
}

/**
 * @author zhi.li
 * @Description
 * @created 2019/1/1 20:32
 */
@Slf4j
public class RedisDistributedLock implements DistributedLock{

    private static final String LOCK_SUCCESS = "OK";
    private static final Long RELEASE_SUCCESS = 1L;
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";

    /**
     * redis 客戶端
     */
    private Jedis jedis;

    /**
     * 分散式鎖的鍵值
     */
    private String lockKey;

    /**
     * 鎖的超時時間 10s
     */
    int expireTime = 10 * 1000;

    /**
     * 鎖等待，防止線程饑餓
     */
    int acquireTimeout  = 1 * 1000;

    /**
     * 獲取指定鍵值的鎖
     * @param jedis jedis Redis客戶端
     * @param lockKey 鎖的鍵值
     */
    public RedisDistributedLock(Jedis jedis, String lockKey) {
        this.jedis = jedis;
        this.lockKey = lockKey;
    }

    /**
     * 獲取指定鍵值的鎖,同時設置獲取鎖超時時間
     * @param jedis jedis Redis客戶端
     * @param lockKey 鎖的鍵值
     * @param acquireTimeout 獲取鎖超時時間
     */
    public RedisDistributedLock(Jedis jedis,String lockKey, int acquireTimeout) {
        this.jedis = jedis;
        this.lockKey = lockKey;
        this.acquireTimeout = acquireTimeout;
    }

    /**
     * 獲取指定鍵值的鎖,同時設置獲取鎖超時時間和鎖過期時間
     * @param jedis jedis Redis客戶端
     * @param lockKey 鎖的鍵值
     * @param acquireTimeout 獲取鎖超時時間
     * @param expireTime 鎖失效時間
     */
    public RedisDistributedLock(Jedis jedis, String lockKey, int acquireTimeout, int expireTime) {
        this.jedis = jedis;
        this.lockKey = lockKey;
        this.acquireTimeout = acquireTimeout;
        this.expireTime = expireTime;
    }

    @Override
    public String acquire() {
        try {
            // 獲取鎖的超時時間，超過這個時間則放棄獲取鎖
            long end = System.currentTimeMillis() + acquireTimeout;
            // 隨機生成一個value
            String requireToken = UUID.randomUUID().toString();
            while (System.currentTimeMillis() < end) {
                String result = jedis.set(lockKey, requireToken, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
                if (LOCK_SUCCESS.equals(result)) {
                    return requireToken;
                }
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        } catch (Exception e) {
            log.error("acquire lock due to error", e);
        }

        return null;
    }

    @Override
    public boolean release(String identify) {
　　　 if(identify == null){
            return false;
        }

        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object result = new Object();
        try {
            result = jedis.eval(script, Collections.singletonList(lockKey),
                Collections.singletonList(identify));
        if (RELEASE_SUCCESS.equals(result)) {
            log.info("release lock success, requestToken:{}", identify);
            return true;
        }}catch (Exception e){
            log.error("release lock due to error",e);
        }finally {
            if(jedis != null){
                jedis.close();
            }
        }

        log.info("release lock failed, requestToken:{}, result:{}", identify, result);
        return false;
    }
}
　　下麵就以秒殺庫存數量為場景，測試下上面實現的分散式鎖的效果。具體測試代碼如下：

public class RedisDistributedLockTest {
    static int n = 500;
    public static void secskill() {
        System.out.println(--n);
    }

    public static void main(String[] args) {
        Runnable runnable = () -> {
            RedisDistributedLock lock = null;
            String unLockIdentify = null;
            try {
                Jedis conn = new Jedis("127.0.0.1",6379);
                lock = new RedisDistributedLock(conn, "test1");
                unLockIdentify = lock.acquire();
                System.out.println(Thread.currentThread().getName() + "正在運行");
                secskill();
            } finally {
                if (lock != null) {
                    lock.release(unLockIdentify);
                }
            }
        };

        for (int i = 0; i < 10; i++) {
            Thread t = new Thread(runnable);
            t.start();
        }
    }
}

　　運行效果如下圖所示。從圖中可以看出，同一個資源在同一個時刻只能被一個線程獲取，從而保證了庫存數量N的遞減是順序的。

五、總結

　　這樣是不是已經完美使用Redis實現了分散式鎖呢？答案是並沒有結束。上面的實現代碼只是針對單機的Redis沒問題。但是現實生產中大部分都是集群的或者是主備的。但上面的實現姿勢在集群或者主備情況下會有相應的問題。這裡先買一個關子，在後面一篇文章將詳細分析集群或者主備環境下Redis分散式鎖的實現方式。

　　本文所有源碼下載地址：https://github.com/learninghard-lizhi/common-util

　　補充：為了暫時滿足大家好奇心，這裡先拋出兩篇文章已供大家瞭解在集群環境下上面實現方式的問題。

《基於Redis的分散式鎖到底安全嗎（上）？》

《基於Redis的分散式鎖到底安全嗎（下）？》