Redis事務鎖

在不同程序需要互斥地訪問共享資源時，分散式鎖是一種非常有用的技術手段。本文采用Spring Data Redis實現一下Redis的分散式事務鎖。

Redis為單程序單執行緒模式，採用佇列模式將併發訪問變成序列訪問，且多客戶端對Redis的連線並不存在競爭關係。

SETNX命令（SET if Not eXists）語法：

SETNX key value

若給定的 key 已經存在，則 SETNX 不做任何動作，並返回0。

安全性：保證互斥，在任何時候，只有一個客戶端可以持有鎖
無死鎖：即使當前持有鎖的客戶端崩潰或者從叢集中被分開了，其它客戶端最終總是能夠獲得鎖。
容錯性：只要大部分的 Redis 節點線上，那麼客戶端就能夠獲取和釋放鎖。

使用Spring redisTemplate的實現

使用redisTemplate實現需要配合redis 的eval實現，在Spring Data Redis的官方文件中Redis Scripting一節有相關的說明。

先看一下Spring Redis文件中是如何使用eval的：

@Bean
public RedisScript<Boolean> script() {
  DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<Boolean>();
  redisScript.setScriptSource(new
 
 ResourceScriptSource(new ClassPathResource("META-INF/scripts/checkandset.lua")));
  redisScript.setResultType(Boolean.class);
}

public class Example {
  @Autowired
  RedisScript<Boolean> script;
  public boolean checkAndSet(String expectedValue, String newValue) {
    return redisTemplate.execute(script, Collections.singletonList("key"
 
), expectedValue, newValue);
  }
}

 -- checkandset.lua local
 current = redis.call('GET', KEYS[1])
 if current == ARGV[1]
   then redis.call('SET', KEYS[1], ARGV[2])
   return true
 end
 return false

關於eval函式以及Lua指令碼在此不進行贅述，下面來看一下我們如何使用redisTemplate實現事務鎖。

定義事務鎖的Bean：

public class RedisLock {
    private String key;
    private final UUID uuid;
    private long lockTimeout;

    private long startLockTimeMillis;
    private long getLockTimeMillis;
    private int tryCount;

    public RedisLock(String key, UUID uuid, long lockTimeout, long startLockTimeMillis, long getLockTimeMillis, int tryCount) {
        this.key = key;
        this.uuid = uuid;
        this.lockTimeout = lockTimeout;
        this.startLockTimeMillis = startLockTimeMillis;
        this.getLockTimeMillis = getLockTimeMillis;
        this.tryCount = tryCount;
    }

    public String getKey() {
        return key;
    }

    public void setKey(String key) {
        this.key = key;
    }

    public UUID getUuid() {
        return uuid;
    }

    public long getLockTimeout() {
        return lockTimeout;
    }

    public void setLockTimeout(long lockTimeout) {
        this.lockTimeout = lockTimeout;
    }

    public long getGetLockTimeMillis() {
        return getLockTimeMillis;
    }

    public void setGetLockTimeMillis(long getLockTimeMillis) {
        this.getLockTimeMillis = getLockTimeMillis;
    }

    public long getStartLockTimeMillis() {
        return startLockTimeMillis;
    }

    public void setStartLockTimeMillis(long startLockTimeMillis) {
        this.startLockTimeMillis = startLockTimeMillis;
    }

    public int getTryCount() {
        return tryCount;
    }

    public void setTryCount(int tryCount) {
        this.tryCount = tryCount;
    }
}

建立獲取鎖操作：

// 鎖的過期時間，單位毫秒
private static final long DEFAULT_LOCK_TIME_OUT = 3000;   // 爭搶鎖的超時時間，單位毫秒，0代表永不超時（一直搶到鎖為止）
private static final long DEFAULT_TRY_LOCK_TIME_OUT = 0;  
//拿鎖的EVAL函式
private static final String LUA_SCRIPT_LOCK = "return redis.call('SET', KEYS[1], ARGV[1], 'NX', 'PX', ARGV[2]) ";
//釋放鎖的EVAL函式
private static RedisScript<String> scriptLock = new DefaultRedisScript<String>(LUA_SCRIPT_LOCK, String.class);

獲取鎖的方法：

public static RedisLock lock(int dbIndex, String key, long lockTimeout, long tryLockTimeout) {
    long timestamp = System.currentTimeMillis();
    try {
        //鎖的名稱
        key = key + ".lock";

        UUID uuid = UUID.randomUUID();

        int tryCount = 0;

        //在超時之前，迴圈嘗試拿鎖
        while (tryLockTimeout == 0 || (System.currentTimeMillis() - timestamp) < tryLockTimeout) {

//執行拿鎖的操作，注意這裡，後面的三個引數分別對應了scriptLock字串中的三個變數值，KEYS[1]，ARGV[1]，ARGV[2]，含義為鎖的key，key對應的value，以及key 的存在時間(單位毫秒)

String result = redisTemplate.execute(scriptLock, redisTemplate.getStringSerializer(), redisTemplate.getStringSerializer(), Collections.singletonList(key), uuid.toString(),
                        String.valueOf(lockTimeout));
    tryCount++;
    //返回“OK”代表拿到鎖
    if (result != null && result.equals("OK")) {
        return new RedisLock(key, uuid, lockTimeout, timestamp, System.currentTimeMillis(), tryCount);
    } else {
        try {
            //如果失敗，休息50毫秒繼續重試（自旋鎖）
            Thread.sleep(50);
        } catch (InterruptedException e) {
                        e.printStackTrace();
        }
    }
}
logger.error("Fail to get lock key");
}
return null;
}

上述程式碼就是通過redisTemplate實現的redis 的分散式鎖，如果建立Bean成功則說明拿到鎖，否則拿鎖失敗，核心是採用Redis 的eval函式，使用類似CAS的操作，進行拿鎖，如果拿鎖成功，則返回“OK”，如果失敗，休眠然後繼續嘗試拿鎖，直到超時。

釋放鎖操作：

private static final String LUA_SCRIPT_UNLOCK = 
            "if (redis.call('GET', KEYS[1]) == ARGV[1]) then "
            + "return redis.call('DEL',KEYS[1]) " 
            + "else " + "return 0 " + "end";


private static RedisScript<String> scriptUnlock = 
        new DefaultRedisScript<String>(LUA_SCRIPT_UNLOCK,
            String.class);

public static void unLock(int dbIndex, RedisLock lock) {                                                                 

    redisTemplate.execute(scriptUnlock,
                  redisTemplate.getStringSerializer(),
                  redisTemplate.getStringSerializer(),     
                  Collections.singletonList(lock.getKey()),     
                  lock.getUuid().toString());
}

上述就是使用Redis來實現分散式鎖，其方法是採用Redis String 的 SET進行實現，SET 命令的行為可以通過一系列引數來修改：

• EX second ：設定鍵的過期時間為 second 秒。 SET key value EX second 效果等同於 SETEX key second value 。
• PX millisecond ：設定鍵的過期時間為 millisecond 毫秒。 SET key value PX millisecond 效果等同於 PSETEX key millisecond value 。
• NX ：只在鍵不存在時，才對鍵進行設定操作。 SET key value NX 效果等同於 SETNX key value 。
• XX ：只在鍵已經存在時，才對鍵進行設定操作。

具體更多詳情，請參看Redis文件