使用Redis單執行個體實作不可重入的分布式鎖

一、前言

在同一個jvm程序中時，可以使用JUC提供的一些鎖來解決多個線程競争同一個共享資源時候的線程安全問題，但是當多個不同jvm程序中的線程共同競争同一個共享資源時候，juc包的鎖就無能無力了，這時候就需要分布式鎖了。常見的有使用zk的最小版本，redis的set函數，資料庫鎖來實作，本節我們談談Redis單執行個體情況下使用set函數來實作分布式鎖。

二、使用Redis單執行個體實作分布式鎖

首先我們來具體看代碼：

package com.jiaduo.DistributedLock;

import java.util.Collections;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

public class DistributedLock {

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

    private static void validParam(JedisPool jedisPool, String lockKey, String requestId, int expireTime) {
        if (null == jedisPool) {
            throw new IllegalArgumentException("jedisPool obj is null");
        }

        if (null == lockKey || "".equals(lockKey)) {
            throw new IllegalArgumentException("lock key  is blank");
        }

        if (null == requestId || "".equals(requestId)) {
            throw new IllegalArgumentException("requestId is blank");
        }

        if (expireTime < 0) {
            throw new IllegalArgumentException("expireTime is not allowed less zero");
        }
    }

    /**
     * 
     * @param jedis
     * @param lockKey
     * @param requestId
     * @param expireTime
     * @return
     */
    public static boolean tryLock(JedisPool jedisPool, String lockKey, String requestId, int expireTime) {

        validParam(jedisPool, lockKey, requestId, expireTime);

        Jedis jedis = null;
        try {

            jedis = jedisPool.getResource();
            String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);

            if (LOCK_SUCCESS.equals(result)) {
                return true;
            }
        } catch (Exception e) {
            throw e;
        } finally {
            if (null != jedis) {
                jedis.close();
            }
        }

        return false;
    }

    /**
     * 
     * @param jedis
     * @param lockKey
     * @param requestId
     * @param expireTime
     */
    public static void lock(JedisPool jedisPool, String lockKey, String requestId, int expireTime) {

        validParam(jedisPool, lockKey, requestId, expireTime);

        while (true) {
            if (tryLock(jedisPool, lockKey, requestId, expireTime)) {
                return;
            }
        }
    }

    /**
     * 
     * @param jedis
     * @param lockKey
     * @param requestId
     * @return
     */
    public static boolean unLock(JedisPool jedisPool, String lockKey, String requestId) {

        validParam(jedisPool, lockKey, requestId, 1);

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

        Jedis jedis = null;
        try {

            jedis = jedisPool.getResource();
            Object result = jedis.eval(script, Collections.singletonList(lockKey),
                    Collections.singletonList(requestId));

            if (RELEASE_SUCCESS.equals(result)) {
                return true;
            }

        } catch (Exception e) {
            throw e;
        } finally {
            if (null != jedis) {
                jedis.close();
            }
        }

        return false;

    }

}           

首先Redis的 public String set(final String key, final String value, final String nxxx, final String expx, final int time)方法參數說明：

• 其中前面兩個是key,value值；
• nxxx為模式，這裡我們設定為NX，意思是說如果key不存在則插入該key對應的value并傳回OK，否者什麼都不做傳回null；
• 參數expx這裡我們設定為PX，意思是設定key的過期時間為time 毫秒

通過tryLock方法嘗試擷取鎖，内部是具體調用Redis的set方法，多個線程同時調用tryLock時候會同時調用set方法，但是set方法本身是保證原子性的，對應同一個key來說，多個線程調用set方法時候隻有一個線程傳回OK，其它線程因為key已經存在會傳回null，是以傳回OK的線程就相當與擷取到了鎖，其它傳回null的線程則相當于擷取鎖失敗。

另外這裡我們要保證value（requestId）值唯一是為了保證隻有擷取到鎖的線程才能釋放鎖,這個下面釋放鎖時候會講解。

通過lock 方法讓使用tryLock擷取鎖失敗的線程本地自旋轉重試擷取鎖，這類似JUC裡面的CAS。

Redis有一個叫做eval的函數，支援Lua腳本執行，并且能夠保證腳本執行的原子性，也就是在執行腳本期間，其它執行redis指令的線程都會被阻塞。這裡解鎖時候使用下面腳本：

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

其中keys[1]為unLock方法傳遞的key，argv[1]為unLock方法傳遞的requestId;腳本redis.call('get', KEYS[1])的作用是擷取key對應的value值，這裡會傳回通過Lock方法傳遞的requetId, 然後看目前傳遞的RequestId是否等于key對應的值，等于則說明目前要釋放鎖的線程就是擷取鎖的線程，則繼續執行redis.call('del', KEYS[1])腳本，删除key對應的值，這相當于釋放了鎖。因為删除後，其他線程中的一個線程在調用set方法發現該key已經不存在了，則會插入對應的value，然後傳回OK，這相當于這個線程擷取到了鎖。

另外可知該鎖是不可重入鎖。

三、總結

本文使用redis單執行個體結合redis的set方法和eval函數實作了一個簡單的分布式鎖，但是這個實作還是明顯有問題的。雖然使用set方法設定了逾時時間，以避免線程擷取到鎖後redis挂了後鎖沒有被釋放的情況，但是逾時時間設定為多少合适那？如果設定太小，可能會存線上程擷取鎖後執行業務邏輯時間大于鎖逾時時間，那麼就會存在邏輯還沒執行完，鎖已經因為逾時自動釋放了，而其他線程可能擷取到鎖，那麼之前擷取鎖的線程的業務邏輯的執行就沒有保證原子性。

另外還有一個問題是Lock方法裡面是自旋調用tryLock進行重試，這就會導緻像JUC中的AtomicLong一樣，在高并發下多個線程競争同一個資源時候造成大量線程占用cpu進行重試操作。這時候其實可以随機生成一個等待時間，等時間到後在進行重試，以減少潛在的同時對一個資源進行競争的并發量。

歡迎大家讨論，如果是你，你會怎麼做那？