Redis分散式事務鎖實現
阿新 • • 發佈:2018-12-25
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文件