一、簡介
一般來說,對資料進行加鎖時,程式先通過acquire擷取鎖來對資料進行排他通路,然後對資料進行一些列的操作,最後需要釋放鎖。Redis 本身用 watch指令進行了加鎖,這個鎖是樂觀鎖。使用 watch指令對于頻繁通路的鍵會引起性能的問題。
二、redis指令介紹
SETNX指令(SET if Not eXists)
當且僅當 key 不存在,将 key 的值設為 value ,并傳回1;若給定的 key 已經存在,則 SETNX 不做任何動作,并傳回0。
SETEX指令
設定逾時時間
GET指令
傳回 key 所關聯的字元串值,如果 key 不存在那麼傳回特殊值 nil 。
DEL指令
删除給定的一個或多個 key ,不存在的 key 會被忽略。
三、實作思路
由于redis的setnx指令天生就适合用來實作鎖的功能,這個指令隻有在鍵不存在的情況下為鍵設定值。擷取鎖之後,其他程式再設定值就會失敗,即擷取不到鎖。擷取鎖失敗。隻需不斷的嘗試擷取鎖,直到成功擷取鎖,或者到設定的逾時時間為止。
另外為了防治死鎖,即某個程式擷取鎖之後,程式出錯,沒有釋放,其他程式無法擷取鎖,進而導緻整個分布式系統無法擷取鎖而導緻一系列問題,甚至導緻系統無法正常運作。這時需要給鎖設定一個逾時時間,即setex指令,鎖逾時後,進而其它程式就可以擷取鎖了。
四、編碼實作
本文采用springboot結合redis 取實作的,是以你需要裝一個redis。
首先引入建立springboot工程,引入redis 。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- 開啟web-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- redis-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
2.建立一個鎖類
/**
* 全局鎖,包括鎖的名稱
* Created by fangzhipeng on 2017/4/1.
*/
public class Lock {
private String name;
private String value;
public Lock(String name, String value) {
this.name = name;
this.value = value;
}
public String getName() {
return name;
}
public String getValue() {
return value;
}
}
3.建立分布式鎖的具體方法,思路已經說清楚了,代碼注釋也寫好了,就不講解了。
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
/**
* Created by fangzhipeng on 2017/4/1.
*/
@Component
public class DistributedLockHandler {
private static final Logger logger = LoggerFactory.getLogger(DistributedLockHandler.class);
private final static long LOCK_EXPIRE = 30 * 1000L;//單個業務持有鎖的時間30s,防止死鎖
private final static long LOCK_TRY_INTERVAL = 30L;//預設30ms嘗試一次
private final static long LOCK_TRY_TIMEOUT = 20 * 1000L;//預設嘗試20s
@Autowired
private StringRedisTemplate template;
/**
* 嘗試擷取全局鎖
*
* @param lock 鎖的名稱
* @return true 擷取成功,false擷取失敗
*/
public boolean tryLock(Lock lock) {
return getLock(lock, LOCK_TRY_TIMEOUT, LOCK_TRY_INTERVAL, LOCK_EXPIRE);
}
/**
* 嘗試擷取全局鎖
*
* @param lock 鎖的名稱
* @param timeout 擷取逾時時間 機關ms
* @return true 擷取成功,false擷取失敗
*/
public boolean tryLock(Lock lock, long timeout) {
return getLock(lock, timeout, LOCK_TRY_INTERVAL, LOCK_EXPIRE);
}
/**
* 嘗試擷取全局鎖
*
* @param lock 鎖的名稱
* @param timeout 擷取鎖的逾時時間
* @param tryInterval 多少毫秒嘗試擷取一次
* @return true 擷取成功,false擷取失敗
*/
public boolean tryLock(Lock lock, long timeout, long tryInterval) {
return getLock(lock, timeout, tryInterval, LOCK_EXPIRE);
}
/**
* 嘗試擷取全局鎖
*
* @param lock 鎖的名稱
* @param timeout 擷取鎖的逾時時間
* @param tryInterval 多少毫秒嘗試擷取一次
* @param lockExpireTime 鎖的過期
* @return true 擷取成功,false擷取失敗
*/
public boolean tryLock(Lock lock, long timeout, long tryInterval, long lockExpireTime) {
return getLock(lock, timeout, tryInterval, lockExpireTime);
}
/**
* 操作redis擷取全局鎖
*
* @param lock 鎖的名稱
* @param timeout 擷取的逾時時間
* @param tryInterval 多少ms嘗試一次
* @param lockExpireTime 擷取成功後鎖的過期時間
* @return true 擷取成功,false擷取失敗
*/
public boolean getLock(Lock lock, long timeout, long tryInterval, long lockExpireTime) {
try {
if (StringUtils.isEmpty(lock.getName()) || StringUtils.isEmpty(lock.getValue())) {
return false;
}
long startTime = System.currentTimeMillis();
do{
if (!template.hasKey(lock.getName())) {
ValueOperations<String, String> ops = template.opsForValue();
ops.set(lock.getName(), lock.getValue(), lockExpireTime, TimeUnit.MILLISECONDS);
return true;
} else {//存在鎖
logger.debug("lock is exist!!!");
}
if (System.currentTimeMillis() - startTime > timeout) {//嘗試超過了設定值之後直接跳出循環
return false;
}
Thread.sleep(tryInterval);
}
while (template.hasKey(lock.getName())) ;
} catch (InterruptedException e) {
logger.error(e.getMessage());
return false;
}
return false;
}
/**
* 釋放鎖
*/
public void releaseLock(Lock lock) {
if (!StringUtils.isEmpty(lock.getName())) {
template.delete(lock.getName());
}
}
}
4.用法:
@Autowired
DistributedLockHandler distributedLockHandler;
Lock lock=new Lock("lockk","sssssssss);
if(distributedLockHandler.tryLock(lock){
doSomething();
distributedLockHandler.releaseLock();
}
五、注意點
在使用全局鎖時為了防止死鎖采用 setex指令,這種指令需要根據具體的業務具體設定鎖的逾時時間。另外一個就是鎖的粒度性。比如在redis實戰中有個案列,為了實作買賣市場交易的功能,把整個交易市場都鎖住了,導緻了性能不足的情況,改進方案隻對買賣的商品進行加鎖而不是整個市場。